zhuqi-lucas opened a new issue, #23601: URL: https://github.com/apache/datafusion/issues/23601
Part of https://github.com/apache/datafusion/issues/23600. ## Is your feature request related to a problem or challenge? `first_value(x ORDER BY o)` and `last_value(x ORDER BY o)` fall back to the per-group `Accumulator` path whenever `x` is a nested type. `groups_accumulator_supported()` in [`first_last.rs`](https://github.com/apache/datafusion/blob/main/datafusion/functions-aggregate/src/first_last.rs) whitelists only scalar primitives (Int, Float, Decimal, Timestamp, Utf8, Binary). The per-group `Accumulator` stores state as `Vec<ScalarValue>`. For `List<Struct<...>>` payloads a `ScalarValue::List` is a heap-allocated deep clone. `update_batch` clones on every candidate row and compares against stored best. Cost scales as `#rows × #groups × sizeof(wide_ScalarValue)`, which explodes on wide payload. Two observable consequences: 1. `SELECT DISTINCT ON (id) * ORDER BY ts LIMIT 10` runs at 1.4 GB peak and is ~4× slower than `SELECT DISTINCT *` on the same dataset (#16620). 2. A production batch pipeline running `SELECT ..., FIRST_VALUE(list_col ORDER BY o DESC), ... GROUP BY p` over 11M rows × ~2.5 KB payload hits an OS memory kill at ~132 GB. The equivalent hand-written two-step (`MAX GROUP BY p` + JOIN + `FIRST_VALUE` **without** `ORDER BY`) finishes at ~3.6 GB peak. The gap is entirely the slow `Accumulator` path. ## Describe the solution you'd like Extend `groups_accumulator_supported()` for `first_value` / `last_value` to accept nested types: - `List(_)`, `LargeList(_)`, `ListView(_)`, `LargeListView(_)` - `FixedSizeList(_, _)` - `Struct(_)` - `Map(_, _)` Implement a columnar `GroupsAccumulator` state: - One `ArrayRef` per accumulator expression, indexed by group index — holds the current winner value per group - One or more `ArrayRef` for the ordering columns (also indexed by group index) - `update_batch(values, group_indices, ...)`: 1. Compare incoming ordering values against stored per-group ordering (vectorized) 2. Build an indices array of winning positions 3. Use `arrow::compute::take` to gather values from the input arrays into the new state array (scatter into group slots) - `merge_batch` follows the same pattern for cross-partition merges - `state()` / `evaluate()` return the accumulated `ArrayRef` directly, no `ScalarValue` conversion Expected memory: `#groups × wide_row × #expressions` — for the production case, ~125 MB instead of ~132 GB. ## Describe alternatives you've considered - **Keep the `Accumulator` path but cache and reuse `ScalarValue` allocations.** Reduces churn but still `O(#rows × sizeof(wide))` allocation work, and cannot reach vectorized update throughput. - **Force users to hand-write a two-step aggregate + JOIN.** Works today (production has done it) but is not discoverable and does not generalize to `DISTINCT ON` or `ROW_NUMBER = 1` patterns. ## Additional context Companion epic: https://github.com/apache/datafusion/issues/23600 Companion sub-issues (this ticket ships value on its own; the rest amplify): - Coalesce peer `FIRST_VALUE(... ORDER BY o)` expressions into a single struct accumulator - Logical rewrite `Filter(row_number() = 1) → Aggregate(FIRST_VALUE(... ORDER BY o))` — depends on this ticket to emit the fast path Related: - #16620 — the `DISTINCT ON` performance report this ticket directly addresses - #12252 — `max_by` (related aggregate primitive, already merged) -- 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]
