ScalarValue::new_list silently discards its data_type argument for non-empty input #24022

Describe the bug

ScalarValue::new_list takes a data_type parameter but honors it only when values is empty:

// datafusion/common/src/scalar/mod.rs (main)
pub fn new_list(
    values: &[ScalarValue],
    data_type: &DataType,
    nullable: bool,
) -> Arc<ListArray> {
    let values = if values.is_empty() {
        new_empty_array(data_type)
    } else {
        Self::iter_to_array(values.iter().cloned()).unwrap()   // data_type ignored
    };
    ...
}

For non-empty input the element type comes entirely from the collected scalars, so the returned ListArray can have an element type that differs from the data_type the caller asked for. new_list_nullable and new_large_list inherit the same behaviour.

The practical consequence is that an accumulator built on new_list cannot guarantee the type it advertises. ArrayAggAccumulator and DistinctArrayAggAccumulator pass self.datatype to new_list, and datafusion-spark's collect_list / collect_set declare their return_type / state_fields from the argument's declared type:

// datafusion/spark/src/function/aggregate/collect.rs (main)
fn state_fields(&self, args: StateFieldsArgs) -> Result<Vec<FieldRef>> {
    Ok(vec![Field::new_list(
        format_state_name(args.name, "collect_set"),
        Field::new_list_field(args.input_fields[0].data_type().clone(), true),
        true,
    ).into()])
}

while the accumulator stores ScalarValue::try_from_array(...) slices of the argument array it actually received. Nothing reconciles the two.

So if an aggregate argument expression's evaluate() output type differs from its data_type(schema) — in our case in the nullability of a nested struct field, which DataType::equals_datatype does compare — then GroupedHashAggregateStream::emit fails validating its own output batch:

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

This hits both emit paths: the normal one (validated against AggregateExec's schema, from return_type) and the spill one (validated against spill_state.spill_schema, built from state_fields()).

To Reproduce

Build collect_set over an argument whose declared type has nullable nested leaves, and feed the accumulator an array whose actual type has non-nullable leaves. state_fields()[0].data_type() and the type of acc.state()[0] then disagree, and constructing the output RecordBatch fails with the message above.

Expected behavior

Either:

1. new_list / new_list_nullable / new_large_list reconcile the concatenated values to data_type (a cast_with_options call, which short-circuits to a shallow clone when the types already match), so the function's declared contract holds; or
2. the parameter is documented as advisory and ArrayAggAccumulator::evaluate / DistinctArrayAggAccumulator::evaluate cast to self.datatype themselves, so an accumulator cannot violate its own state_fields() / return_type().

(1) seems preferable — it fixes every current and future caller rather than two.

Additional context

Downstream, this shows up in Apache DataFusion Comet, which has to pre-normalize collect_list / collect_set arguments with an explicit cast to work around it.

Separately, and more broadly: there is currently no invariant check that a PhysicalExpr's evaluate() output type matches its data_type(schema). A debug-only assertion (in ProjectionExec / aggregate argument evaluation, alongside the existing ExecutionPlan invariant checks) would surface this class of bug at the drifting expression rather than several operators downstream in a spill path.