[SPARK-57383][SQL][PYTHON][4.1] Honor configured Arrow zstd compression level when writing Arrow batches - #56451
Closed
viirya wants to merge 1 commit into
Closed
[SPARK-57383][SQL][PYTHON][4.1] Honor configured Arrow zstd compression level when writing Arrow batches#56451viirya wants to merge 1 commit into
viirya wants to merge 1 commit into
Conversation
…vel when writing Arrow batches This PR fixes a bug where the zstd compression level configured via `spark.sql.execution.arrow.compression.zstd.level` was silently ignored everywhere Arrow batches are compressed. Three places shared the same broken pattern: - `ArrowConverters.ArrowBatchIterator` (SPARK-54134) - `PythonArrowInput` (SPARK-54226; also covers `GroupedPythonArrowInput`, which reuses this codec via SPARK-55328) - `CoGroupedArrowPythonRunner` (SPARK-54226) They constructed `new ZstdCompressionCodec(level)` only to read its codec type, then rebuilt the codec through `CompressionCodec.Factory.INSTANCE.createCodec(codecType)`. The codec type enum does not carry a level, so that single-argument factory overload always builds a codec at the zstd default level (3), dropping the configured one. The codec construction is extracted into a shared `ArrowCompressionUtils.createCompressionCodec` helper that constructs the level-carrying codec instance directly (the helper lives in `sql/core` because `sql/api`, where `ArrowUtils` is, has no `arrow-compression` dependency). The level only matters on the write side; the read side looks up the codec by the type recorded in the IPC message, so reads are unaffected and the on-wire format is unchanged. The same bug class was found by dbtsai during review of apache#56334 (apache#56334 (comment)); that PR fixes the cache-side instance of the pattern, and this PR fixes the remaining three pre-existing instances. Users tuning `spark.sql.execution.arrow.compression.zstd.level` for Python UDF exchange or `df.toArrow()` got no effect at all: every level compressed identically at the default level 3, with no error or warning. Yes. The configured zstd level now actually takes effect; previously all levels behaved like the default level 3. The bug exists in released Spark 4.1.0/4.1.1/4.1.2 (SPARK-54134 and SPARK-54226 were backported to branch-4.1) as well as 4.2.0 RCs and master, so this fix is a candidate for backporting to branch-4.1 and branch-4.2. Note that a branch-4.1 backport needs to fix a fourth copy of the pattern: there `GroupedPythonArrowInput` still has its own codec construction, since the SPARK-55328 deduplication is master-only. New `ArrowCompressionUtilsSuite`. The regression test compresses the same compressible-but-varying batch at zstd level -5 and level 19 and asserts level 19 produces a strictly smaller payload. Against the old codec construction this test fails with byte-identical sizes at both levels (verified locally). A second test covers the `none` codec and the unsupported-codec error. Generated-by: Claude Code Closes apache#56444 from viirya/fix-arrow-zstd-level. Authored-by: Liang-Chi Hsieh <viirya@gmail.com> Signed-off-by: Liang-Chi Hsieh <viirya@gmail.com> (cherry picked from commit e33017a)
dongjoon-hyun
approved these changes
Jun 11, 2026
dongjoon-hyun
left a comment
dongjoon-hyun
left a comment
Member
There was a problem hiding this comment.
+1, LGTM (Pending CIs).
viirya
added a commit
that referenced
this pull request
Jun 11, 2026
…on level when writing Arrow batches ### What changes were proposed in this pull request? Backport of #56444 (commit e33017a) to branch-4.1. This fixes a bug where the zstd compression level configured via `spark.sql.execution.arrow.compression.zstd.level` was silently ignored everywhere Arrow batches are compressed. The affected code constructed `new ZstdCompressionCodec(level)` only to read its codec type, then rebuilt the codec through `CompressionCodec.Factory.INSTANCE.createCodec(codecType)`. The codec type enum does not carry a level, so that single-argument factory overload always builds a codec at the zstd default level (3), dropping the configured one. Compared to the master patch, this backport fixes one additional copy of the broken pattern: on branch-4.1 `GroupedPythonArrowInput` still has its own codec construction in `createUnloaderForGroup` (the SPARK-55328 deduplication that makes it reuse `PythonArrowInput.codec` is master/4.2-only). The four fixed sites are: - `ArrowConverters.ArrowBatchIterator` - `PythonArrowInput` - `GroupedPythonArrowInput` (branch-4.1 only; folded into `PythonArrowInput` on master) - `CoGroupedArrowPythonRunner` All four now use the shared `ArrowCompressionUtils.createCompressionCodec` helper, which constructs the level-carrying codec instance directly. The level only matters on the write side; the read side looks up the codec by the type recorded in the IPC message, so reads are unaffected and the on-wire format is unchanged. ### Why are the changes needed? Users tuning `spark.sql.execution.arrow.compression.zstd.level` for Python UDF exchange or `df.toArrow()` got no effect at all: every level compressed identically at the default level 3, with no error or warning. Released 4.1.0/4.1.1/4.1.2 are affected. ### Does this PR introduce _any_ user-facing change? Yes. The configured zstd level now actually takes effect; previously all levels behaved like the default level 3. ### How was this patch tested? `ArrowCompressionUtilsSuite`, brought over with the backport. The regression test compresses the same compressible-but-varying batch at zstd level -5 and level 19 and asserts level 19 produces a strictly smaller payload; against the old codec construction it fails with byte-identical sizes at both levels. Verified the suite passes on this branch. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code Closes #56451 from viirya/SPARK-57383-branch-4.1. Authored-by: Liang-Chi Hsieh <viirya@gmail.com> Signed-off-by: Liang-Chi Hsieh <viirya@gmail.com>
Member
Author
|
Merged to branch-4.1. Thanks @dongjoon-hyun |
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
2 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?
Backport of #56444 (commit e33017a) to branch-4.1.
This fixes a bug where the zstd compression level configured via
spark.sql.execution.arrow.compression.zstd.levelwas silently ignored everywhere Arrow batches are compressed. The affected code constructednew ZstdCompressionCodec(level)only to read its codec type, then rebuilt the codec throughCompressionCodec.Factory.INSTANCE.createCodec(codecType). The codec type enum does not carry a level, so that single-argument factory overload always builds a codec at the zstd default level (3), dropping the configured one.Compared to the master patch, this backport fixes one additional copy of the broken pattern: on branch-4.1
GroupedPythonArrowInputstill has its own codec construction increateUnloaderForGroup(the SPARK-55328 deduplication that makes it reusePythonArrowInput.codecis master/4.2-only). The four fixed sites are:ArrowConverters.ArrowBatchIteratorPythonArrowInputGroupedPythonArrowInput(branch-4.1 only; folded intoPythonArrowInputon master)CoGroupedArrowPythonRunnerAll four now use the shared
ArrowCompressionUtils.createCompressionCodechelper, which constructs the level-carrying codec instance directly. The level only matters on the write side; the read side looks up the codec by the type recorded in the IPC message, so reads are unaffected and the on-wire format is unchanged.Why are the changes needed?
Users tuning
spark.sql.execution.arrow.compression.zstd.levelfor Python UDF exchange ordf.toArrow()got no effect at all: every level compressed identically at the default level 3, with no error or warning. Released 4.1.0/4.1.1/4.1.2 are affected.Does this PR introduce any user-facing change?
Yes. The configured zstd level now actually takes effect; previously all levels behaved like the default level 3.
How was this patch tested?
ArrowCompressionUtilsSuite, brought over with the backport. The regression test compresses the same compressible-but-varying batch at zstd level -5 and level 19 and asserts level 19 produces a strictly smaller payload; against the old codec construction it fails with byte-identical sizes at both levels. Verified the suite passes on this branch.Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code