andygrove opened a new issue, #24022:
URL: https://github.com/apache/datafusion/issues/24022
### Describe the bug
`ScalarValue::new_list` takes a `data_type` parameter but honors it **only
when `values` is empty**:
```rust
// 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*:
```rust
// 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.
--
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]
