Repository navigation
[SPARK-61143][CONNECT] Decode the Arrow batches that the server compresses with zstd or lz4 in the Rust client, as PySpark does - #336
Closed
HyukjinKwon wants to merge 1 commit into
Conversation
…esses with zstd or lz4 in the Rust client, as PySpark does
With spark.sql.execution.arrow.compression.codec set to zstd or lz4, the server compresses the
Arrow batches of the results, which PySpark reads with PyArrow. The client built arrow without
the ipc_compression feature, so it failed to decode them ("zstd IPC decompression requires the
zstd feature"), in collect, toArrow and toPandas, also of the results of pandas UDFs. arrow now
has the feature, which adds no crate to the lock file but the zstd dependency of arrow-ipc.
Co-authored-by: Isaac <no-reply@databricks.com>
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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
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?
SPARK-61143: With
spark.sql.execution.arrow.compression.codecset tozstdorlz4, the Spark Connect server compresses the Arrow batches of the results. PySpark reads them with PyArrow.The Rust client now builds
arrowwith itsipc_compressionfeature, so it decodes the batches of either codec. Both clients get this: the Rust client, and the Python client, which decodes the results in Rust.The only change to
Cargo.lockis thezstddependency ofarrow-ipc.zstd, ofpolars-arrow, andlz4_flex, ofarrow-ipc, were already in the lock file.Why are the changes needed?
The client failed to decode the compressed batches:
So under that conf,
collect,toArrowandtoPandasfailed, also on the results of pandas UDFs andapplyInPandas. Nine tests of Apache Spark failed on this client, the*_with_compression_codec*tests oftest_parity_arrow.py,test_parity_pandas_grouped_map.pyandtest_parity_pandas_udf_scalar.py.Does this PR introduce any user-facing change?
Yes. The results compressed by the server with
zstdorlz4are decoded, as with PySpark 4.3.0.How was this patch tested?
none. It passes with official PySpark 4.3.0, and fails without this change.compressiontests of Apache Spark'stest_parity_arrow.py,test_parity_pandas_grouped_map.pyandtest_parity_pandas_udf_scalar.pypass on this client against a local server.wasm-udf, the offline Python suites (2202 passed, 8 skipped), and the live drop-in and Arrow suites (186 passed) and live cases (10 passed) against a local Spark 4.3.0 server. Clippy reports no diagnostics in the changed lines.Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
This pull request and its description were written by Isaac.