parthchandra opened a new issue, #7277:
URL: https://github.com/apache/datafusion-comet/issues/7277
Part of #7267.
## Problem
Comet's `ArrowCachedBatchSerializer` stores cached tables as Arrow batches.
Reads
emit one batch per cached batch (`ProjectedBatch`,
`ArrowCachedBatchSerializer.scala:612-653`)
and `CometInMemoryTableScanExec` (`:177`) feeds them to the native plan
through
`wrapColumnarBatchRDD` and `ScanExec`. Nothing on that path slices them.
The batch size is fixed when the cache is written, and three write paths
exist:
- **Row input** (`ArrowCachedBatchSerializer.scala:664`): Spark's
`conf.columnBatchSize`, i.e. `spark.sql.inMemoryColumnarStorage.batchSize`,
default 10000.
- **AQE cache build** (Spark 4.0+, `CometCacheBuildRule.scala:63`): builds
the cache
with `CometSparkToColumnarExec(child, Some(conf.columnBatchSize))`, also
10000.
- **Columnar input** (`ArrowCachedBatchSerializer.scala:519-530`): keeps the
batch
size of the plan that wrote it, i.e. the `spark.comet.batchSize` in effect
then.
None of them follows the current `spark.comet.batchSize`, so lowering it has
no
effect on queries over a cached table.
## How it fails
The row writers fail on the JVM before native code is involved:
`ArrowWriters.scala:589-616` (`setSafe`, `reserveData` / `checkDataEnd`)
throw
`OversizedAllocationException` once one column of one batch passes 2^31-1
bytes. At
10000 rows that is an average above ~215 KB per value. Slicing on read can't
help
that case.
## Proposed fix
- Write: use `min(spark.sql.inMemoryColumnarStorage.batchSize,
spark.comet.batchSize)`
in `convertInternalRowToCachedBatch` and in `CometCacheBuildRule`. This is
required, not optional.
- Read: slice any cached batch larger than the current
`spark.comet.batchSize`. Do it
in native `ScanExec` with `RecordBatch::slice`, which is zero-copy.
Slicing on the
Java side (`VectorSchemaRoot.slice` / `splitAndTransfer`) shares the data
buffer
but copies and rebases offsets, and can copy validity.
## Tests
- Scala test: cache a table with `spark.comet.batchSize` smaller than the
cache batch
size, then check the batches the native side receives are no bigger than
the
batch size. Cover the row, AQE (Spark 4.0+) and columnar write paths.
- SQL-file test: cache a table with a binary column of values a few hundred
KB in
size, filter it, and compare with Spark.
## Notes
- Related: #6834 (cache audit epic) and #6950 / PR #6988, which merge small
cached
batches on the columnar path up to
`spark.sql.inMemoryColumnarStorage.batchSize`.
That makes the columnar path follow the same 10000-row default, so the
write-side
`min()` here should apply to it as well.
- Changing the write-side size changes cache layout and the number of cached
batches. Benchmark cache build and scan before and after.
--
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]
