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

[andygrove]

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.

[tomatotomata]

I was thinking of taking a first pass on the narrow contract fix in ScalarValue::new_list: reconcile the collected non-empty values to the supplied data_type, then add regression coverage for nested nullability and the empty/non-empty paths. I would leave the broader PhysicalExpr invariant check separate so the PR stays focused. Does that boundary fit the DataFusion direction, and is there an existing scalar test module you would prefer for the case?