[
https://issues.apache.org/jira/browse/SPARK-60395?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
L. C. Hsieh updated SPARK-60395:
--------------------------------
Description:
SliceBytesArrowOutputProcessorImpl, used when the internal config
spark.sql.execution.arrow.maxBytesPerOutputBatch is set, has two problems in
how it measures and compares the size of an output batch:
1. getBatchBytes sums FieldVector.getBufferSize of the top-level vectors into a
Long, but for a struct vector the size is already truncated to an int inside
Arrow: NonNullableStructVector.getBufferSize sums its children in a long and
returns (int) buffer, and StructVector adds its validity size as an int. The
outputs of mapInArrow, mapInPandas, applyInPandas and Python UDTFs are wrapped
in one top-level struct vector, so a batch over 2 GiB can be measured as a
negative size. For example, a mapInArrow UDF that yields one batch of three
int64 columns with 2^27 rows (about 3 GiB) is measured as -1,006,632,960 bytes,
so with maxBytesPerOutputBatch=256m the whole batch is output without slicing.
The check on each slice can accept an oversized slice the same way. The size
should be computed by summing the buffer sizes of the leaf vectors (and the
validity buffers of the nested vectors) as Long.
2. The doc of spark.sql.execution.arrow.maxBytesPerOutputBatch says Spark will
try to create batches with the size equal or less than the value, but both
checks in SliceBytesArrowOutputProcessorImpl use <, so a batch whose size
equals the limit exactly is sliced. The records processor uses <=. The checks
or the doc should be aligned.
3. The shrink search in SliceBytesArrowOutputProcessorImpl.getRoot restarts
from all remaining rows for every output batch, and each probe that starts at a
non-zero row allocates and refills its offset buffers, so splitting one batch
of R rows into chunks of c rows copies O(R^2/c) offsets instead of O(R). The
first probe, slice(0, rowCount), also always fails since it measures the same
as the root that was just checked. The search should start from the last row
count that fit.
4. In the same loop, the loop ends only through return or with sliceBatchCount
== 0, so the following if (sliceBatchCount == 0) is always true and the initial
value of rootSlice is never read; the fallback can simply be
root.slice(currentRowIdx, 1).
5. The checkValue of spark.sql.execution.arrow.maxBytesPerOutputBatch bounds
only the upper side, while PythonArrowOutput narrows the value with .toInt, so
a negative value below Int.MinValue, which the doc calls no limit, wraps to a
positive limit. For example, -3g parses to -3221225472 and becomes a 1 GiB
limit, and -4294967295 becomes a 1-byte limit; if maxRecordsPerOutputBatch is
also set, the check that both limits are not set at once then throws although
the bytes limit is off. The value should be clamped before narrowing (e.g.
math.max(value, 0L).toInt), or the lower side bounded in checkValue.
6. In BaseSliceArrowOutputProcessor.loadBatch, the rowCount > 0 && clause never
changes the result, since currentRowIdx < rowCount already implies rowCount >=
1, and can be dropped.
Found while reviewing SPARK-60367 (https://github.com/apache/spark/pull/59616),
which this builds on.
was:
SliceBytesArrowOutputProcessorImpl, used when the internal config
spark.sql.execution.arrow.maxBytesPerOutputBatch is set, has two problems in
how it measures and compares the size of an output batch:
1. getBatchBytes sums FieldVector.getBufferSize of the top-level vectors into a
Long, but for a struct vector the size is already truncated to an int inside
Arrow: NonNullableStructVector.getBufferSize sums its children in a long and
returns (int) buffer, and StructVector adds its validity size as an int. The
outputs of mapInArrow, mapInPandas, applyInPandas and Python UDTFs are wrapped
in one top-level struct vector, so a batch over 2 GiB can be measured as a
negative size. For example, a mapInArrow UDF that yields one batch of three
int64 columns with 2^27 rows (about 3 GiB) is measured as -1,006,632,960 bytes,
so with maxBytesPerOutputBatch=256m the whole batch is output without slicing.
The check on each slice can accept an oversized slice the same way. The size
should be computed by summing the buffer sizes of the leaf vectors (and the
validity buffers of the nested vectors) as Long.
2. The doc of spark.sql.execution.arrow.maxBytesPerOutputBatch says Spark will
try to create batches with the size equal or less than the value, but both
checks in SliceBytesArrowOutputProcessorImpl use <, so a batch whose size
equals the limit exactly is sliced. The records processor uses <=. The checks
or the doc should be aligned.
3. The shrink search in SliceBytesArrowOutputProcessorImpl.getRoot restarts
from all remaining rows for every output batch, and each probe that starts at a
non-zero row allocates and refills its offset buffers, so splitting one batch
of R rows into chunks of c rows copies O(R^2/c) offsets instead of O(R). The
first probe, slice(0, rowCount), also always fails since it measures the same
as the root that was just checked. The search should start from the last row
count that fit.
4. In the same loop, the loop ends only through return or with sliceBatchCount
== 0, so the following if (sliceBatchCount == 0) is always true and the initial
value of rootSlice is never read; the fallback can simply be
root.slice(currentRowIdx, 1).
Found while reviewing SPARK-60367 (https://github.com/apache/spark/pull/59616),
which this builds on.
> Fix the batch size computation and limit check of maxBytesPerOutputBatch in
> sliced Arrow output
> -----------------------------------------------------------------------------------------------
>
> Key: SPARK-60395
> URL: https://issues.apache.org/jira/browse/SPARK-60395
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 4.2.0, 4.3.0, 4.1.3, 4.4.0
> Reporter: L. C. Hsieh
> Assignee: L. C. Hsieh
> Priority: Major
>
> SliceBytesArrowOutputProcessorImpl, used when the internal config
> spark.sql.execution.arrow.maxBytesPerOutputBatch is set, has two problems in
> how it measures and compares the size of an output batch:
> 1. getBatchBytes sums FieldVector.getBufferSize of the top-level vectors into
> a Long, but for a struct vector the size is already truncated to an int
> inside Arrow: NonNullableStructVector.getBufferSize sums its children in a
> long and returns (int) buffer, and StructVector adds its validity size as an
> int. The outputs of mapInArrow, mapInPandas, applyInPandas and Python UDTFs
> are wrapped in one top-level struct vector, so a batch over 2 GiB can be
> measured as a negative size. For example, a mapInArrow UDF that yields one
> batch of three int64 columns with 2^27 rows (about 3 GiB) is measured as
> -1,006,632,960 bytes, so with maxBytesPerOutputBatch=256m the whole batch is
> output without slicing. The check on each slice can accept an oversized slice
> the same way. The size should be computed by summing the buffer sizes of the
> leaf vectors (and the validity buffers of the nested vectors) as Long.
> 2. The doc of spark.sql.execution.arrow.maxBytesPerOutputBatch says Spark
> will try to create batches with the size equal or less than the value, but
> both checks in SliceBytesArrowOutputProcessorImpl use <, so a batch whose
> size equals the limit exactly is sliced. The records processor uses <=. The
> checks or the doc should be aligned.
> 3. The shrink search in SliceBytesArrowOutputProcessorImpl.getRoot restarts
> from all remaining rows for every output batch, and each probe that starts at
> a non-zero row allocates and refills its offset buffers, so splitting one
> batch of R rows into chunks of c rows copies O(R^2/c) offsets instead of
> O(R). The first probe, slice(0, rowCount), also always fails since it
> measures the same as the root that was just checked. The search should start
> from the last row count that fit.
> 4. In the same loop, the loop ends only through return or with
> sliceBatchCount == 0, so the following if (sliceBatchCount == 0) is always
> true and the initial value of rootSlice is never read; the fallback can
> simply be root.slice(currentRowIdx, 1).
> 5. The checkValue of spark.sql.execution.arrow.maxBytesPerOutputBatch bounds
> only the upper side, while PythonArrowOutput narrows the value with .toInt,
> so a negative value below Int.MinValue, which the doc calls no limit, wraps
> to a positive limit. For example, -3g parses to -3221225472 and becomes a 1
> GiB limit, and -4294967295 becomes a 1-byte limit; if
> maxRecordsPerOutputBatch is also set, the check that both limits are not set
> at once then throws although the bytes limit is off. The value should be
> clamped before narrowing (e.g. math.max(value, 0L).toInt), or the lower side
> bounded in checkValue.
> 6. In BaseSliceArrowOutputProcessor.loadBatch, the rowCount > 0 && clause
> never changes the result, since currentRowIdx < rowCount already implies
> rowCount >= 1, and can be dropped.
> Found while reviewing SPARK-60367
> (https://github.com/apache/spark/pull/59616), which this builds on.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
