andygrove opened a new issue, #6087:
URL: https://github.com/apache/datafusion-comet/issues/6087
### Describe the bug
Comet's cache serializer fails with `ArrayIndexOutOfBoundsException` when
the columnar batch it is
handed has more columns than the cached relation's output schema.
```
java.lang.ArrayIndexOutOfBoundsException: Index 1 out of bounds for length 1
at
org.apache.spark.sql.comet.execution.arrow.ArrowWriter.writeColumns(ArrowWriters.scala:171)
at
org.apache.spark.sql.comet.execution.arrow.CometArrowConverters$.columnarBatchToArrowBatch(CometArrowConverters.scala:102)
at
org.apache.spark.sql.comet.execution.arrow.ArrowCachedBatchSerializer.$anonfun$encodeBatches$1(ArrowCachedBatchSerializer.scala:231)
at
org.apache.spark.sql.execution.columnar.CachedRDDBuilder$$anon$2.next(InMemoryRelation.scala:342)
at
org.apache.spark.storage.memory.MemoryStore.putIterator(MemoryStore.scala:232)
```
### Root cause
`ArrowWriter.writeColumns` drives its loop from the *batch* width while
indexing the writer's
fields, which are built from the cached relation's attributes:
```scala
def writeColumns(input: ColumnarBatch, startRow: Int, numRows: Int): Unit = {
var columnIndex = 0
while (columnIndex < input.numCols()) {
fields(columnIndex).writeColumnSlice(input.column(columnIndex),
startRow, numRows)
columnIndex += 1
}
```
`InMemoryRelation.buildBuffers` passes `cachedPlan.output` as the cache
schema and
`cachedPlan.executeColumnar()` as the input. Because `supportsColumnarInput`
returns true, Spark
strips the `ColumnarToRow` above the cached plan, so the serializer receives
the scan's raw
batches, whatever shape the connector produces them in.
Iceberg's vectorized reader emits batches wider than the scan's output
schema.
`BaseBatchReader.BatchDeleteFilter.filterBatch` reads with
`deletes.requiredSchema()`, which
carries `_pos` when a data file has position deletes, and it only trims the
extras back with
`ColumnarBatchUtil.removeExtraColumns` when `deletes.hasEqDeletes()` is
true. A merge-on-read
UPDATE writes position deletes and no equality deletes, so the extra column
survives: caching
`SELECT name FROM t` over `id INT, name STRING` hands the writer a `[name,
_pos]` batch (2 columns)
while it holds 1 field, which is the out-of-bounds index above. Iceberg
removes extras from the end
of the vector array, so the projected columns are always the leading ones.
The direct-write path is unaffected: `CachedBatchIpc.matchesReaderLayout`
requires
`batch.numCols() == readerFields.length`, so a wider batch already declines
that path and falls
into the conversion path, which is where it fails.
For comparison, Spark's own `ArrowCachedBatchSerializer` (SPARK-57268)
converts a non-Arrow batch
through `batch.rowIterator()` and `ArrowWriter.write(row)`, and that loop is
driven by
`fields.length`, so it tolerates a wider batch. Comet's
`ArrowWriter.write(row)` is driven by
`fields.length` too. `writeColumns` is the only path in Comet that trusts
the batch width.
### Steps to reproduce
With `spark.comet.exec.inMemoryCache.enabled=true`, against an Iceberg
merge-on-read table
(`format-version` 2 or 3, Parquet, vectorized reads):
```sql
CREATE TABLE t (id INT, name STRING);
INSERT INTO t VALUES (1, 'n1');
CACHE TABLE tmp AS SELECT name FROM t;
MERGE INTO t USING source s ON t.id = s.id WHEN MATCHED THEN UPDATE SET
t.name = s.name;
SELECT * FROM tmp;
```
In CI this reproduces as
`TestMergeOnReadMerge.testMergeRefreshesRelationCache`
(`catalogName = testhadoop`, `format = PARQUET`, `vectorized = true`,
`formatVersion = 3`).
### Expected behavior
Writing a cached batch should be driven by the cache schema rather than the
batch width: write the
first `fields.length` columns and ignore trailing ones, the same columns
Iceberg itself keeps when
it trims. A batch with fewer columns than the schema is a genuine contract
violation and should
still fail loudly.
### Additional context
Found by #5634, which enables `spark.comet.exec.inMemoryCache.enabled` by
default so that the whole
CI matrix exercises the cache format rather than only the two suites that
opt in.
Failing job:
https://github.com/apache/datafusion-comet/actions/runs/34270155816/job/102213466325
--
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]
