andygrove opened a new pull request, #4843:
URL: https://github.com/apache/datafusion-comet/pull/4843
## Which issue does this PR close?
Closes #4789.
## Rationale for this change
Several Comet array/map expressions crash natively when the input comes
through `CometLocalTableScanExec` with a non-nullable child field. An in-memory
`Seq[Int]` / `Map[Int, Int]` column encodes its array element / map value as
non-null (`containsNull=false` / `valueContainsNull=false`), but Comet
expression serdes promise nullable children. Feeding a non-null child into a
native kernel disagrees with the planned output type:
- `slice`: `Assertion failed: result_data_type == *expected_type: Function
'spark_array_slice' returned value of type 'List(non-null Int32)' while ...
expected: 'List(Int32)'`
- `map_entries`: `ListArray expected data type Struct("key": non-null Int32,
"value": Int32) got Struct("key": non-null Int32, "value": non-null Int32)`
- `array_insert` / SPARK-41233 array prepend: analogous item-vs-array type
mismatch
It does not reproduce over a native Parquet scan (which normalizes children
to nullable) or with a SQL `array(...)` / `map(...)` literal (literal children
are nullable), so it is specific to the local scan path.
## What changes are included in this PR?
The non-null child is carried in two places, so both must be reconciled:
1. The Arrow batch produced by `CometLocalTableScanExec.mapToReaders` (built
from `originalPlan.schema`).
2. The native `ScanExec`'s declared schema, serialized from `op.output` in
`CometSink.convert`. `scan.rs:build_record_batch` casts the imported batch back
to this declared schema, so widening only the Arrow data is reverted by the
cast.
Changes:
- Add `QueryPlanSerde.widenChildNullability`, which recursively widens array
elements, map values, and struct fields to nullable while leaving top-level
types and map keys unchanged (map keys stay non-null, matching Arrow, and their
nullability is never serialized).
- Add an overridable `scanFieldType` hook in `CometSink` (default identity)
so the widening is scoped to the local table scan and does not affect
Union/Coalesce/shuffle sinks, whose children come from already-normalized
native execution.
- `CometLocalTableScanExec` overrides the hook and applies the same widening
to its Arrow schema, so the declared schema and the batch agree exactly and no
per-batch cast is needed.
This mirrors how the Parquet scan path already normalizes children to
nullable. No native/Rust changes are required.
## How are these changes tested?
Added regression tests that run through the local table scan with
`ConvertToLocalRelation` disabled (so the expression executes natively rather
than being folded at plan time):
- `CometArrayExpressionSuite`: `slice` and `array_insert` on
non-null-element arrays.
- `CometMapExpressionSuite`: `map_entries` on a non-null-value map.
All new tests pass, along with the existing `CometArrayExpressionSuite`,
`CometMapExpressionSuite`, and the `LocalTableScan` tests in `CometExecSuite`
(including nested `NullType` in struct/array/map).
--
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]
