You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "xi-db (via GitHub)" <gi...@apache.org> on 2025/11/05 10:20:59 UTC
[PR] [SPARK-54194] Spark Connect Proto Plan Compression [spark]
xi-db opened a new pull request, #52894:
URL: https://github.com/apache/spark/pull/52894
### What changes were proposed in this pull request?
Currently, Spark Connect enforce gRPC message limits on both the client and the server. These limits are largely meant to protect the server from potential OOMs by rejecting abnormally large messages. However, there are several cases where genuine messages exceed the limit and cause execution failures.
To improve Spark Connect stability, this PR implements compressing unresolved proto plans to mitigate the issue of oversized messages from the client to the server. The compression applies to ExecutePlan and AnalyzePlan - the only two methods that might hit the message limit. The other issue of message limit from the server to the client is a different issue, and it’s out of the scope (that one is already fixed in https://github.com/apache/spark/pull/52271).
(Scala client changes are being implemented in a follow-up PR.)
To reproduce the existing issue we are solving here, run this code on Spark Connect:
```
import random
import string
def random_letters(length: int) -> str:
return ''.join(random.choices(string.ascii_letters, k=length))
num_unique_small_relations = 5
size_per_small_relation = 512 * 1024
small_dfs = [spark.createDataFrame([(random_letters(size_per_small_relation),)],) for _ in range(num_unique_small_relations)]
result_df = small_dfs[0]
for _ in range(512):
result_df = result_df.unionByName(small_dfs[random.randint(0, len(small_dfs) - 1)])
result_df.collect()
```
It fails with `StatusCode.RESOURCE_EXHAUSTED` error with message`Sent message larger than max (269178955 vs. 134217728)`, because the client was trying to send a too large message to the server.
Note: repeated small local relations is only one way causing a large plan, the size of the plan can also be contributed by repeated subtrees of plan transformations, serialized UDFs, captured external variables by UDFs, etc.
With the improvement introduced by the PR, the above code runs successfully and prints the expected result.
### Why are the changes needed?
It improves Spark Connect stability when executing and analyzing large plans.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
New tests on both the server side and the client side.
### Was this patch authored or co-authored using generative AI tooling?
No.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54194][CONNECT] Spark Connect Proto Plan Compression [spark]
Posted by "xi-db (via GitHub)" <gi...@apache.org>.
xi-db commented on code in PR #52894:
URL: https://github.com/apache/spark/pull/52894#discussion_r2510344089
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -901,6 +901,24 @@
},
"sqlState" : "56K00"
},
+ "CONNECT_INVALID_PLAN" : {
+ "message" : [
+ "The Spark Connect plan is invalid."
+ ],
+ "subClass" : {
+ "CANNOT_PARSE" : {
+ "message" : [
+ "Cannot decompress or parse the input plan (<errorMsg>)"
+ ]
+ },
+ "PLAN_SIZE_LARGER_THAN_MAX" : {
+ "message" : [
+ "The plan size is larger than max (<planSize> vs. <maxPlanSize>)"
+ ]
Review Comment:
Yes, I've updated the message to include the possible resolution.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54194][CONNECT] Spark Connect Proto Plan Compression [spark]
Posted by "xi-db (via GitHub)" <gi...@apache.org>.
xi-db commented on PR #52894:
URL: https://github.com/apache/spark/pull/52894#issuecomment-3527900852
Thank you @dongjoon-hyun !
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54194][CONNECT] Spark Connect Proto Plan Compression [spark]
Posted by "asf-gitbox-commits (via GitHub)" <gi...@apache.org>.
asf-gitbox-commits closed pull request #52894: [SPARK-54194][CONNECT] Spark Connect Proto Plan Compression
URL: https://github.com/apache/spark/pull/52894
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54194][CONNECT] Spark Connect Proto Plan Compression [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #52894:
URL: https://github.com/apache/spark/pull/52894#issuecomment-3524326866
It seems that we missed `MacOS` CI. I made a PR to recover the CI, @xi-db , @hvanhovell .
- https://github.com/apache/spark/pull/53025
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54194][CONNECT] Spark Connect Proto Plan Compression [spark]
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on PR #52894:
URL: https://github.com/apache/spark/pull/52894#issuecomment-3519733251
@xi-db please also update the doc https://apache.github.io/spark/api/python/getting_started/install.html#spark-connect
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54194][CONNECT] Spark Connect Proto Plan Compression [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #52894:
URL: https://github.com/apache/spark/pull/52894#discussion_r2528191887
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/config/Connect.scala:
##########
@@ -418,4 +419,34 @@ object Connect {
.bytesConf(ByteUnit.BYTE)
// 90% of the max message size by default to allow for some overhead.
.createWithDefault((ConnectCommon.CONNECT_GRPC_MAX_MESSAGE_SIZE * 0.9).toInt)
+
+ private[spark] val CONNECT_MAX_PLAN_SIZE =
+ buildStaticConf("spark.connect.maxPlanSize")
+ .doc(
+ "The maximum size of a (decompressed) proto plan that can be executed in Spark " +
+ "Connect. If the size of the plan exceeds this limit, an error will be thrown. " +
+ "The size is in bytes.")
+ .version("4.1.0")
+ .internal()
+ .bytesConf(ByteUnit.BYTE)
+ .createWithDefault(512 * 1024 * 1024) // 512 MB
+
+ val CONNECT_SESSION_PLAN_COMPRESSION_THRESHOLD =
+ buildConf("spark.connect.session.planCompression.threshold")
+ .doc("The threshold in bytes for the size of proto plan to be compressed. " +
+ "If the size of proto plan is smaller than this threshold, it will not be compressed.")
+ .version("4.1.0")
+ .internal()
+ .intConf
+ .createWithDefault(10 * 1024 * 1024) // 10 MB
+
+ val CONNECT_PLAN_COMPRESSION_DEFAULT_ALGORITHM =
+ buildConf("spark.connect.session.planCompression.defaultAlgorithm")
+ .doc("The default algorithm of proto plan compression.")
+ .version("4.1.0")
+ .internal()
+ .stringConf
+ .transform(_.toUpperCase(Locale.ROOT))
+ .checkValues(ConnectPlanCompressionAlgorithm.values.map(_.toString))
Review Comment:
Is there a way to unset this feature? It seems that we enforce to use ZSTD because it's the only one, @xi-db and @hvanhovell ?
```scala
object ConnectPlanCompressionAlgorithm extends Enumeration {
val ZSTD = Value
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54194][CONNECT] Spark Connect Proto Plan Compression [spark]
Posted by "xi-db (via GitHub)" <gi...@apache.org>.
xi-db commented on code in PR #52894:
URL: https://github.com/apache/spark/pull/52894#discussion_r2533330082
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/config/Connect.scala:
##########
@@ -418,4 +419,34 @@ object Connect {
.bytesConf(ByteUnit.BYTE)
// 90% of the max message size by default to allow for some overhead.
.createWithDefault((ConnectCommon.CONNECT_GRPC_MAX_MESSAGE_SIZE * 0.9).toInt)
+
+ private[spark] val CONNECT_MAX_PLAN_SIZE =
+ buildStaticConf("spark.connect.maxPlanSize")
+ .doc(
+ "The maximum size of a (decompressed) proto plan that can be executed in Spark " +
+ "Connect. If the size of the plan exceeds this limit, an error will be thrown. " +
+ "The size is in bytes.")
+ .version("4.1.0")
+ .internal()
+ .bytesConf(ByteUnit.BYTE)
+ .createWithDefault(512 * 1024 * 1024) // 512 MB
+
+ val CONNECT_SESSION_PLAN_COMPRESSION_THRESHOLD =
+ buildConf("spark.connect.session.planCompression.threshold")
+ .doc("The threshold in bytes for the size of proto plan to be compressed. " +
+ "If the size of proto plan is smaller than this threshold, it will not be compressed.")
+ .version("4.1.0")
+ .internal()
+ .intConf
+ .createWithDefault(10 * 1024 * 1024) // 10 MB
+
+ val CONNECT_PLAN_COMPRESSION_DEFAULT_ALGORITHM =
+ buildConf("spark.connect.session.planCompression.defaultAlgorithm")
+ .doc("The default algorithm of proto plan compression.")
+ .version("4.1.0")
+ .internal()
+ .stringConf
+ .transform(_.toUpperCase(Locale.ROOT))
+ .checkValues(ConnectPlanCompressionAlgorithm.values.map(_.toString))
Review Comment:
To disable the feature, user can set the spark.connect.session.planCompression.threshold to -1 ([link](https://github.com/apache/spark/blob/722bcc0f0d15245a39fae62c0c1c764e4b6a02f8/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/config/Connect.scala#L434-L438)). Still, it is better to have a enum "None". Thanks for your PR!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54194][CONNECT] Spark Connect Proto Plan Compression [spark]
Posted by "xi-db (via GitHub)" <gi...@apache.org>.
xi-db commented on PR #52894:
URL: https://github.com/apache/spark/pull/52894#issuecomment-3516527383
Thanks @dongjoon-hyun , there is no specific reason why 0.23.0 is chosen. It's simply I didn't realize the new stable version was released a few months ago.
I've updated to use the latest stable version 0.25.0. cc @hvanhovell
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54194][CONNECT] Spark Connect Proto Plan Compression [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on PR #52894:
URL: https://github.com/apache/spark/pull/52894#issuecomment-3516861584
Merging to master/4.1. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54194][CONNECT] Spark Connect Proto Plan Compression [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #52894:
URL: https://github.com/apache/spark/pull/52894#issuecomment-3534037502
I made a PR to support `NONE`.
- https://github.com/apache/spark/pull/53068
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54194][CONNECT] Spark Connect Proto Plan Compression [spark]
Posted by "vicennial (via GitHub)" <gi...@apache.org>.
vicennial commented on code in PR #52894:
URL: https://github.com/apache/spark/pull/52894#discussion_r2494487523
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -901,6 +901,24 @@
},
"sqlState" : "56K00"
},
+ "CONNECT_INVALID_PLAN" : {
+ "message" : [
+ "The Spark Connect plan is invalid."
+ ],
+ "subClass" : {
+ "CANNOT_PARSE" : {
+ "message" : [
+ "Cannot decompress or parse the input plan (<errorMsg>)"
+ ]
+ },
+ "PLAN_SIZE_LARGER_THAN_MAX" : {
+ "message" : [
+ "The plan size is larger than max (<planSize> vs. <maxPlanSize>)"
+ ]
Review Comment:
Let's add more info here on what the user facing resolution is.
For `CANNOT_PARSE`, it would be to disable compression and for the large plan size, add some information on when these situations may occur and offer an alternative (e.g split/reduce operations, utilise temp views for intermediate materialization etc)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54194][CONNECT] Spark Connect Proto Plan Compression [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #52894:
URL: https://github.com/apache/spark/pull/52894#issuecomment-3528618072
Hi, @xi-db , @hvanhovell , @zhengruifeng . This seems to break our Python Classic CIs. I made a follow-up. Could you review the PR?
- https://github.com/apache/spark/pull/53037
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54194][CONNECT] Spark Connect Proto Plan Compression [spark]
Posted by "xi-db (via GitHub)" <gi...@apache.org>.
xi-db commented on PR #52894:
URL: https://github.com/apache/spark/pull/52894#issuecomment-3522452675
Thanks @zhengruifeng , I've just added the doc change to the followup PR: https://github.com/apache/spark/pull/53003.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org