[SPARK-53917][CONNECT] Support large local relations - follow-ups #52973
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
vicennial
reviewed
Nov 10, 2025
| "materialize all chunks in memory.") | ||
| .version("4.1.0") | ||
| .longConf | ||
| .checkValue(_ > 0, "The batch size in bytes must be positive") |
Contributor
There was a problem hiding this comment.
We should check if this value is greater than the chunk size value value as an initial step
Contributor
There was a problem hiding this comment.
In the case of conflicts, this conf should be respected and the operation should error out as we wouldn't want to bypass an explicitly set max materialisation size (to avoid system failures)
| @@ -6003,6 +6003,19 @@ object SQLConf { | |||
| .bytesConf(ByteUnit.BYTE) | |||
| .createWithDefaultString("3GB") | |||
|
|
|||
| val LOCAL_RELATION_BATCH_OF_CHUNKS_SIZE_BYTES = | |||
| buildConf(SqlApiConfHelper.LOCAL_RELATION_BATCH_OF_CHUNKS_SIZE_BYTES_KEY) | |||
Contributor
There was a problem hiding this comment.
We could update the name here to be a bit more explicit in the sense that this pertains to the maximum number of bytes that we will materialise in memory for the specific local relation
Contributor
There was a problem hiding this comment.
(specific because multi-threading can result in multiple artifacts being materialised at once)
hvanhovell
approved these changes
Nov 10, 2025
khakhlyuk
force-pushed
the
largelocalrelations-followup
branch
from
3ceed3c to
84a54fe
Compare
vicennial
approved these changes
Nov 12, 2025
Contributor
|
merging to master/4.1 |
asf-gitbox-commits
pushed a commit
that referenced
this pull request
Nov 12, 2025
### What changes were proposed in this pull request? This PR adds several fixes and improvements over #52613 which added support for 2GB+ local relations in Spark Connect. #### Uploading batches of chunks Currently, before caching the local relation on the server via `ChunkedCachedLocalRelation`, the client materializes all chunks in memory (1 schema chunk and N data chunks). This can lead to high memory pressure on the client when uploading very large local relations. In this PR, I'm changing how the client uploads the local relation. Instead of materializing all chunks in memory, the client will materialize a batch of chunks in memory, upload the batch of chunks to the server, and proceed to collecting the next batch of chunks. The size of the batch of chunks is controlled via `spark.sql.session.localRelationBatchOfChunksSizeBytes` (1GB by default). This way, the uploading mechanism only consumes 1GB of memory at each point in time. Alternatives to this approach are: a) uploading each chunk separately - would require one pair of `ArtifactStatuses` and `AddArtifactsRequest` RPC calls for each chunk, which is inefficient. b) (current implementation) materializing all chunks in memory and uploading them via a single pair of `ArtifactStatuses` and `AddArtifactsRequest` RPC calls. This can lead to high memory pressure on the client. Uploading batches of chunks is a middle-ground solution. Changes are implemented both for the python and scala clients. #### Minor fixes and improvements - Replace `ArraySeq.unsafeWrapArray(data.map(_.copy()).toArray))` with `data.map(_.copy()).toArray.toImmutableArraySeq` in `SparkConnectPlanner.scala`. The latter is compatible with both scala 2.13 and scala 2.12 and is consistent with how arrays are converted to sequences in other places in the code base. - Improved asserts and tests in the python client. ### Why are the changes needed? Reduce memory pressure in the spark connect python and scala clients when uploading very large local relations to the server. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Existing tests. ### Was this patch authored or co-authored using generative AI tooling? No Closes #52973 from khakhlyuk/largelocalrelations-followup. Authored-by: Alex Khakhlyuk <[email protected]> Signed-off-by: Herman van Hovell <[email protected]> (cherry picked from commit 40ba971) Signed-off-by: Herman van Hovell <[email protected]>
Member
|
Thank you, @khakhlyuk and @hvanhovell . |
Member
|
@khakhlyuk, I hit |
Contributor
Author
Member
|
@khakhlyuk, yes, I use an older version 4.1.0-preview3 connect server for testing. do we have a compatible policy for connect client/server, if it requires client version must be <= server version, I think we can ignore the issue. |
Contributor
Author
|
@pan3793 yes, exactly, the policy is to maintain forward compatibility - new servers have to support old clients. So client version <= server version. |
zifeif2
pushed a commit
to zifeif2/spark
that referenced
this pull request
Nov 22, 2025
### What changes were proposed in this pull request? This PR adds several fixes and improvements over apache#52613 which added support for 2GB+ local relations in Spark Connect. #### Uploading batches of chunks Currently, before caching the local relation on the server via `ChunkedCachedLocalRelation`, the client materializes all chunks in memory (1 schema chunk and N data chunks). This can lead to high memory pressure on the client when uploading very large local relations. In this PR, I'm changing how the client uploads the local relation. Instead of materializing all chunks in memory, the client will materialize a batch of chunks in memory, upload the batch of chunks to the server, and proceed to collecting the next batch of chunks. The size of the batch of chunks is controlled via `spark.sql.session.localRelationBatchOfChunksSizeBytes` (1GB by default). This way, the uploading mechanism only consumes 1GB of memory at each point in time. Alternatives to this approach are: a) uploading each chunk separately - would require one pair of `ArtifactStatuses` and `AddArtifactsRequest` RPC calls for each chunk, which is inefficient. b) (current implementation) materializing all chunks in memory and uploading them via a single pair of `ArtifactStatuses` and `AddArtifactsRequest` RPC calls. This can lead to high memory pressure on the client. Uploading batches of chunks is a middle-ground solution. Changes are implemented both for the python and scala clients. #### Minor fixes and improvements - Replace `ArraySeq.unsafeWrapArray(data.map(_.copy()).toArray))` with `data.map(_.copy()).toArray.toImmutableArraySeq` in `SparkConnectPlanner.scala`. The latter is compatible with both scala 2.13 and scala 2.12 and is consistent with how arrays are converted to sequences in other places in the code base. - Improved asserts and tests in the python client. ### Why are the changes needed? Reduce memory pressure in the spark connect python and scala clients when uploading very large local relations to the server. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Existing tests. ### Was this patch authored or co-authored using generative AI tooling? No Closes apache#52973 from khakhlyuk/largelocalrelations-followup. Authored-by: Alex Khakhlyuk <[email protected]> Signed-off-by: Herman van Hovell <[email protected]>
huangxiaopingRD
pushed a commit
to huangxiaopingRD/spark
that referenced
this pull request
Nov 25, 2025
### What changes were proposed in this pull request? This PR adds several fixes and improvements over apache#52613 which added support for 2GB+ local relations in Spark Connect. #### Uploading batches of chunks Currently, before caching the local relation on the server via `ChunkedCachedLocalRelation`, the client materializes all chunks in memory (1 schema chunk and N data chunks). This can lead to high memory pressure on the client when uploading very large local relations. In this PR, I'm changing how the client uploads the local relation. Instead of materializing all chunks in memory, the client will materialize a batch of chunks in memory, upload the batch of chunks to the server, and proceed to collecting the next batch of chunks. The size of the batch of chunks is controlled via `spark.sql.session.localRelationBatchOfChunksSizeBytes` (1GB by default). This way, the uploading mechanism only consumes 1GB of memory at each point in time. Alternatives to this approach are: a) uploading each chunk separately - would require one pair of `ArtifactStatuses` and `AddArtifactsRequest` RPC calls for each chunk, which is inefficient. b) (current implementation) materializing all chunks in memory and uploading them via a single pair of `ArtifactStatuses` and `AddArtifactsRequest` RPC calls. This can lead to high memory pressure on the client. Uploading batches of chunks is a middle-ground solution. Changes are implemented both for the python and scala clients. #### Minor fixes and improvements - Replace `ArraySeq.unsafeWrapArray(data.map(_.copy()).toArray))` with `data.map(_.copy()).toArray.toImmutableArraySeq` in `SparkConnectPlanner.scala`. The latter is compatible with both scala 2.13 and scala 2.12 and is consistent with how arrays are converted to sequences in other places in the code base. - Improved asserts and tests in the python client. ### Why are the changes needed? Reduce memory pressure in the spark connect python and scala clients when uploading very large local relations to the server. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Existing tests. ### Was this patch authored or co-authored using generative AI tooling? No Closes apache#52973 from khakhlyuk/largelocalrelations-followup. Authored-by: Alex Khakhlyuk <[email protected]> Signed-off-by: Herman van Hovell <[email protected]>
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
5 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
This PR adds several fixes and improvements over #52613 which added support for 2GB+ local relations in Spark Connect.
Uploading batches of chunks
Currently, before caching the local relation on the server via
ChunkedCachedLocalRelation, the client materializes all chunks in memory (1 schema chunk and N data chunks). This can lead to high memory pressure on the client when uploading very large local relations.In this PR, I'm changing how the client uploads the local relation. Instead of materializing all chunks in memory, the client will materialize a batch of chunks in memory, upload the batch of chunks to the server, and proceed to collecting the next batch of chunks. The size of the batch of chunks is controlled via
spark.sql.session.localRelationBatchOfChunksSizeBytes(1GB by default). This way, the uploading mechanism only consumes 1GB of memory at each point in time. Alternatives to this approach are:a) uploading each chunk separately - would require one pair of
ArtifactStatusesandAddArtifactsRequestRPC calls for each chunk, which is inefficient.b) (current implementation) materializing all chunks in memory and uploading them via a single pair of
ArtifactStatusesandAddArtifactsRequestRPC calls. This can lead to high memory pressure on the client.Uploading batches of chunks is a middle-ground solution.
Changes are implemented both for the python and scala clients.
Minor fixes and improvements
ArraySeq.unsafeWrapArray(data.map(_.copy()).toArray))withdata.map(_.copy()).toArray.toImmutableArraySeqinSparkConnectPlanner.scala. The latter is compatible with both scala 2.13 and scala 2.12 and is consistent with how arrays are converted to sequences in other places in the code base.Why are the changes needed?
Reduce memory pressure in the spark connect python and scala clients when uploading very large local relations to the server.
Does this PR introduce any user-facing change?
No
How was this patch tested?
Existing tests.
Was this patch authored or co-authored using generative AI tooling?
No