Repository navigation
perf: add a GroupsAccumulator for array_agg(DISTINCT) - #26357
zhuqi-lucas wants to merge 3 commits into
Conversation
array_agg(DISTINCT x) had no GroupsAccumulator, so grouped queries went through GroupsAccumulatorAdapter with one DistinctArrayAggAccumulator (and its own hash table and encoded rows) per group. DistinctArrayAggGroupsAccumulator encodes each distinct value once with a RowConverter into a Rows buffer shared by all groups, and deduplicates (group_index, value_index) pairs in a single hash set. It also covers array_agg(DISTINCT x ORDER BY x), sorting each group by the encoded rows. state_compat now compares list results as multisets for DISTINCT without ORDER BY, where the element order is not defined. Closes apache#26347.
There was a problem hiding this comment.
🟡 Changes recommended
Missing DISTINCT instrumentation breaks existing metric expectations, and unnecessary encoding introduces CPU and memory regressions.
3 open findings
What changed in this PR
Adds a native grouped accumulator for array_agg(DISTINCT) to reduce per-group memory and CPU overhead.
Changes:
- Shares distinct-value storage across groups and supports ordering and partial emission.
- Reuses state-conversion helpers and adjusts unordered DISTINCT result comparisons.
- Adds unit and SQL coverage for grouped DISTINCT aggregation.
| File | Description |
|---|---|
| datafusion/sqllogictest/test_files/array_agg.slt | Adds grouped DISTINCT query coverage. |
| datafusion/functions-aggregate/tests/state_compat.rs | Compares unordered DISTINCT lists as multisets. |
| datafusion/functions-aggregate/src/array_agg.rs | Implements the native accumulator, shared helpers, and unit tests. |
🧠 Review effort: Balanced
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| impl GroupsAccumulator for DistinctArrayAggGroupsAccumulator { | ||
| fn update_batch( |
| if rows.is_empty() { | ||
| return Ok(()); | ||
| } | ||
| self.insert(input, rows) |
| if rows.is_empty() { | ||
| return Ok(()); | ||
| } | ||
| self.insert(lists.values(), rows) |
…DISTINCT) - Report the DISTINCT submetric from the native groups accumulator, and keep it across a full emit. - Gather filtered or null-skipped rows before encoding them, and encode only the offset span of a sliced state list when merging. - Build the adapter explicitly in the tests that cover its metrics, and give the spilling test enough groups to spill with the native accumulator.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #26357 +/- ##
========================================
Coverage 82.87% 82.88%
========================================
Files 1153 1153
Lines 457101 457617 +516
Branches 457101 457617 +516
========================================
+ Hits 378818 379278 +460
- Misses 55268 55273 +5
- Partials 23015 23066 +51 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
jayzhan211
left a comment
There was a problem hiding this comment.
Thanks @zhuqi-lucas , overall LGTM
| + self.values.as_ref().map_or(0, Rows::size) | ||
| + self.value_map_size | ||
| + self.pairs.capacity() * size_of::<(u32, u32)>() | ||
| + self.pair_set.capacity() * (size_of::<(u32, u32)>() + 1) |
There was a problem hiding this comment.
| + self.pair_set.capacity() * (size_of::<(u32, u32)>() + 1) | |
| + estimate_memory_size::<(u32, u32)>(self.pair_set.capacity(), 0).unwrap_or_default() |
There was a problem hiding this comment.
Should we use this one?

Which issue does this PR close?
Rationale for this change
array_agg(DISTINCT x)had noGroupsAccumulator. Grouped queries fell back toGroupsAccumulatorAdapter, which keeps oneDistinctArrayAggAccumulatorper group, each with its own hash table and encoded rows. With many groups this dominates memory and CPU.What changes are included in this PR?
New
DistinctArrayAggGroupsAccumulator:RowConverterinto aRowsbuffer shared by all groups, so a value that appears in many groups is stored once.(group_index, value_index)pairs, deduplicated in one hash set.evaluatebuilds theListArraywith a counting sort by group.array_agg(DISTINCT x ORDER BY x), sorting each group by the encoded rows.EmitTo::Firstdrops values only referenced by emitted groups.Small refactors: the DISTINCT / ORDER BY check is shared by the row and groups paths (
distinct_sort_options), and the one-row-per-listconvert_to_stateis shared withArrayAggGroupsAccumulator.state_compatnow compares list results as multisets for DISTINCT without ORDER BY, since the element order is not defined there. The row accumulator returns hash table order.Are these changes tested?
EmitTo::First, state/merge,convert_to_state, dictionary types, and a differential check againstDistinctArrayAggAccumulator.state_compatpasses with the new accumulator included.array_agg.sltgains grouped cases mixing DISTINCT withcount(*), FILTER, ORDER BY and IGNORE NULLS. The existing aggregate slt files pass.Benchmarks
Measured locally with a criterion benchmark (not included here, happy to add it): one 8192-row Utf8 batch,
update_batchplusevaluate, the new accumulator againstGroupsAccumulatorAdapteroverDistinctArrayAggAccumulator. Apple M-series:Live heap after
update_batch, measured with a counting global allocator: 65 536 Utf8 rows in 8 batches.size()tracks these within ~10% for both.Memory now grows with the number of distinct values instead of the number of groups.
Are there any user-facing changes?
No API changes.
array_agg(DISTINCT x)without ORDER BY may return elements in a different (still unspecified) order.