zhuqi-lucas opened a new pull request, #26357: URL: https://github.com/apache/datafusion/pull/26357
## Which issue does this PR close? - Closes #26347. ## Rationale for this change `array_agg(DISTINCT x)` had no `GroupsAccumulator`. Grouped queries fell back to `GroupsAccumulatorAdapter`, which keeps one `DistinctArrayAggAccumulator` per 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`: * Each distinct value is encoded once with a `RowConverter` into a `Rows` buffer shared by all groups, so a value that appears in many groups is stored once. * Groups record `(group_index, value_index)` pairs, deduplicated in one hash set. * `evaluate` builds the `ListArray` with a counting sort by group. * It also covers `array_agg(DISTINCT x ORDER BY x)`, sorting each group by the encoded rows. * `EmitTo::First` drops 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-list `convert_to_state` is shared with `ArrayAggGroupsAccumulator`. `state_compat` now 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? * **Unit tests** cover dedup across batches, nulls with and without IGNORE NULLS, FILTER, ORDER BY DESC, `EmitTo::First`, state/merge, `convert_to_state`, dictionary types, and a differential check against `DistinctArrayAggAccumulator`. * **`state_compat`** passes with the new accumulator included. * **`array_agg.slt`** gains grouped cases mixing DISTINCT with `count(*)`, 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_batch` plus `evaluate`, the new accumulator against `GroupsAccumulatorAdapter` over `DistinctArrayAggAccumulator`. Apple M-series: | groups | input | groups accumulator | adapter (before) | speedup | | -- | -- | -- | -- | -- | | 100 | ~25 distinct | 154 µs | 251 µs | 1.6x | | 100 | ~7800 distinct | 384 µs | 969 µs | 2.5x | | 1000 | ~25 distinct | 173 µs | 1.42 ms | 8.2x | | 1000 | ~7800 distinct | 382 µs | 2.26 ms | 5.9x | | 8192 | ~25 distinct | 279 µs | 10.2 ms | 36x | | 8192 | ~7800 distinct | 414 µs | 10.4 ms | 25x | 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. | groups | input | groups accumulator | adapter (before) | | -- | -- | -- | -- | | 100 | ~25 distinct | 11 KB | 720 KB | | 1000 | ~7800 distinct | 2.2 MB | 9.9 MB | | 8192 | ~25 distinct | 1.7 MB | 18.3 MB | | 65536 | ~7800 distinct | 2.2 MB | 100.7 MB | 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. -- 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]
