[
https://issues.apache.org/jira/browse/SPARK-60707?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Andrew Grove updated SPARK-60707:
---------------------------------
Description:
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(n) + 8}} bytes.
# It points zstd 8 bytes into the buffer, past the uncompressed-length prefix.
# It still passes {{compressBound(n) + 8}} as the destination capacity.
zstd guarantees that it never writes past the capacity it is given, not that it
stays within {{{}compressBound(n){}}}. 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(n)}} 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(n){}}}.
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(n){}}}. {{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(n)}} 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.
was:
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(n) + 8}} bytes.
# It points zstd 8 bytes into the buffer, past the uncompressed-length prefix.
# It still passes \{{compressBound(n) + 8}} as the destination capacity.
zstd guarantees that it never writes past the capacity it is given, not that it
stays within \{{compressBound(n)}}. 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(n)}} 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(n)}}.
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(n)}}. \{{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(n)}} 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.
> 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: 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(n) + 8}} bytes.
> # It points zstd 8 bytes into the buffer, past the uncompressed-length
> prefix.
> # It still passes {{compressBound(n) + 8}} as the destination capacity.
> zstd guarantees that it never writes past the capacity it is given, not that
> it stays within {{{}compressBound(n){}}}. 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(n)}} 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(n){}}}.
> 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(n){}}}. {{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(n)}} 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]
