goutamadwant commented on PR #24767:
URL: https://github.com/apache/datafusion/pull/24767#issuecomment-5470753487
> ## P1
> ### P1-1 - Output array column stays `nullable`, Spark declares it
non-nullable
> **Where:** `SparkCollectList` / `SparkCollectSet` in `collect.rs` do not
override `is_nullable()`.
>
> **Problem:** `AggregateUDFImpl::is_nullable()` defaults to `true`
(`datafusion/expr/src/udaf.rs:547`), and `udaf_default_return_field`
(`udaf.rs:1201`) stamps the output field's nullability from it. So both
aggregates report an output field `nullable: true`. Spark's `Collect.nullable`
is `false` (`collect.scala:49`); the array is never null. The PR fixed the
element axis (`containsNull=false`) but left the array axis, so the Spark
schema fix is half-applied.
>
> **Why it matters:** Divergence from the Spark schema this PR exists to
match. It is a schema/nullability contract mismatch for the Spark-compat layer,
not a wrong-value bug (the value is never null), so down-rank to P2 if the
consumer (Comet) does not validate field nullability. `verdict: CONFIRMED` on
the divergence; `PLAUSIBLE` on a downstream schema-check failure.
>
> **Fix:** add `fn is_nullable(&self) -> bool { false }` to both impls. Safe
and precedented: `default_value` is already a non-null empty list
(`empty_list_scalar`), which satisfies the `is_nullable=false` contract noted
at `udaf.rs:546`. Precedent: `count.rs:296`, `approx_distinct.rs:723`,
`regr.rs:467`. Assert it in `collect_types_have_non_nullable_elements`
(`aggregate.is_nullable()` is not currently checked).
>
> ## P2
> ### P2-1 - No `.slt` for the schema change; existing `collect.slt` is
value-only
> **Where:**
`datafusion/sqllogictest/test_files/spark/aggregate/collect.slt` exists but
every query is `query ?` / `query I?` (values only). The PR adds Rust unit
tests, no `.slt`.
>
> **Problem:** The PR's entire behavioral change (element
`containsNull=false`) is SQL-visible via `arrow_typeof`, but no SQL-level test
asserts it, and the existing SLT cannot catch it (unchanged printed values).
Checklist item 8 and Testing line 100: SQL-visible changes need an `.slt`; Rust
unit tests bypass the planner and type plumbing.
>
> **Fix:** extend `collect.slt` with element-type assertions, e.g. `SELECT
arrow_typeof(collect_list(a)) FROM (VALUES (1)) t(a);` and the `collect_set`
variant. `arrow_typeof` renders the element field (`List(Field { ... nullable:
false ... })`), so it pins the change. The outer `nullable=false` (P1-1) is not
visible via `arrow_typeof`; cover that in the Rust test instead.
>
> ### P2-2 - `normalize_list_scalar` null-list branch is untested
> **Where:** `collect.rs`, the `array.is_null(0)` branch (new lines ~55-61,
`new_null_list`).
>
> **Problem:** This branch is only reachable through `state()` on an
all-null / empty group (`ArrayAggAccumulator::state()` -> `evaluate()` ->
`new_null_list(..., true, 1)`, then normalized). It is real partial-aggregation
behavior that flows into shuffle and `merge_batch`.
`empty_results_have_non_nullable_elements` calls only `evaluate()` (which
bypasses normalize via `empty_list_scalar`), and
`accumulator_state_and_output_preserve_nested_type` calls `state()` only after
a non-null `update_batch`. The branch has no coverage.
>
> **Fix:** add a test calling `state()` on a fresh (or all-null-updated)
accumulator; assert one null list whose element field is non-nullable and named
`item`, for both distinct and non-distinct.
>
> ### P2-3 - New `cast` can turn a previously-working query into a runtime
error (regression risk)
> **Where:** `collect.rs`, `normalize_list_scalar` cast (new lines ~64-68):
`cast(values.as_ref(), field.data_type())`.
>
> **Problem:** For nested elements (e.g. structs), when the runtime array's
nested nullability is looser than the declared type (`arg_types[0]`) and the
runtime data actually contains nulls under a field the declared type marks
non-nullable, `cast` -> `cast_struct_to_struct` -> `StructArray::try_new`
errors ("Found unmasked nulls for non-nullable field",
`../arrow-rs/arrow-array/src/array/struct_array.rs:169`). Before this PR
(nullable element, no normalize) the value passed through. DataFusion
nullability inference is imprecise across joins, so this can regress a real
query. It fails safe (errors, no corruption) and only when `values.data_type()
!= field.data_type()`, so primitives and matching structs are unaffected.
`verdict: PLAUSIBLE`.
>
> **Fix / coverage:** add a `collect_list` test over a struct whose nested
field is declared non-nullable but whose runtime input carries a null there,
and decide the intended behavior (error vs. widen).
`accumulator_state_and_output_preserve_nested_type` differs in nullability but
uses no actual nulls, so it does not exercise this path.
Addressed in `ddbadd3`.
- Both `collect_list` and `collect_set` now report a non-nullable outer
result field, with `return_field` coverage.
- `collect.slt` now pins `List(non-null Int64)` through `arrow_typeof` for
both aggregates.
- Empty partial state is covered for distinct and non-distinct paths,
including state merge to the final empty non-null result.
- Runtime input whose metadata is wider but whose data is valid is
normalized to the declared type. An actual unmasked null under a declared
non-nullable nested field is rejected fallibly at the accumulator boundary for
both aggregates rather than widening the output schema or allowing
`collect_set` to panic. Widening would recreate the `AggregateExec` schema
mismatch this PR fixes.
Mutation checks confirmed each new test fails when its corresponding
behavior is disabled. Please let me know if you ahve any other suggestions.
thanks!
--
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]
