[Enhancement]: Add adaptive partition-parallel execution for exact DISTINCT aggregates

[XuPeng-SH]

Motivation

Exact DISTINCT aggregation currently derives most of its parallelism from ownership of the final GROUP BY key. This works well when there are enough balanced groups, but it does not provide robust parallelism when the final group cardinality is small or skewed.

Examples:

SELECT COUNT(DISTINCT user_id) FROM events;

SELECT region_id, COUNT(DISTINCT user_id)
FROM events
GROUP BY region_id;

In the first query there is only one final group. In the second query, a few regions or one hot region can own most rows. Hashing only by region_id therefore limits useful aggregate parallelism to the number and balance of regions, regardless of DOP. Increasing DOP alone cannot remove that bottleneck.

The current exact argument state also uses an ordered per-group skiplist. Admission requires key encoding, comparisons, pointer traversal, and retained-key storage. Partial-state merging can add another pass over those keys. These costs grow with DISTINCT NDV even when the final number of groups remains small.

The optimization in #27693 improves the balanced, many-group path by assigning each complete group to one owner and avoiding an expensive downstream exact-state merge. This issue covers the complementary execution shapes that cannot obtain enough parallelism from the final group key alone.

Required invariant

Exact DISTINCT work must be distributable by the identity being deduplicated, not only by the final aggregate group:

canonical key = (active GROUP BY keys, DISTINCT arguments)

Every equal canonical key must reach exactly one final deduplication owner, while different keys—even within one hot aggregate group—may be processed concurrently. Final aggregate results must remain bit-for-bit equivalent to the existing exact semantics.

Proposed execution model

Introduce an adaptive partition-parallel path for mergeable exact DISTINCT aggregates:

1. Project and encode the canonical (group keys, DISTINCT arguments) key once per input row or vector batch.
2. Compute its stable hash once and carry/reuse it across local admission, exchange, and final partitioning.
3. Perform optional thread-local batch/vectorized deduplication.
4. Partition surviving keys by the canonical-key hash, rather than only by the final group key.
5. Finalize partitions concurrently; each partition performs exact deduplication independently.
6. Feed unique (group keys, DISTINCT arguments) rows into the ordinary final aggregate by group key.

The implementation should reuse existing pipeline, shuffle, hash-table, and spill ownership primitives where their contracts fit. A new framework is justified only if those contracts cannot express independently owned DISTINCT partitions with lower total complexity.

Adaptive path selection

No single topology is optimal for every data shape. Planning and/or bounded runtime sampling should select among:

• Small input or low NDV: keep the simple local path and avoid partition machinery.
• Enough balanced final groups: keep the group-owner fast path introduced by perf: parallelize high-cardinality mixed DISTINCT aggregation #27693.
• One/few groups or high group skew: partition by (group keys, DISTINCT arguments) to expose key-level parallelism.
• High duplicate ratio: deduplicate locally before exchange to reduce CPU and network bytes.
• Nearly all keys unique: avoid repeated local hash probes that cannot remove meaningful input; append/partition directly and deduplicate during partition finalization.

Selection must be based on general cardinality, skew, memory, and ownership estimates—not query text, benchmark names, fixed schemas, or issue-specific constants. Missing or unreliable statistics require a conservative fallback.

Correctness and ownership requirements

The design document must be reviewed before implementation and define:

• canonical equality and hashing for fixed/variable-width values, multiple DISTINCT arguments, NULL, constants, decimals, collations, floating signed zero/NaN behavior, and hash collisions;
• applicability to COUNT(DISTINCT) and the explicit boundary for other DISTINCT aggregates;
• sharing of one deduplication pipeline by aggregates with identical DISTINCT arguments and compatible filters;
• behavior with aggregate filters, grouping sets, empty input, one/multiple CNs, DOP=1, cancellation, errors, Reset/Free, and operator reuse;
• ownership and lifecycle of local tables, partition buffers, exchange batches, final tables, and any spill state;
• bounded memory accounting and integration with Support bounded distinct-key spill for high-NDV exact DISTINCT aggregates #27698, without creating a second incompatible spill mechanism;
• prevention of duplicate hash computation, unbounded per-row allocation, per-key goroutines, locks, or high-cardinality metrics;
• planner cost inputs, including effective owner count and skew—not only global argument NDV or final group count.

