924060929 opened a new pull request, #66570:
URL: https://github.com/apache/doris/pull/66570
## Problem
With the FE local-shuffle planner enabled (default
`enable_local_shuffle_planner=true`), a scalar `COUNT(DISTINCT k)` over joins
can return a wrong result that grows linearly with `parallel_pipeline_task_num`
(e.g. expected 10, got 30 with 3 tasks).
The bad plan shape:
```
VAGGREGATE (merge finalize) output: sum0(multi_distinct_count(k)) -- sums
per-instance values
VAGGREGATE (merge finalize) output: multi_distinct_count(k) -- no
group keys
VHASH JOIN (LEFT OUTER BROADCAST)
VLOCAL-EXCHANGE (PASSTHROUGH) --
scatters hash-partitioned rows
VHASH JOIN (RIGHT OUTER PARTITIONED) output: hash-partitioned by k --
key-aligned here
```
`AggregationNode.enforceAndDeriveLocalExchange` gave a **NoRequire**
distribution to a finalize merge agg with no group keys, treating it like
`COUNT(*)`. Unlike `COUNT(*)`, a `multi_distinct_count` finalize agg emits
per-instance **scalar values** that the parent `sum0` adds 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
equals `correct × local task count`.
The BE-native path was already protected
(`AggSinkOperatorX::required_data_distribution` checks `_partition_exprs`, and
`child_breaks_local_key_distribution` from a prior fix), so
`enable_local_shuffle_planner=false` was 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_exprs` is 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
- `AggregationNode` now mirrors BE's `_partition_exprs` semantics via
`hasPartitionRequirement()` (grouping exprs or `multi_distinct_*` functions): a
finalize agg with a partition requirement demands HASH from its child; only
partition-less aggs (`COUNT(*)`-style) keep NoRequire.
- The finalize branch that previously trusted the child's distribution now
requires HASH explicitly — when the child already provides hash distribution
the `satisfy()` check passes and no LE is inserted, so the common case is
unchanged and free.
- `requiresShuffleForCorrectness()` now covers DISTINCT aggregates to match
BE's `is_shuffled_operator()`.
## Tests
- `LocalShuffleNodeCoverageTest`: unit coverage for `AggregationNode` across
finalize/LOCAL/FIRST_MERGE phases × distinct/no-distinct ×
`enable_local_exchange_before_agg` on/off, plus `requiresShuffleForCorrectness`
cases. 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 a `LOCAL_EXECUTION_HASH_SHUFFLE`
local exchange below the distinct finalize agg. Verified this test fails
without the fix (plan only has `PASSTHROUGH`) and passes with it.
- End-to-end on a 3-BE cluster with the RQG dataset: expected 10; the
pre-fix behavior (30, scaling with `parallel_pipeline_task_num`) now returns 10
under all session-var combinations, including
`parallel_pipeline_task_num=1/2/4/6`.
--
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]
