[I] collect_list/collect_set can fail with "column types must match schema types" on nested-field nullability drift [datafusion-comet]

["andygrove (via GitHub)" <gi...@apache.org>]

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: github-unsubscribe@datafusion.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [I] collect_list/collect_set can fail with "column types must match schema types" on nested-field nullability drift [datafusion-comet]

["andygrove (via GitHub)" <gi...@apache.org>]

andygrove closed issue #5158: collect_list/collect_set can fail with "column types must match schema types" on nested-field nullability drift
URL: https://github.com/apache/datafusion-comet/issues/5158


-- 
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: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org