L. C. Hsieh created SPARK-60367:
-----------------------------------
Summary: Fix duplicated rows and closed reader root in sliced
Arrow output processors
Key: SPARK-60367
URL: https://issues.apache.org/jira/browse/SPARK-60367
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 5.0.0
Reporter: L. C. Hsieh
The sliced Arrow output processors of the Python runners (enabled by the
internal configs spark.sql.execution.arrow.maxRecordsPerOutputBatch and
spark.sql.execution.arrow.maxBytesPerOutputBatch, both disabled by default)
have two bugs:
1. SliceRecordsArrowOutputProcessorImpl returns the whole reader root for the
last chunk of an input batch that it has already sliced. Whenever a batch
returned from Python has more rows than maxRecordsPerOutputBatch, the rows
already output are output again (duplicated rows, silently wrong results). For
example, a batch of 10 rows with maxRecordsPerOutputBatch=3 is output as 3, 3,
3, then the whole 10 rows.
2. BaseSliceArrowOutputProcessor closes the previous output root when it
produces a different root, including the reader root itself. The reader root is
reused for the next input batches, so when an output batch is the whole reader
root and the next one is a slice, the next read fails, e.g. with
IndexOutOfBoundsException from StructVector.isSet or IllegalArgumentException
from slicing an empty vector. This affects both the records and the bytes
slicing.
The existing PySpark tests (MapInArrowWithOutputArrowBatchSlicing*Tests) did
not catch this because test_different_output_length compares a set of the
output values (which hides duplicates), and no test outputs a batch that fits
the limit followed by one that needs slicing.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
