perf: parallelize high-cardinality mixed DISTINCT aggregation

[XuPeng-SH]

What changed

Fixes #27672.

This change removes the serial high-NDV exact-state bottleneck exposed by ClickBench Q10 while preserving exact COUNT(DISTINCT) semantics.

• Teach group-shuffle planning to account for mergeable COUNT(DISTINCT) argument state, not only the small final group hash table. For a large exact state with enough logical groups, rows are partitioned by the group key before final aggregation, so each logical group has one owner and no exact-state MergeGroup is required.
• Estimate the DISTINCT state and effective group owners after selection. Low state/input ratios, too few surviving owners, unknown statistics, and unsupported keys stay on the existing topology.
• Explicitly disable shuffle for aggregate nodes that contain a non-mergeable DISTINCT state and are forced to one CN, including stale/reused shuffle state.
• At compile time, skip a normal group shuffle when the scheduled topology exposes only one physical aggregate owner. Single-CN max_dop=1 reuses the ordinary non-shuffle topology; multi-CN DOP 1 and reused shuffles retain their parallel layouts.
• Relocate arena-backed skiplists during preflight capacity growth instead of rebuilding every retained node, and grow arenas geometrically.
• Clamp speculative geometric growth to MPool's real single-allocation limit before allocation. If account or pool capacity rejects the preferred arena while the old arena is live, retry the smaller former chunked-growth capacity. Terminal ownership/invariant errors are not retried.
• Use one monotonic target iterator and one cached Inserter for sorted state merges instead of restarting target searches for every candidate.

The independent cached-Inserter correctness fix and batch-fill preflight optimization are owned by #27688 and are already in main.

Root cause

The single-aggregate DISTINCT rewrite does not apply to Q10 because the node also contains SUM, COUNT, and AVG. Exact COUNT(DISTINCT) states therefore remain inside each local Group operator. The planner costed shuffle from roughly 100 final RegionID groups and ignored the roughly one-million retained UserID values, so many partial ordered states converged on a serial MergeGroup. Fixed-size arena growth then repeatedly rebuilt historical state, while preflight and publication restarted ordered searches.

Two boundary regressions found during review are also closed:

• a geometric request could exceed MPool's roughly 2 GiB single-allocation ceiling and fail around a 1 GiB current arena even though the linear fallback still fit;
• a single-CN, single-owner plan could still compile a one-bucket Shuffle plus Dispatch, adding per-row hashing and pipeline overhead without exposing parallel aggregate ownership.

Design and boundaries

DuckDB uses a separate radix-partitioned table keyed by group keys plus DISTINCT arguments, then finalizes those unique tuples into the regular aggregate table in parallel. This PR adopts the same partition-before-finalize principle through MatrixOne's existing shuffle-group topology; it does not introduce a second aggregate framework.

The planner rule is deliberately bounded:

• only mergeable COUNT(DISTINCT) contributes exact-state cost;
• table/expression NDV is capped by input rows and conservatively adjusted by available selection statistics;
• low exact-state/input ratios do not trigger a full-row shuffle;
• fewer than 64 estimated logical owners do not trigger;
• existing supported integer/string shuffle keys are required;
• non-mergeable DISTINCT aggregates retain the single-stage contract and cannot retain stale shuffle state;
• a normal shuffle must expose at least two physical aggregate owners after DOP and CN placement are known.

Extreme single-group or strongly skewed distributions cannot be solved by group-key shuffle and remain on the optimized executor merge path. A full independent radix DISTINCT pipeline could cover those shapes, but needs a separate design-reviewed change and better conditional/top-frequency statistics.

Performance and memory

BenchmarkCountDistinctSavedArgumentMerge models 1,000,000 unique values, 16 partial states, and 100 groups on the same Apple M4 host:

• before this PR: 1.632 s/op;
• latest head, 5 runs: 146.6-153.5 ms/op, median 148.1 ms/op;
• improvement: about 10.6x-11.1x;
• latest allocation sample: 21-23 KiB/op and 4,813-4,819 allocs/op.

The allocator ceiling is checked only when an arena actually needs relocation, not per row or per key. Successful growth still uses the largest admissible geometric capacity; the smaller retry remains limited to classified account/pool capacity pressure.

Validation

• full UT: pkg/common/mpool, pkg/common/arenaskl, pkg/sql/colexec/aggexec, pkg/sql/colexec/group, pkg/sql/plan, and pkg/sql/compile;
• full race: pkg/common/arenaskl, pkg/sql/colexec/aggexec, pkg/sql/colexec/group, and pkg/sql/plan;
• 100 race repetitions of arena relocation, account-pressure and allocator-ceiling fallback, and each new single-owner/multi-CN compiler topology counterexample;
• go vet on the affected/consumer packages with the repository CGo environment;
• git diff --check.

The tests cover forward/reverse arena links, values, duplicate rejection, post-growth insertion, poisoned/overlapping arenas, transactional allocation failure, account pressure, allocator ceiling, filtered NDV, insufficient logical owners, low state/input ratio, invalid statistics, stale single-stage shuffle state, single-owner single/multiple inputs, ordered aggregates, and multi-CN DOP 1. No sleeps, large-data UTs, or weakened assertions were added.

