Repository navigation
Description
Part of #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 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:
SELECT DISTINCT ON (id) * ORDER BY ts LIMIT 10runs at 1.4 GB peak and is ~4× slower thanSELECT DISTINCT *on the same dataset (Performance ofdistinct on (columns)#16620).- A production batch pipeline running
SELECT ..., FIRST_VALUE(list_col ORDER BY o DESC), ... GROUP BY pover 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_VALUEwithoutORDER BY) finishes at ~3.6 GB peak. The gap is entirely the slowAccumulatorpath.
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
ArrayRefper accumulator expression, indexed by group index — holds the current winner value per group - One or more
ArrayReffor the ordering columns (also indexed by group index) update_batch(values, group_indices, ...):- Compare incoming ordering values against stored per-group ordering (vectorized)
- Build an indices array of winning positions
- Use
arrow::compute::taketo gather values from the input arrays into the new state array (scatter into group slots)
merge_batchfollows the same pattern for cross-partition mergesstate()/evaluate()return the accumulatedArrayRefdirectly, noScalarValueconversion
Expected memory: #groups × wide_row × #expressions — for the production case, ~125 MB instead of ~132 GB.
Describe alternatives you've considered
- Keep the
Accumulatorpath but cache and reuseScalarValueallocations. Reduces churn but stillO(#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 ONorROW_NUMBER = 1patterns.
Additional context
Companion epic: #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:
- Performance of
distinct on (columns)#16620 — theDISTINCT ONperformance report this ticket directly addresses - Support
max_byin Aggregation function #12252 —max_by(related aggregate primitive, already merged)
Activity
saadtajwar commentedon Jul 15, 2026
Hey @zhuqi-lucas , happy to take a stab at this one!
saadtajwar commentedon Jul 15, 2026
take
zhuqi-lucas commentedon Jul 16, 2026
Sorry @saadtajwar , i already in progress a PR, and it related our prod improvement.
saadtajwar commentedon Jul 16, 2026
Oh no worries at all!
saadtajwar commentedon Jul 16, 2026
untake
zhuqi-lucas commentedon Jul 16, 2026
@saadtajwar You can take this one #23603 which i not started.
saadtajwar commentedon Jul 16, 2026
Sweet, will do! Thank you! Looking forward to working together :)
feat(functions-aggregate): support nested types (List, Struct, Map) i…
feat(functions-aggregate): support nested types (List, Struct, Map) i…
feat(optimizer): logical rewrite Filter(row_number() = 1) over PARTIT…