zhuqi-lucas opened a new pull request, #23628: URL: https://github.com/apache/datafusion/pull/23628
## Which issue does this PR close? Closes #23601. Part of the epic #23600. ## Rationale for this change `first_value(x ORDER BY o)` / `last_value(x ORDER BY o)` currently fall back to the per-group `Accumulator` path whenever `x` is a nested type (`List`, `LargeList`, `Struct`, `Map`, ...), because [`groups_accumulator_supported()`](https://github.com/apache/datafusion/blob/main/datafusion/functions-aggregate/src/first_last.rs) whitelists only scalar primitives and byte types. Two consequences we hit: - The per-group `Accumulator` path calls `ScalarValue::try_from_array` on every candidate row, and stores `ScalarValue::List` (or `ScalarValue::Struct`, ...) as an `Arc` slice into the source batch. For wide payloads this both allocates heavily per row and *pins the source batch* in memory for every group whose current winner came from it. On high-cardinality dedup queries (`SELECT DISTINCT ON`, `ROW_NUMBER() = 1`, single-pass `FIRST_VALUE(...ORDER BY)` over a wide `List<Struct>`) that combination can OOM even when the final output would fit comfortably. - This is also what #16620 (`Performance of DISTINCT ON (columns)`) reports at the SQL layer. ## What changes are included in this PR? Adds a new `GenericValueState` in `first_last/state.rs` — a `Vec<Option<ScalarValue>>`-backed `ValueState` — and wires it into `create_groups_accumulator` for nested types. The state uses `ScalarValue::compact()` after `try_from_array` so the stored winner is an owned copy instead of an `Arc` slice into the source batch (otherwise the fast path would still pin source batches even though the accumulator's own reported size looks small). Types now supported by the `GroupsAccumulator` fast path: - `List`, `LargeList`, `ListView`, `LargeListView` - `FixedSizeList` - `Struct` - `Map` The existing `PrimitiveValueState` (scalar primitives) and `BytesValueState` (Utf8 / Binary variants) paths are unchanged. ## Are these changes tested? Yes. - 9 unit tests in `first_last/state.rs` covering `List<Utf8>`, `Struct`, `LargeList`, `FixedSizeList`, `Map`, `EmitTo::First(n)` partial emit, null handling, size accounting on overwrite and shrink, and a dedicated regression test (`test_generic_value_state_compact_releases_parent_batch`) that verifies `compact()` releases the `Arc` reference to the source batch. - 2 integration tests in `first_last.rs` exercising the full `FirstLastGroupsAccumulator<GenericValueState>` for `List<Int32>`: one functional multi-batch test, and one that asserts the accumulator's reported `size()` stays proportional to `#groups` rather than `#rows` on a 10-group × 10 000-row workload. `cargo fmt` and `cargo clippy -p datafusion-functions-aggregate --lib --tests -- -D warnings` are clean. ## Are there any user-facing changes? Yes, but only in the sense that a previously-slow / OOM-prone path becomes fast for the affected data types. No SQL or API surface changes; the same queries run without modification and previously-passing tests continue to pass. ## Follow-ups - Add a benchmark comparing the two code paths on wide-payload aggregates so the memory improvement is measurable in-tree. - The two remaining sub-issues in the epic build on this one: - #23602 — coalesce peer `FIRST_VALUE(... ORDER BY o)` expressions into a single struct accumulator. - #23603 — logical rewrite `Filter(row_number() = 1) → Aggregate(FIRST_VALUE(... ORDER BY o))`. Depends on this PR to emit the fast 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]