[qodo-code-review]

Qodo reviews are paused for this user.

Troubleshooting steps vary by plan Learn more →

On a Teams plan?
Reviews resume once this user has a paid seat and their Git account is linked in Qodo.
Link Git account →

Using GitHub Enterprise Server, GitLab Self-Managed, or Bitbucket Data Center?
These require an Enterprise plan - Contact us
Contact us →

[LeftHandCold]

Reviewed the exact head b1118eb. Checked arena relocation and link preservation, cached Inserter merge behavior, target/source DISTINCT deduplication, COUNT(DISTINCT) shuffle eligibility and single-stage safeguards. Focused arenaskl tests and diff-check passed; no reachable correctness or normal-aggregation regression blocker found. The remaining Ubuntu/SCA/BVT checks were still running at review time.

[aptend]

Reviewed exact head 0a232e7, including the full diff, prior review, arena/accounting failure paths, DISTINCT merge state, planner decisions, and compile topology. Build, vet, owning-package tests, targeted race repetitions, and the new benchmark pass locally. One grouping-sets correctness blocker remains.

[aptend]

[P1] Exclude inactive grouping-set keys from this shuffle path. For a grouping-set branch, GroupingFlag[i] == false means Group replaces that key with the rollup constant, but constructShuffleArgForGroup hashes the raw child column before Group performs that replacement. The new distinctStateShuffle threshold can therefore enable a topology that was previously rejected. For example: SELECT region_id, SUM(x), COUNT(DISTINCT user_id) FROM t GROUP BY GROUPING SETS ((region_id), ()) with 1M rows, about 100 regions, and high user_id NDV. In the () branch the planner chooses raw region_id, distributes one logical grand-total group across owners, then every owner changes the key to the same rollup constant locally; there is no downstream MergeGroup, so the query can emit duplicate grand-total rows with partial SUM/COUNT results. A direct counterexample in TestDetermineShuffleForGroupByAccountsForCountDistinctState using GroupingFlag = []bool{false} currently observes Shuffle == true. Please select only active grouping keys here (or disable the path for inactive grouping-set branches) and cover the empty grouping set.

[XuPeng-SH]

Fixed in 19ee024. determineShuffleForGroupBy now selects the distribution key only from active GroupingFlag entries; an empty grouping set with no active key leaves shuffle disabled, while a mixed branch still shuffles on its active key. Added deterministic coverage for both cases in TestDetermineShuffleForGroupByAccountsForCountDistinctState. Planner owning-package and compile shuffle consumer tests pass.

[XuPeng-SH]

Deep review completed on exact head 0a232e7. No blocking correctness, resource-lifecycle, performance, or unhappy-path issue found. Planner NDV/state guards, single-owner topology handling, arena relocation, allocator-pressure fallback, and DISTINCT deduplication are covered. Focused arenaskl, mpool, planner, compiler, and aggexec tests passed. The pr-size-label failure is an unrelated GitHub API TLS failure.

[aptend]

Deep re-review of exact head 19ee0241dd22453de010225225d937988d642648 against merge-base ff4270c844c4b630cf1d921da813ba119d5b5e89. I re-read the complete diff, all reviews/inline comments/replies, and the remaining unresolved thread, then checked the one-commit delta from my prior reviewed head 0a232e7492b040b1984b25caf60f7af49c3e15a9. The grouping-sets blocker is closed: inactive keys are excluded before distribution-key selection, an empty grouping set keeps the merge topology, and a mixed branch can select only an active key. Owning-package list/build/vet/full tests, 50 race repetitions across arena/preflight/planner/compile paths, and the relevant benchmarks pass. No blocking issue found.

[aunjgr]

Approved at exact head 19ee0241dd22453de010225225d937988d642648 against base ff4270c844c4b630cf1d921da813ba119d5b5e89.

The planner now costs mergeable COUNT(DISTINCT) exact state separately from final group cardinality, applies conservative filtered-NDV/state-ratio/owner/key guards, clears stale shuffle state for non-mergeable DISTINCT aggregates, and avoids a one-owner shuffle at compile time. Shuffling by an active logical group key gives every exact state one owner and preserves results; inactive grouping-set keys and empty grouping sets correctly retain MergeGroup.

Arena growth allocates the replacement under the existing account while the old arena remains charged, clamps speculative capacity to MPool’s real single-allocation ceiling, retries only classified capacity pressure at the linear fallback, copies the valid arena prefix, rebinds relative links, and frees the old owner only after publication. Failure leaves the old list unchanged. Cached target iteration/insertion remains monotonic within each sorted source group and resets if an unexpected rebuild occurs. The tests cover forward/reverse links, duplicates, allocation rollback, ceiling fallback, grouping sets, stale topology, single/multi-CN ownership, and race/reuse paths. No correctness, leak, hang, or material regression blocker found.