[fix](local shuffle) Require hash input for distinct finalize agg without group keys - #66570
[fix](local shuffle) Require hash input for distinct finalize agg without group keys#66570924060929 wants to merge 1 commit into
Conversation
…roup keys The FE local-shuffle planner handed a NoRequire distribution to a finalize merge agg that has no group keys but DISTINCT aggregates (e.g. count(distinct k)). Unlike COUNT(*), such an agg emits per-instance scalar values that the parent sums (sum0(multi_distinct_count(...))), so its input must be hash-partitioned by the distinct key. When a PASSTHROUGH local exchange (e.g. broadcast-join probe fan-out) scatters same-key rows across instances, the parent double-counts overlapping keys — result = correct value × local task count. Mirror BE AggSinkOperatorX::update_operator's _partition_exprs (grouping exprs, or distinct/distribute exprs): aggs with a partition requirement must demand HASH from their child; only partition-less aggs (COUNT(*)-style) keep NoRequire. requiresShuffleForCorrectness now also covers DISTINCT aggregates to match BE is_shuffled_operator(). Regression tests: AggregationNode unit coverage for every phase/flag combination, plus a sql-level distributed-plan test asserting the LOCAL_HASH exchange appears below the distinct finalize agg.
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
morrySnow
changed the title
|
run buildall |
TPC-H: Total hot run time: 29178 ms |
TPC-DS: Total hot run time: 157704 ms |
ClickBench: Total hot run time: 23.68 s |
FE UT Coverage ReportIncrement line coverage |
FE Regression Coverage ReportIncrement line coverage |
|
/review |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
Review outcome
Changes requested. The keyed scalar DISTINCT fix is directionally sound, but the name-only partition predicate introduces a separate zero-key HASH path for direct multi-distinct calls, and the added SQL tests do not yet prove the intended keyed aggregate edge independently.
Findings
- [P1] Direct scalar multi_distinct_* partial aggregates can receive a zero-expression HASH exchange, collapsing all rows and distinct state onto one task per BE.
- [P2] The positive SQL regression checks only the HASH enum, not the affected aggregate edge or its distinct-key expressions.
- [P2] The no-force test neither resets/selects its intended session shape nor asserts the claimed no-redundant-exchange behavior.
Details and remediation are attached as inline comments.
Critical checkpoints
- Goal and test proof: The original wrong-result mechanism and the keyed final-multi-distinct path were traced end to end. The implementation fixes that path, but M1 exposes a direct-function regression; M2 and M3 leave the key placement and already-satisfied negative path unproven.
- Scope, conditions, and parallel paths: All three changed files were reviewed, plus Nereids aggregate splitting/property derivation, legacy translation, local-exchange insertion, join/serial/colocate/set-operation propagation, Thrift serialization, and BE aggregation/partitioner consumers. LOCAL, FIRST_MERGE, finalize, group-key, no-key, compatible-hash, and passthrough branches were checked.
- Concurrency and lifecycle: No production locking or shared-state lifecycle change is introduced. The test fixture does retain one ConnectContext across methods, and M3 identifies the unreset session-variable lifecycle defect.
- Configuration and dynamic behavior: No new configuration is added. Existing agg phase, local-exchange, broadcast force-passthrough, serial-source, and distinct-streaming settings were checked in their relevant branches.
- Compatibility and FE-BE contract: No persisted schema, public API, or Thrift field changes are present. Function-name detection agrees across FE and BE for the four registered multi-distinct functions; keyed paths align, while M1 diverges from BE base behavior when the effective key is empty.
- Test coverage and results: All 23 completed non-skipped PR checks are green, including FE UT, compilation, and CheckStyle. Per the review-only task constraint, no local build or test command was run. The remaining coverage defects are M2 and M3.
- Observability: This planner-only change adds no new runtime error path or operational state requiring logging/counters; existing plan/profile visibility remains unchanged.
- Persistence and writes: No metadata persistence, transaction, replay, cleanup, or data-write path is modified.
- Performance: M1 is blocking because zero-key hashing removes intra-BE partial-aggregation parallelism and concentrates multi-distinct memory in one task.
Review focus
No additional user-provided focus was supplied; the complete PR scope was reviewed.
Completion
Two review rounds converged with every candidate accepted, merged, or dismissed with evidence. No unresolved review candidate remains.
| // causing wrong-result (e.g. PASSTHROUGH over serial child breaks the | ||
| // group-by-key invariant — DORIS-25413). | ||
| if (!hasKeys) { | ||
| if (!hasPartitionRequirement()) { |
There was a problem hiding this comment.
[P1] Do not create a zero-key HASH exchange for direct multi-distinct calls
This predicate also matches a directly called scalar multi_distinct_count(col). That function has isDistinct=false, so SplitAggWithoutDistinct builds a two-phase LOCAL aggregate with no partitionExpressions; over a non-hash scan both the child distribution list and grouping list are empty. With the default LE setting this branch now reaches requireHash(), and getLocalExchangeDistributeExprs() supplies zero expressions. BE's hash partitioner initializes every row to hash 0, so the resulting FE-planned HASH exchange sends the whole input to channel 0, collapsing the partial aggregate to one task per BE and concentrating its distinct state in one task. BE's own AggSinkOperatorX sees empty _partition_exprs on this path and uses the base requirement instead. Please base the LOCAL decision on a non-empty effective key (while retaining the keyed finalize fix), and add a direct-function plan case that rejects a zero-expression HASH exchange.
| sv.enableBroadcastJoinForcePassthrough = true; | ||
| sv.aggPhase = 1; | ||
| }); | ||
| assertHasLocalExchangeOfType("select count(distinct a.k2) from test.t1 a " |
There was a problem hiding this comment.
[P2] Assert the distinct key, not just the HASH type
assertHasLocalExchangeOfType flattens every local exchange in every fragment to an enum set, discarding both its plan-tree location and distributeExprLists. The correctness invariant here is that the exchange directly below the scalar multi_distinct_count finalize aggregate is keyed by a.k2; a keyless or wrong-key LOCAL_EXECUTION_HASH_SHUFFLE still passes this assertion (and the new mock fixture currently accepts a zero-key HASH node). Please use the existing structural plan-shape traversal, or an equivalent targeted walk, to assert that aggregate-to-exchange edge and its partition expressions.
| // hash-partitioned by the probe key, and the satisfy() check lets the agg keep | ||
| // that distribution without inserting a redundant LE — the hash demand must | ||
| // still be recognized (a plain hash join probe output satisfies it). | ||
| setupLocalShuffleSession(null); |
There was a problem hiding this comment.
[P2] Configure and verify the no-force scenario explicitly
This call resets neither aggPhase nor enableBroadcastJoinForcePassthrough. Because the test class keeps one ConnectContext, it may inherit aggPhase=1 and force-passthrough=true from the other new test; in a clean session it instead uses the default aggregate strategy that the preceding comment says is already safe. In either case, anyTree(agg()) is true for every successful count(distinct ...) plan and cannot show that a compatible child hash avoided a redundant LE. Set aggPhase=1 and force-passthrough=false here (and reset shared state in setup), then assert the claimed aggregate-child distribution.
2 participants
Problem
With the FE local-shuffle planner enabled (default
enable_local_shuffle_planner=true), a scalarCOUNT(DISTINCT k)over joins can return a wrong result that grows linearly withparallel_pipeline_task_num(e.g. expected 10, got 30 with 3 tasks).The bad plan shape:
AggregationNode.enforceAndDeriveLocalExchangegave a NoRequire distribution to a finalize merge agg with no group keys, treating it likeCOUNT(*). UnlikeCOUNT(*), amulti_distinct_countfinalize agg emits per-instance scalar values that the parentsum0adds up — correctness requires the input to be hash-partitioned by the distinct key. When a PASSTHROUGH local exchange (broadcast-join probe fan-out) scatters same-key rows across instances, the parent double-counts overlapping keys. The result equalscorrect × local task count.The BE-native path was already protected (
AggSinkOperatorX::required_data_distributionchecks_partition_exprs, andchild_breaks_local_key_distributionfrom a prior fix), soenable_local_shuffle_planner=falsewas unaffected — only the FE-planned path was wrong.Root cause
The FE planner used
hasKeys(grouping exprs empty?) as the partition-requirement test, but BE's_partition_exprsis non-empty whenever the agg has group keys or DISTINCT aggregates (distribute_expr_lists+has_distinct). The FE fell back to NoRequire for the distinct case, skipping the hash local exchange that the agg needs.Changes
AggregationNodenow mirrors BE's_partition_exprssemantics viahasPartitionRequirement()(grouping exprs ormulti_distinct_*functions): a finalize agg with a partition requirement demands HASH from its child; only partition-less aggs (COUNT(*)-style) keep NoRequire.satisfy()check passes and no LE is inserted, so the common case is unchanged and free.requiresShuffleForCorrectness()now covers DISTINCT aggregates to match BE'sis_shuffled_operator().Tests
LocalShuffleNodeCoverageTest: unit coverage forAggregationNodeacross finalize/LOCAL/FIRST_MERGE phases × distinct/no-distinct ×enable_local_exchange_before_aggon/off, plusrequiresShuffleForCorrectnesscases. Pre-fix the distinct-finalize case asserted NoRequire; post-fix it asserts RequireHash.LocalExchangePlannerTest: sql-level distributed-plan test — the RQG-shaped query (count(distinct)over a shuffle join + a broadcast join with probe forced to PASSTHROUGH) must contain aLOCAL_EXECUTION_HASH_SHUFFLElocal exchange below the distinct finalize agg. Verified this test fails without the fix (plan only hasPASSTHROUGH) and passes with it.parallel_pipeline_task_num) now returns 10 under all session-var combinations, includingparallel_pipeline_task_num=1/2/4/6.