andygrove opened a new pull request, #6090:
URL: https://github.com/apache/datafusion-comet/pull/6090
## Which issue does this PR close?
Closes #6087.
## Rationale for this change
`ArrowWriter.writeColumns` drives its loop from the width of the input
`ColumnarBatch` while
indexing the writer's fields, which are built from the schema the batch is
being written under:
```scala
while (columnIndex < input.numCols()) {
fields(columnIndex).writeColumnSlice(input.column(columnIndex), startRow,
numRows)
```
That assumes every producer hands over a batch exactly as wide as the
schema. Iceberg's vectorized
reader does not. `BaseBatchReader.BatchDeleteFilter.filterBatch` reads with
`deletes.requiredSchema()`, which carries `_pos` after the projected columns
when a data file has
position deletes, and it only trims the extras back with
`ColumnarBatchUtil.removeExtraColumns`
when the file also has equality deletes. A merge-on-read UPDATE writes
position deletes and no
equality deletes, so the extra column survives into the batch.
Caching such a relation crashes. `InMemoryRelation` passes
`cachedPlan.output` as the cache schema,
and because `supportsColumnarInput` is true Spark strips the `ColumnarToRow`
above the cached plan,
so the serializer receives the scan's batches unaltered. Caching `SELECT
name FROM t` over
`id INT, name STRING` then gives the writer a 2-column batch and 1 field:
```
java.lang.ArrayIndexOutOfBoundsException: Index 1 out of bounds for length 1
at ArrowWriter.writeColumns(ArrowWriters.scala:171)
at
CometArrowConverters$.columnarBatchToArrowBatch(CometArrowConverters.scala:102)
at
ArrowCachedBatchSerializer.$anonfun$encodeBatches$1(ArrowCachedBatchSerializer.scala:231)
```
Comet's own row path already does the right thing: `ArrowWriter.write(row)`
loops `fields.length`.
So does Spark's `ArrowCachedBatchSerializer` (SPARK-57268), which converts a
non-Arrow batch
through `batch.rowIterator()` and the same schema-driven writer.
`writeColumns` was the one path
that trusted the batch instead of the schema.
## What changes are included in this PR?
`writeColumns` is now driven by the writer's fields rather than the input's
width, so trailing
columns the schema does not describe are ignored. Those are exactly the
columns Iceberg itself
discards when it trims, since `removeExtraColumns` keeps the leading
`expectedSchema` prefix.
The opposite mismatch is a real contract violation and stays fatal, but it
now fails with a message
that names both widths instead of an `ArrayIndexOutOfBoundsException` from
inside the write loop.
The direct-write path needed no change: `CachedBatchIpc.matchesReaderLayout`
already requires
`batch.numCols() == readerFields.length`, so a wider batch declines it and
falls into the
conversion path this PR fixes.
## How are these changes tested?
Two tests in `CometArrowStreamSuite`, which covers this boundary contract
directly: a batch wider
than the schema is written as the schema describes it, with the trailing
column ignored and the
projected values intact; a batch narrower than the schema is refused, with
the allocator left at
zero so the failure does not leak the partially built root.
--
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]
