andygrove opened a new issue, #5137:
URL: https://github.com/apache/datafusion-comet/issues/5137
### Describe the bug
Comet has two independent sources of truth for a column's Arrow type:
1. **Declared** — Spark catalyst's type, serialized per nested field over
protobuf (`StructInfo.field_nullable`, `ListInfo.contains_null` in
`native/proto/src/proto/types.proto`) and rebuilt by `to_arrow_datatype`
(`native/core/src/execution/serde.rs`).
2. **Actual** — the Arrow type of the array a kernel, scan, or shuffle block
really produced.
Arrow treats nested field nullability as part of `DataType` identity:
`DataType::equals_datatype` compares `is_nullable()` at every list element,
struct field, and map entry, and `RecordBatch::try_new_with_options` has no
option to relax it (`match_field_names: false` only ignores field *names*). So
any operator that **stamps** the declared schema onto a runtime array
hard-fails when the two disagree by a single nested `nullable` flag, e.g.:
```
Invalid argument error: column types must match schema types,
expected List(Struct("id": Int64, "flag": Boolean))
but found List(Struct("id": Int64, "flag": non-null Boolean))
at column index 0
```
The data is fine in this situation — a non-nullable child is a strict subset
of a nullable one, and the batch would produce identical results. It is a
metadata assertion failure, not corruption.
Most boundaries already **normalize** instead of asserting, and that is why
this drift is usually invisible:
- `ScanExec::build_record_batch`
(`native/core/src/execution/operators/scan.rs`) casts every imported column to
the declared type. Arrow's cast rewrites nested fields from the target
(`cast_struct_to_struct` builds with `to_fields`, `cast_list_values` with
`to.clone()`), so the FFI boundary absorbs nullability drift.
- `SchemaAlignExec` (`native/shuffle/src/schema_align.rs`) casts the shuffle
writer's input, added for #4515.
- `CometLocalTableScanExec` widens its output with `.asNullable`, added for
#4789.
Two stamp points still assert:
- `ShuffleScanExec::poll_next`
(`native/core/src/execution/operators/shuffle_scan.rs`) — stamps the catalyst
schema onto the decoded shuffle block with no cast. This is the direct-read
shuffle path, enabled by default (`spark.comet.shuffle.directRead.enabled`).
- `ExpandStream::expand` (`native/core/src/execution/operators/expand.rs`) —
the schema is derived from `projections[0]` only (see `OpStruct::Expand` in
`native/core/src/execution/planner.rs`) and then stamped onto the output of
*every* projection, so the operator fails if any two projections disagree on a
nested nullable flag for the same output column.
Because the surviving stamp points are both inside native plans (past
`ScanExec`'s cast), the failure mode is: a native kernel or non-Parquet source
produces a nested type that is narrower than what catalyst declared, nothing
corrects it, and a downstream `ShuffleScanExec` or `ExpandExec` aborts the
task. Retries do not help; it is deterministic for the partition's data shape.
Upstream nested-nullability drift is tracked by #4515 (with concrete kernel
instances in #4789 and #4528). This issue is about the boundaries: the same
drift should be absorbed rather than fatal, the way the other boundaries
already do.
A secondary problem: the arrow error text names neither the operator nor the
column name, only `at column index N`. For a wide schema of deeply nested
structs this makes the report very hard to act on, since the two printed types
can differ by one flag hundreds of characters in.
### Steps to reproduce
Not yet reduced to a standalone test. Both sites need a native plan in which
a column's actual nested nullability differs from the catalyst-declared type:
- `ShuffleScanExec`: a native plan reading shuffle output where the block's
nested field nullability differs from the consuming stage's declared type.
- `ExpandExec`: a grouping-sets / rollup / multiple-distinct-aggregate plan
over an `array<struct<...>>` column where one projection's expression yields a
different nested nullable flag than `projections[0]`.
Writing the tests is part of the work here. The `SchemaAlignExec` unit tests
are a good model — they pin each drift pair so the test flips to a passthrough
assertion once the upstream type is corrected.
### Expected behavior
Neither boundary should fail on a nested nullability difference:
- `ShuffleScanExec` should reconcile the decoded block with the declared
schema the way `ScanExec` does (cast per column, or reuse `SchemaAlignExec`'s
alignment logic on the read side).
- `ExpandExec` should build its output schema so that it is valid for every
projection, not just `projections[0]` — e.g. widen nested nullability across
all projections, or align each projection's output to the operator's declared
schema before stamping.
Additionally, the error raised when a stamp genuinely cannot be reconciled
should name the operator and the offending column path, not just the column
index.
### Additional context
Related: #4515 (epic: return-type drift from DataFusion / datafusion-spark),
#4789 (non-null nested child fields from local table scans), #4528
(`make_array` with nullability-divergent struct children).
--
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]
