Repository navigation
[SPARK-60816][SQL][PYTHON] Write empty Arrow buffers uncompressed so that PyArrow can read LZ4 batches - #60034
Closed
dongjoon-hyun wants to merge 1 commit into
Closed
dongjoon-hyun wants to merge 1 commit into
dongjoon-hyun wants to merge 1 commit into
Conversation
…that PyArrow can read LZ4 batches
dongjoon-hyun
force-pushed
the
SPARK-60816
branch
from
November 2, 2026 19:57
3282065 to
7fd1cea
Compare
Member
Author
|
Could you review this PR when you have some time, @peter-toth ? |
peter-toth
reviewed
Nov 3, 2026
peter-toth
left a comment
Contributor
There was a problem hiding this comment.
Thanks for the PR, @dongjoon-hyun!
The new UncompressedEmptyBuffers mixin writes an empty buffer as a -1 prefix with no body, the same output as apache/arrow-java#1248. Every compressed write path gets it, since they all build their codec with createCompressionCodec. Arrow C++ 18.0.0 and Arrow Java both read that shape back as an empty buffer. I ran the description's toArrow example on master with lz4 and got the LZ4 frame error, so the new Python test does fail without the fix. No findings from my side.
peter-toth
approved these changes
Nov 3, 2026
Member
Author
|
Thank you so much, @peter-toth ! |
dongjoon-hyun
added a commit
that referenced
this pull request
Nov 3, 2026
…that PyArrow can read LZ4 batches ### What changes were proposed in this pull request? This PR adds `UncompressedEmptyBuffers`, a stackable trait for Arrow's `CompressionCodec`, and mixes it into both codecs that `ArrowCompressionUtils.createCompressionCodec` creates for `spark.sql.execution.arrow.compression.codec`: `BoundedZstdCompressionCodec` and `Lz4CompressionCodec`. For an empty input buffer (`writerIndex == 0`), the trait writes the 8-byte length prefix `-1` (`CompressionUtil.NO_COMPRESSION_LENGTH`), which means that the body is stored uncompressed. All other buffers go to the codec unchanged. Like Arrow's `AbstractCompressionCodec.compress`, the trait closes the input buffer only on success, because `TrackingCompressionCodec` (SPARK-60776) releases the input when `compress` fails. This is the same output as the upstream fix, apache/arrow-java#1248, which is only available in the unreleased Arrow 20.0.0. The trait has a note to remove it once Spark is on that version. ### Why are the changes needed? With `spark.sql.execution.arrow.compression.codec=lz4`, PyArrow fails to read a batch if a string or binary column in the batch has an empty data buffer. ```python spark.conf.set("spark.sql.execution.arrow.compression.codec", "lz4") spark.range(3).selectExpr("CAST(NULL AS STRING) AS s").toArrow() ``` ``` OSError: Lz4 compressed input contains less than one frame ``` Arrow Java up to [v19.0.0](https://github.com/apache/arrow-java/releases/tag/v19.0.0) (2026-03-16) writes an empty buffer as the length prefix `0` with no bytes after it (apache/arrow-java#1196). The Arrow Java reader treats that as an empty buffer, but Arrow C++ passes the zero-byte body to the codec, and the LZ4 frame decoder rejects it because it contains no frame. The same code path is in Arrow C++ 18.0.0, the minimum PyArrow version of PySpark. A data buffer is empty when all values of a batch are null or `""`, including nested fields, and in 0-row batches. So the following fail with `lz4`: - `toArrow` and `toPandas` - Arrow-based Python UDFs and functions such as `pandas_udf`, `arrow_udf`, `useArrow=True`, `mapInArrow`, `mapInPandas`, and `applyInPandas` - Cogrouped `applyInPandas` with a key on only one side, even without nulls, because the other side is sent as a 0-row batch - Spark Connect `collect`, `toArrow`, and `toPandas`, including an empty result that has a string column, even without nulls Other buffers also get the `0` prefix when they are empty, but Arrow C++ does not decompress them in that case, so they do not fail. The `-1` prefix is part of the Arrow IPC format, and Arrow Java already writes it for a buffer that does not compress, so readers already handle it. This PR applies it to zstd as well because a `0` prefix with no bytes is not a valid zstd frame either, even though Arrow C++ accepts it, and Arrow 20.0.0 writes `-1` for all codecs. Arrow 20.0.0 is not released yet, and a Spark-side fix can also be backported to the maintenance branches, which are on Arrow 18.3.0 and 19.0.0. ### Does this PR introduce _any_ user-facing change? Yes. The reads above no longer fail when `spark.sql.execution.arrow.compression.codec` is `lz4`. For example, the above `toArrow` call now returns a table whose `s` column has three nulls. This failure exists since [v4.1.0](https://github.com/apache/spark/releases/tag/v4.1.0) (2025-12-11), which added the configuration. The default value is `none`, which is not affected. With `zstd`, and in `ArrowCachedBatchSerializer`, which is read back by Arrow Java, the only change is the prefix of the empty buffers. The results are the same. ### How was this patch tested? Pass the CIs with the newly added test cases. - A new test in `ArrowCompressionUtilsSuite` checks that `lz4` and `zstd` write an empty buffer as an 8-byte `-1` prefix and that the input buffer is released. Without this fix, it fails with `0 did not equal -1`. - Another new test in the same suite makes the allocation of the prefix fail and checks that `createVectorUnloader` releases the empty buffer exactly once. It also passes without this fix. It fails with `RefCnt has gone negative` if the trait closes the input buffer on failure as well. - `test_toArrow_with_compression_codec_empty_data_buffers` in `test_arrow.py` reads all-null strings with PyArrow for each codec. It also runs over Spark Connect through `ArrowParityTests`. Without this fix, its `lz4` case fails with the above error in both modes. I also ran the following locally with PyArrow 25.0.0. - `ArrowCompressionUtilsSuite`, `ArrowCachedBatchSerializerSuite`, `ArrowConvertersSuite`, and `ExecuteExternalUDFExecSuite` - The existing compression tests in `test_arrow.py`, `test_pandas_udf_scalar.py`, `test_pandas_grouped_map.py`, and `test_pandas_udf_grouped_agg.py` - An ad-hoc script that runs the failing paths listed above and their controls with all three codecs (78 cases), and another that reads cached batches from `ArrowCachedBatchSerializer` through `collect`, `toArrow`, and `pandas_udf` (48 cases) ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Opus 5.5 Closes #60034 from dongjoon-hyun/SPARK-60816. Authored-by: Dongjoon Hyun <dongjoon@apache.org> Signed-off-by: Dongjoon Hyun <dongjoon@apache.org> (cherry picked from commit 57f7ad6) Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>
Member
Author
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?
This PR adds
UncompressedEmptyBuffers, a stackable trait for Arrow'sCompressionCodec, and mixes it into both codecs thatArrowCompressionUtils.createCompressionCodeccreates forspark.sql.execution.arrow.compression.codec:BoundedZstdCompressionCodecandLz4CompressionCodec.For an empty input buffer (
writerIndex == 0), the trait writes the 8-byte length prefix-1(CompressionUtil.NO_COMPRESSION_LENGTH), which means that the body is stored uncompressed. All other buffers go to the codec unchanged.Like Arrow's
AbstractCompressionCodec.compress, the trait closes the input buffer only on success, becauseTrackingCompressionCodec(SPARK-60776) releases the input whencompressfails.This is the same output as the upstream fix, apache/arrow-java#1248, which is only available in the unreleased Arrow 20.0.0. The trait has a note to remove it once Spark is on that version.
Why are the changes needed?
With
spark.sql.execution.arrow.compression.codec=lz4, PyArrow fails to read a batch if a string or binary column in the batch has an empty data buffer.Arrow Java up to v19.0.0 (2026-03-16) writes an empty buffer as the length prefix
0with no bytes after it (apache/arrow-java#1196). The Arrow Java reader treats that as an empty buffer, but Arrow C++ passes the zero-byte body to the codec, and the LZ4 frame decoder rejects it because it contains no frame. The same code path is in Arrow C++ 18.0.0, the minimum PyArrow version of PySpark.A data buffer is empty when all values of a batch are null or
"", including nested fields, and in 0-row batches. So the following fail withlz4:toArrowandtoPandaspandas_udf,arrow_udf,useArrow=True,mapInArrow,mapInPandas, andapplyInPandasapplyInPandaswith a key on only one side, even without nulls, because the other side is sent as a 0-row batchcollect,toArrow, andtoPandas, including an empty result that has a string column, even without nullsOther buffers also get the
0prefix when they are empty, but Arrow C++ does not decompress them in that case, so they do not fail.The
-1prefix is part of the Arrow IPC format, and Arrow Java already writes it for a buffer that does not compress, so readers already handle it. This PR applies it to zstd as well because a0prefix with no bytes is not a valid zstd frame either, even though Arrow C++ accepts it, and Arrow 20.0.0 writes-1for all codecs.Arrow 20.0.0 is not released yet, and a Spark-side fix can also be backported to the maintenance branches, which are on Arrow 18.3.0 and 19.0.0.
Does this PR introduce any user-facing change?
Yes. The reads above no longer fail when
spark.sql.execution.arrow.compression.codecislz4. For example, the abovetoArrowcall now returns a table whosescolumn has three nulls. This failure exists since v4.1.0 (2025-12-11), which added the configuration. The default value isnone, which is not affected.With
zstd, and inArrowCachedBatchSerializer, which is read back by Arrow Java, the only change is the prefix of the empty buffers. The results are the same.How was this patch tested?
Pass the CIs with the newly added test cases.
ArrowCompressionUtilsSuitechecks thatlz4andzstdwrite an empty buffer as an 8-byte-1prefix and that the input buffer is released. Without this fix, it fails with0 did not equal -1.createVectorUnloaderreleases the empty buffer exactly once. It also passes without this fix. It fails withRefCnt has gone negativeif the trait closes the input buffer on failure as well.test_toArrow_with_compression_codec_empty_data_buffersintest_arrow.pyreads all-null strings with PyArrow for each codec. It also runs over Spark Connect throughArrowParityTests. Without this fix, itslz4case fails with the above error in both modes.I also ran the following locally with PyArrow 25.0.0.
ArrowCompressionUtilsSuite,ArrowCachedBatchSerializerSuite,ArrowConvertersSuite, andExecuteExternalUDFExecSuitetest_arrow.py,test_pandas_udf_scalar.py,test_pandas_grouped_map.py, andtest_pandas_udf_grouped_agg.pyArrowCachedBatchSerializerthroughcollect,toArrow, andpandas_udf(48 cases)Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5