[SPARK-57383][SQL][PYTHON] Honor configured Arrow zstd compression level when writing Arrow batches - #56444
Closed
viirya wants to merge 1 commit into
Closed
[SPARK-57383][SQL][PYTHON] Honor configured Arrow zstd compression level when writing Arrow batches#56444viirya wants to merge 1 commit into
viirya wants to merge 1 commit into
Conversation
…vel when writing Arrow batches The zstd compression level configured via spark.sql.execution.arrow.compression.zstd.level was silently ignored in all three places that compress Arrow batches: ArrowConverters, PythonArrowInput (also reused by GroupedPythonArrowInput) and CoGroupedArrowPythonRunner. They constructed ZstdCompressionCodec(level) only to read its codec type, then rebuilt the codec through CompressionCodec.Factory.createCodec(codecType), whose single-argument overload always builds a codec at the default level (3). Extract the codec construction into a shared ArrowCompressionUtils.createCompressionCodec helper that 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 it is unaffected. Add a regression test asserting that zstd level 19 compresses the same batch strictly smaller than level -5; it fails against the old codec construction (both levels produced byte-identical output). Co-authored-by: Claude Code
viirya
force-pushed
the
fix-arrow-zstd-level
branch
from
June 11, 2026 08:13
57f35dc to
5d7035b
Compare
zhengruifeng
approved these changes
Jun 11, 2026
viirya
added a commit
that referenced
this pull request
Jun 11, 2026
…vel when writing Arrow batches ### What changes were proposed in this pull request? 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 #56334 (#56334 (comment)); that PR fixes the cache-side instance of the pattern, and this PR fixes the remaining three pre-existing instances. ### 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. ### 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. 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. ### How was this patch tested? 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. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code Closes #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) Signed-off-by: Liang-Chi Hsieh <viirya@gmail.com>
viirya
added a commit
that referenced
this pull request
Jun 11, 2026
…vel when writing Arrow batches ### What changes were proposed in this pull request? 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 #56334 (#56334 (comment)); that PR fixes the cache-side instance of the pattern, and this PR fixes the remaining three pre-existing instances. ### 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. ### 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. 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. ### How was this patch tested? 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. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code Closes #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) Signed-off-by: Liang-Chi Hsieh <viirya@gmail.com>
Member
Author
|
Merged to master/branch-4.x/branch-4.2. I will create a backport for branch-4.1 manually. Thanks @zhengruifeng |
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>
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?
This PR fixes a bug where the zstd compression level configured via
spark.sql.execution.arrow.compression.zstd.levelwas silently ignored everywhere Arrow batches are compressed. Three places shared the same broken pattern:ArrowConverters.ArrowBatchIterator(SPARK-54134)PythonArrowInput(SPARK-54226; also coversGroupedPythonArrowInput, 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 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.The codec construction is extracted into a shared
ArrowCompressionUtils.createCompressionCodechelper that constructs the level-carrying codec instance directly (the helper lives insql/corebecausesql/api, whereArrowUtilsis, has noarrow-compressiondependency). 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 #56334 (#56334 (comment)); that PR fixes the cache-side instance of the pattern, and this PR fixes the remaining three pre-existing instances.
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.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. 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
GroupedPythonArrowInputstill has its own codec construction, since the SPARK-55328 deduplication is master-only.How was this patch tested?
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 thenonecodec and the unsupported-codec error.Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code