Performance requirements

• A single-group high-NDV COUNT(DISTINCT) must use multiple workers when DOP > 1.
• Few-group and hot-group workloads must scale through canonical-key partitions rather than being capped by final group count.
• The existing balanced many-group Q10 path covered by perf: parallelize high-cardinality mixed DISTINCT aggregation #27693 must not regress materially; the adaptive selector should retain that topology when it is cheaper.
• Local pre-deduplication must reduce exchanged rows/bytes when duplicates are common and avoid material overhead when keys are nearly unique.
• Common fixed-width keys should use a compact vectorized/hash-table path; an ordered skiplist should not remain mandatory when aggregate semantics require only equality.
• CPU, allocations, peak accounted memory, exchanged rows/bytes, and partition skew must be measured before and after. Wall-clock time alone is insufficient.

Validation matrix

Cover independent dimensions rather than one benchmark-shaped case:

Dimension	Required cases
Final groups	global/one, few, 64+, high cardinality
Distribution	uniform, one hot group, Zipf-like skew
DISTINCT NDV	low, medium, near input cardinality
Duplicate ratio	high duplicate, mixed, nearly all unique
Key shape	fixed width, VARCHAR, multi-column, NULL-containing
Execution	DOP=1 and multi-DOP; one CN and multi-CN
Memory	no spill and forced bounded-memory spill via injection
Lifecycle	success, error, cancellation, Reset/Free, repeated reuse

Correctness tests must compare with an independent exact oracle and include deliberate hash collisions. Plan-shape assertions may supplement but cannot replace result assertions.

Unit tests must use small deterministic data and injected thresholds. Do not add sleeps, large-data UTs, or timing-only assertions. Put larger scaling and performance measurements in a dedicated benchmark/integration job.

Acceptance criteria

1. Exact results match the independent oracle across the full validation matrix.
2. A global or single-hot-group high-NDV case demonstrates real multi-worker partition finalization and meaningful speedup over DOP=1 on the same hardware.
3. The balanced many-group control does not regress materially relative to perf: parallelize high-cardinality mixed DISTINCT aggregation #27693.
4. Identical DISTINCT argument sets are deduplicated once and safely shared where semantics permit.
5. Peak accounted memory respects the configured bound and the spill lifecycle satisfies Support bounded distinct-key spill for high-NDV exact DISTINCT aggregates #27698's cleanup and failure-path requirements.
6. Runtime path selection is observable through bounded counters/profile labels for selected topology, partition count/skew, input/surviving keys, exchanged bytes, and spill activity.
7. Focused benchmarks report CPU time, allocations, memory, network/shuffle volume, and wall time for every relevant topology.

Related

• [Tech Request]: Optimize aggregate functions with distinct keyword #9049 tracks the broader DISTINCT aggregate optimization direction.
• perf: parallelize high-cardinality mixed DISTINCT aggregation #27693 optimizes the balanced many-group mixed-aggregation path.
• Support bounded distinct-key spill for high-NDV exact DISTINCT aggregates #27698 tracks exact bounded-memory DISTINCT spill and must share a compatible partition ownership model with this work.

[XuPeng-SH]

#27753 tracks a possible generic transport/reuse contract for canonical grouping hashes. This issue retains ownership of the DISTINCT-specific canonical identity (active GROUP BY keys, DISTINCT arguments) and adaptive partitioning semantics. The two should share machinery only if their equality, lifecycle, and cost contracts are actually compatible; #27720 must not depend on carrying a hash when recomputation is cheaper.