andygrove opened a new issue, #5158:
URL: https://github.com/apache/datafusion-comet/issues/5158
### Describe the bug
`collect_list` / `collect_set` over a nested argument can fail at run time
with:
```
Invalid argument error: column types must match schema types,
expected List(Struct("flag": Boolean, "items": List(Struct("n": Int32))))
but found List(Struct("flag": non-null Boolean, "items": List(Struct("n":
non-null Int32))))
at column index 1
```
The two types are identical apart from the **nullability of nested fields**.
Arrow treats that as fatal: `RecordBatch::try_new` compares with
`DataType::equals_datatype`, which does compare nested field nullability.
There are two independent sources of truth for the element type, and nothing
forces them to agree:
* **Declared** — `SparkCollectSet::state_fields` / `return_type` copy the
aggregate argument's *declared* type verbatim (`datafusion-spark`,
`function/aggregate/collect.rs`).
* **Produced** — `DistinctArrayAggAccumulator` stores
`ScalarValue::try_from_array(...)` slices of the *actual* argument array, and
`ScalarValue::new_list` honors its `data_type` parameter **only when the value
set is empty** (`datafusion/common/src/scalar/mod.rs`); otherwise it just
concatenates the stored scalars. So the emitted list's element type is whatever
the runtime array carried.
If the declared argument type and the runtime array type disagree in nested
nullability, `GroupedHashAggregateStream::emit` fails when it validates its
output batch. This surfaces on both the normal emit path (validated against
`AggregateExec`'s schema, derived from `return_type`) and the spill path
(validated against `spill_state.spill_schema`, built from `state_fields()`).
### Why the existing guard does not cover this
`PhysicalPlanner::coerce_collect_child_nullability` (added in #4720) casts
the argument to the all-nullable variant of its type, but returns early when
the declared type is *already* all-nullable:
```rust
let child_type = child.data_type(schema.as_ref())?;
let nullable_type = make_all_fields_nullable(&child_type);
if child_type.equals_datatype(&nullable_type) {
Ok(child) // no cast inserted
} else {
Ok(Arc::new(CastExpr::new(child, nullable_type, None)))
}
```
That is exactly the failing case: the declared type is all-nullable, the
runtime array is not, no cast is inserted, and the drift reaches the
accumulator. The cast only helps in the opposite direction.
Separately, the rationale comment on that function is stale for DataFusion
54.1.0. It says the accumulator "build[s] their result list with all element
fields marked nullable, regardless of the input's nullability" — that is no
longer true; the accumulator preserves the input array's type, so declared and
produced already agree when there is no drift. The coercion is currently a
normalization, not a defence.
### To Reproduce
Nullability drift is hard to produce from Comet's own operators today,
because `ProjectionExec` and `ScanExec` both validate/cast their output batches
against their declared schemas. The failure is reachable when an argument
expression's `evaluate()` output type disagrees with its `data_type(schema)` —
for example `CreateNamedStruct` derives its fields from `batch.schema()` in
`evaluate()` but from the plan-time schema in `data_type()`
(`native/spark-expr/src/struct_funcs/create_named_struct.rs`), so the two
diverge for any batch whose schema object differs from the plan-time schema.
A unit test that builds `collect_set` against a schema with nullable leaves
and feeds it a batch whose array has non-nullable leaves reproduces the error
message above exactly.
### Expected behavior
The array handed to the `collect_list` / `collect_set` accumulator should
have exactly the type the plan declared for it, so the aggregate cannot fail
validating its own output.
### Additional context
Two fixes, not mutually exclusive:
1. Make the coercion an unconditional normalization barrier for nested
argument types, so the cast forces the runtime array to the declared type
regardless of what the argument produced. This is metadata-only —
`cast_with_options` shallow-clones when the source and target types already
match, and otherwise `cast_struct_to_struct` / `cast_list_values` stamp the
target fields and clone child buffers — so it does not copy data. The declared
output type is unchanged in every case, so it is not observable from Spark.
2. Harden `CreateNamedStruct` so it cannot drift: resolve its `Fields` once
from the plan-time schema at construction instead of recomputing them from
`batch.schema()` on every `evaluate()`.
I have a PR ready for (1); (2) is a behaviour-changing signature change and
is better as a follow-up.
The underlying defect is upstream: `ScalarValue::new_list` accepts a
`data_type` it then discards for non-empty input, so no accumulator built on it
can guarantee the type its `state_fields()` advertises. Filing that separately
against DataFusion.
--
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]
