collect_list/collect_set can fail with "column types must match schema types" on nested-field nullability drift #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:

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.