You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
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 10 runs at 1.4 GB peak and is ~4× slower than SELECT DISTINCT * on the same dataset (Performance of distinct on (columns) #16620).
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_VALUEwithoutORDER 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:
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, ...):
Compare incoming ordering values against stored per-group ordering (vectorized)
Build an indices array of winning positions
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.
Part of #23600.
Is your feature request related to a problem or challenge?
first_value(x ORDER BY o)andlast_value(x ORDER BY o)fall back to the per-groupAccumulatorpath wheneverxis a nested type.groups_accumulator_supported()infirst_last.rswhitelists only scalar primitives (Int, Float, Decimal, Timestamp, Utf8, Binary).The per-group
Accumulatorstores state asVec<ScalarValue>. ForList<Struct<...>>payloads aScalarValue::Listis a heap-allocated deep clone.update_batchclones 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).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()forfirst_value/last_valueto accept nested types:List(_),LargeList(_),ListView(_),LargeListView(_)FixedSizeList(_, _)Struct(_)Map(_, _)Implement a columnar
GroupsAccumulatorstate:ArrayRefper accumulator expression, indexed by group index — holds the current winner value per groupArrayReffor the ordering columns (also indexed by group index)update_batch(values, group_indices, ...):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, noScalarValueconversionExpected memory:
#groups × wide_row × #expressions— for the production case, ~125 MB instead of ~132 GB.Describe alternatives you've considered
Accumulatorpath but cache and reuseScalarValueallocations. Reduces churn but stillO(#rows × sizeof(wide))allocation work, and cannot reach vectorized update throughput.DISTINCT ONorROW_NUMBER = 1patterns.Additional context
Companion epic: #23600
Companion sub-issues (this ticket ships value on its own; the rest amplify):
FIRST_VALUE(... ORDER BY o)expressions into a single struct accumulatorFilter(row_number() = 1) → Aggregate(FIRST_VALUE(... ORDER BY o))— depends on this ticket to emit the fast pathRelated:
distinct on (columns)#16620 — theDISTINCT ONperformance report this ticket directly addressesmax_byin Aggregation function #12252 —max_by(related aggregate primitive, already merged)