bug: `SparkCollectList`/`SparkCollectSet` declare nullable list elements, diverging from Spark's `containsNull = false`
- Dominant language
- Rust
- Stars
- 9.3k
- Forks
- 2.4k
- Avg merge
- 3d 7h
- Merged PRs (30d)
- 344
Description
### Describe the bug
`SparkCollectList` and `SparkCollectSet` in `datafusion-spark` declare their result as a list with a **nullable** element field:
https://github.com/apache/datafusion/blob/main/datafusion/spark/src/function/aggregate/collect.rs
```rust
fn return_type(&self, arg_types: &[DataType]) -> Result {
Ok(DataType::List(Arc::new(Field::new_list_field(
arg_types[0].clone(),
true,
))))
}
```
Spark declares both `collect_list` and `collect_set` as `ArrayType(child.dataType, containsNull = false)` — correctly, because both functions unconditionally drop null inputs (and the accumulators here hardcode `ignore_nulls = true`), so a null element can never appear in the result.
Since the elements are provably never null, the declared element field should be non-nullable, and the arrays the accumulators produce should carry the same non-nullable field.
There is a related inconsistency for nested inputs: `return_type` clones the input type verbatim (preserving any non-nullable *inner* fields, e.g. a non-nullable struct field), while the accumulators (`ArrayAggAccumulator` / `DistinctArrayAggAccumulator` + `SingleRowListArrayBuilder`) produce arrays whose nested fields are all nullable. So for nested types the declared and produced types disagree with each other, not just with Spark.
### To Reproduce
In DataFusion Comet, which maps Spark's `collect_set` to `SparkCollectSet`, the produced `List(Field { data_type: Int32, nullable: true })` does not match the Catalyst-derived schema `List(Field { data_type: Int32 })`, and a compensating cast has to be inserted per batch. Observed for every element type tested (Int8/16/32/64, Boolean, Utf8, Binary, Decimal128, Date32, Timestamp):
```
WARN shuffle/src/schema_align.rs: ShuffleWriter input schema mismatch on col[1]
'sort_array(collect_set(i), true)': child produced List(Field { data_type: Int32, nullable: true }),
catalyst declared List(Field { data_type: Int32 })
```
The nested-type disagreement between `return_type` and the accumulator output surfaces as `Invalid argument error: column types must match schema types` from `AggregateExec` output validation; Comet currently works around it by casting the input to an all-nullable variant before the aggregate (`coerce_collect_child_nullability` in its planner).
### Expected behavior
- `return_type` / `return_field` declares a non-nullable list element for both functions.
- The accumulator output arrays carry a matching field, including for nested inputs, so no compensating cast is needed.
### Additional context
Tracked on the Comet side in apache/datafusion-comet#4515 (a collection of functions whose Arrow return type drifts from Spark Catalyst's declared type). Sibling example already fixed the same way: #22602 (`width_bucket` returning `Int32` instead of `Int64`).
Contributor guide
Research direction
Read datafusion/spark/src/function/aggregate/collect.rs, starting with SparkCollectList and SparkCollectSet return_type/return_field implementations. Trace ArrayAggAccumulator, DistinctArrayAggAccumulator, and SingleRowListArrayBuilder to compare declared and produced nullability, including nested inputs. Done means both declarations and accumulator arrays consistently use non-nullable elements without schema mismatches.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust
- Domain
- data-engineering
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Active
- Clarity
- Mostly clear
- Newbie friendliness
- 68/100