andygrove opened a new issue, #7502:
URL: https://github.com/apache/datafusion-comet/issues/7502
Part of #7267.
## Problem
`CometSparkToColumnarExec` converts Spark output to Arrow in batches of
`spark.comet.batchSize` rows (`CometSparkToColumnarExec.scala:89`), with no
byte limit:
- From columnar input, `SparkColumnarArrowReader` appends consecutive Spark
batches until the Arrow batch has that many rows
(`SparkColumnarArrowReader.scala:108-131`). Spark's readers emit 4096 rows
(Parquet, ORC) or 5000 (Iceberg) and never merge them, so a column Spark reads
as two 1.1 GiB batches becomes one 2.2 GiB Arrow column.
- From row input, `RowArrowReader.scala:62-67` and
`CometArrowConverters.scala:73-79` (also used by `CometLocalTableScanExec`)
write rows until the batch has that many, so 8192 rows of 262 KB or more in one
column pass 2 GiB.
Either way, the Arrow writers throw `OversizedAllocationException`
(`ArrowWriters.scala:589-594`) and the task fails. Spark succeeds on the same
data.
The columnar path is on by default for the build side of a broadcast join
over a scan that Comet doesn't run natively, such as ORC or a V2 scan
(`spark.comet.convert.broadcastBuildSide.enabled`). That includes Iceberg
tables Comet reads with Iceberg's Java reader, which answers the epic's
"Iceberg Java fallback reader" item. A build side that large also fails in
`HashJoinExec` (#7273). The other routes are opt-in:
`spark.comet.convert.{parquet,json,csv,text}.enabled`,
`spark.comet.sparkToColumnar.enabled` and
`spark.comet.exec.localTableScan.enabled`.
Line references are for main at 4e6b12290f.
## Workaround
Lower `spark.comet.batchSize`.
## Proposed fix
Track the bytes of each variable-width column (nested children included) as
a Long while appending, and end the Arrow batch before any column would pass
the limit. The writers already compute a value's end offset before writing it
(`checkedEnd`, `ArrowWriters.scala:1504-1513`), so a row can be held back for
the next batch. `CometArrowPythonRunnerBase.inputBatchRanges` already plans
byte-bounded ranges the same way for Python input.
## Tests
Inject a small byte budget, feed two Spark batches (and the same with rows)
that together pass it, and assert that two Arrow batches come out.
--
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]
