[
https://issues.apache.org/jira/browse/SPARK-60707?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated SPARK-60707:
-----------------------------------
Labels: pull-request-available security (was: security)
> Arrow's ZstdCompressionCodec writes up to 8 bytes past its output buffer when
> Spark compresses Arrow batches with zstd
> ----------------------------------------------------------------------------------------------------------------------
>
> Key: SPARK-60707
> URL: https://issues.apache.org/jira/browse/SPARK-60707
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 4.1.0
> Reporter: Andrew Grove
> Priority: Major
> Labels: pull-request-available, security
>
> When {{spark.sql.execution.arrow.compression.codec}} is {{zstd}}, Spark
> compresses Arrow buffers with Arrow's {{ZstdCompressionCodec}}. In Arrow
> 18.3.0 and 19.0.0, the versions Spark uses on these branches, that codec
> tells zstd it has more room than its output buffer has, so zstd can write up
> to 8 bytes past the end of the buffer.
> {{ZstdCompressionCodec.doCompress}} does this:
> # It allocates {{compressBound(len) + 8}} bytes.
> # It points zstd 8 bytes into the buffer, past the uncompressed-length prefix.
> # It still passes {{compressBound(len) + 8}} as the destination capacity.
> zstd guarantees that it never writes past the capacity it is given, not that
> it stays within {{compressBound(len)}}. When it entropy-codes the literals of
> a block that turns out not to compress, it fills the destination up to the
> capacity it was given before falling back to a stored block. The write can
> therefore run 8 bytes past the buffer. Arrow's allocator rounds requests up
> to a power of two, which hides the overrun except when the request lands
> within 8 bytes below a power of two. When it does, the overrun lands in
> whatever follows the buffer, with no error. With Native Memory Tracking on,
> the JVM aborts when the buffer is freed.
> This is fixed in Arrow by [apache/arrow-java pull
> 1119|https://github.com/apache/arrow-java/pull/1119], released in Arrow
> 20.0.0, which passes {{compressBound(len)}} as the capacity.
> h4. Where Spark is exposed
> Spark constructs Arrow's codec directly in two places:
> * {{ArrowCompressionUtils.createCompressionCodec}}
> ([source|https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowCompressionUtils.scala])
> builds {{new ZstdCompressionCodec(zstdCompressionLevel)}}.
> {{ArrowConverters}} uses it for the Arrow IPC batches written for Python and
> Spark Connect ({{toPandas}}, {{toArrow}}, Connect result batches).
> * {{ArrowCachedBatchSerializer.createCompressionCodec}}
> ([source|https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ArrowCachedBatchSerializer.scala])
> builds {{new ZstdCompressionCodec(compressionLevel)}} for cached batches
> when {{spark.sql.cache.serializer}} is set to it.
> ||Branch||Version||Arrow||ArrowCompressionUtils||ArrowCachedBatchSerializer||
> |master|5.0.0-SNAPSHOT|19.0.0|yes|yes|
> |branch-4.3|4.3.1-SNAPSHOT|19.0.0|yes|yes|
> |branch-4.2|4.2.2-SNAPSHOT|19.0.0|yes|no|
> |branch-4.1|4.1.5-SNAPSHOT|18.3.0|yes|no|
> The default is {{spark.sql.execution.arrow.compression.codec=none}}, so only
> sessions that opt in to zstd are affected. I checked the bytecode of
> {{arrow-compression}} 19.0.0: {{doCompress}} loads {{compressBound + 8}} once
> and uses it both for {{allocator.buffer(...)}} and as the capacity argument
> to {{Zstd.compressUnsafe(address + 8, capacity, ...)}}.
> h4. Expected behavior
> The destination capacity passed to zstd is the size of the region that
> follows the length prefix, {{compressBound(len)}}.
> h4. Possible fix
> * Move to an Arrow release that has [arrow-java pull
> 1119|https://github.com/apache/arrow-java/pull/1119] (20.0.0), or
> * Until then, construct a subclass of {{ZstdCompressionCodec}} in
> {{ArrowCompressionUtils}} and {{ArrowCachedBatchSerializer}} whose
> {{doCompress}} passes {{compressBound(len)}}. {{doCompress}} is
> {{protected}}, and only compression needs to change; decompression stays
> Arrow's.
> Since a major Arrow bump cannot go into the 4.x branches, the Spark-side
> subclass is the fix to land on master and 4.x. It can be removed when Arrow
> is upgraded to 20.0.0 on master (Spark 5.0).
> h4. Context
> I found this while fixing the same codec in DataFusion Comet's cached-batch
> writer ([apache/datafusion-comet issue
> 7305|https://github.com/apache/datafusion-comet/issues/7305], [pull request
> 7307|https://github.com/apache/datafusion-comet/pull/7307]). That test
> allocates the codec's output from an allocator that places a sentinel guard
> directly behind each buffer. On Arrow 18.3.0 the guard is overwritten for a
> large share of inputs that zstd cannot compress, and never with
> {{compressBound(len)}} as the capacity. A reviewer reproduced the capacity
> arithmetic in a standalone program against zstd-jni 1.5.2-5 through 1.5.7-23,
> with the same result on every version.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
