dongjoon-hyun opened a new pull request, #60034:
URL: https://github.com/apache/spark/pull/60034
### 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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
