Repository navigation
Amortize state compaction in DistinctArrayAggGroupsAccumulator on EmitTo::First #26512
Description
Is your feature request related to a problem or challenge?
DistinctArrayAggGroupsAccumulator (added in #26357) rebuilds its remaining state on every EmitTo::First(n): retain_after copies the remaining encoded values, rehashes them into value_map, and rebuilds pair_set. The cost is proportional to the remaining state, so emitting in k chunks does k rebuilds.
The distinct_array_agg groups benchmark in datafusion/functions-aggregate/benches/array_agg.rs shows it with few groups and high distinct cardinality (one 8192-row Utf8 batch):
| case | groups accumulator | adapter |
|---|---|---|
| 100 groups, ~7800 distinct, emit all | 0.37 ms | 0.97 ms |
| 100 groups, ~7800 distinct, emit in 8 chunks | 1.55 ms | 0.98 ms |
Describe the solution you'd like
Make the compaction amortized: on EmitTo::First(n) record a group offset and drop the emitted pairs lazily, and only rebuild values, value_map and pair_set once at least half of the state belongs to emitted groups. Repeated partial emission then costs O(total state) overall instead of O(k × state).
Describe alternatives you've considered
Leave it: the hash aggregation path emits with EmitTo::All and slices, so First(n) mainly comes from early emission on ordered input, where few groups remain.
Additional context
Follow-up to #26357, raised in review there.
Activity