apache / apache/datafusion

bug: `SparkCollectList`/`SparkCollectSet` declare nullable list elements, diverging from Spark's `containsNull = false`

Open
#24,741 2 comments 0 reactions 0 assignees View on GitHub
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.