…3766)

Related PR: apache#63529, apache#62438

Problem Summary: `enable_local_exchange_before_agg=false` allows
first-phase aggregation to skip the local hash exchange before agg for
performance. This is only correct when the input still preserves local
key distribution.

After apache#62438, nested loop join and other operators may introduce
non-hash local exchanges such as `ADAPTIVE_PASSTHROUGH`. Those exchanges
can split rows with the same group/distinct key across local pipeline
tasks. If agg still skips the hash local exchange, partial aggregation
states for the same key are built in different tasks and later
`COUNT(DISTINCT ...)` can over-count. The reproduced query in
`output/ddl.txt` returned wrong counts such as `18/20` instead of `10`.

This PR preserves correctness while keeping the knob usable:

- Aggregation operators now skip local exchange with
`enable_local_exchange_before_agg=false` only when the child preserves
local key distribution.
- The shared child-distribution check is reused by `AggSinkOperatorX`,
`StreamingAggOperatorX`, and `DistinctStreamingAggOperatorX`.
- `Pipeline::need_to_local_exchange()` also handles the case where the
current pipeline source is a non-hash `LocalExchangeSource` but the
downstream target requires hash distribution, so inherited hash-ish
pipeline state cannot incorrectly suppress the required local exchange.
- Regression coverage is added for the nested-loop-join + distinct
aggregation wrong-result case, and unit tests cover the agg distribution
decisions.

…xchange

DORIS-25413 (closed in April as 'fix after the FE local exchange planner'):
count(distinct)+std with RIGHT JOIN under use_serial_exchange=true +
enable_local_exchange_before_agg=false returned an inflated distinct count —
the BE inserted PASSTHROUGH after the serial hash exchange, breaking the
HASH(s) invariant before the merge agg.

Both halves are now covered and this case pins them:
- BE-planned path: fixed upstream by child_breaks_local_key_distribution
  (#63766), inherited via the master rebase.
- FE-planned path: the require framework asks for hash semantically, so a
  hash local exchange is inserted instead of PASSTHROUGH by construction.

Verified with the original reproduction recipe on a 3-BE cluster: both
planner modes match the no-serial baseline (29 distinct values).

…efore aggregation (#66104)

### What problem does this PR solve?

Issue Number: None

Related PR: #62438, #63529, #63766

Problem Summary:

This safely backports #62438 to `branch-4.1`, adding the
`enable_local_exchange_before_agg` session variable so users can skip
unnecessary local exchanges before eligible non-finalizing aggregation.

A literal cherry-pick of #62438 is unsafe: when a serial or another
non-hash local exchange precedes aggregation, disabling the exchange can
split equal group or DISTINCT keys across local pipeline tasks and
produce incorrect results. This backport therefore also includes the
final correctness guard from #63766 and the serial-exchange regression
coverage from #63529:

- Skip the local exchange only when the child preserves local key
distribution.
- Require a hash exchange after serial and other non-hash local
exchanges.
- Handle a non-hash `LocalExchangeSource` when the downstream target
requires hash distribution.
- Cover the session-variable Thrift propagation, distribution decisions,
serial DISTINCT aggregation, and nested-loop-join DISTINCT aggregation.

`branch-4.1` already contains the compatible
`enable_local_exchange_before_agg` Thrift field, so this PR keeps the
existing field id and adds no duplicate IDL field.

### Release note

Add the `enable_local_exchange_before_agg` session variable. Disabling
it can reduce local exchanges before eligible non-finalizing aggregation
while preserving required hash exchanges for correctness.

### Check List (For Author)

- Test
    - [x] Regression test
- Worktree-isolated config:
`output/local-regression/regression-conf-11600.groovy` (FE query port
`20630`, BE heartbeat port `20650`)
- `doris-local-regression --network 10.26.20.3/24 --offset 11600 all -d
nereids_syntax_p0 -s agg_4_phase` (1 suite passed; also completed
`./build.sh --be --fe`)
- `doris-local-regression --network 10.26.20.3/24 --offset 11600 start
&& doris-local-regression --network 10.26.20.3/24 --offset 11600 run -d
query_p0/join -s test_agg_after_nested_loop_join_local_exchange` (1
suite passed)
    - [x] Unit Test
- `./run-be-ut.sh --run
--filter='StreamingAggOperatorTest.require_hash_shuffle_after_non_hash_local_exchange:DistinctStreamingAggOperatorTest.require_hash_shuffle_after_non_hash_local_exchange:AggOperatorRequiredDistributionTest.*:AggOperatorGroupByLimitOptTestWithGroupBy.*:AggOperatorTestWithGroupBy.test_2_phase:AggOperatorTestWithGroupBy.other_case_3:AggOperatorTestWithOutGroupBy.*'
-j 64` (21 tests passed)
- `./run-fe-ut.sh --run org.apache.doris.qe.VariableMgrTest` (12 tests
passed)
- `build-support/clang-format.sh`, `build-support/check-format.sh`, and
`git diff --check` passed
- `build-support/run-clang-tidy.sh` is absent on `branch-4.1`; a direct
focused `clang-tidy` attempt was blocked by pre-existing
toolchain/header diagnostics (`stddef.h` lookup and
`be/src/core/types.h` unmatched `NOLINTEND`), with no changed-line
diagnostic reported before the blocker
    - [ ] Manual test (add detailed scripts or steps below)
    - [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
        - [ ] Previous test can cover this change.
        - [ ] No code files have been changed.
        - [ ] Other reason

- Behavior changed:
    - [ ] No.
- [x] Yes. Add a session variable that controls eligible local exchanges
before aggregation while retaining correctness-required hash exchanges.

- Does this need documentation?
    - [x] No. This is a branch backport of an existing session variable.
    - [ ] Yes.

### Check List (For Reviewer who merge this PR)

- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label

Co-authored-by: Gabriel <[email protected]>
Co-authored-by: Pxl <[email protected]>

…local shuffle planner (#70177)

### What problem does this PR solve?

Issue Number: None

Related PR: #66216, #66222, #63766

Problem Summary:

**In short.** This brings the FE local shuffle planner in line with
`enable_local_exchange_before_streaming_agg`, which branch-4.1 / 4.2
#66216 and master #66222 added on the BE side only. With the FE planner,
which master uses by default, a streaming pre-aggregation still asked
for a local hash exchange on its grouping keys whatever the variable
said. Master plans therefore had a `LOCAL_EXECUTION_HASH_SHUFFLE` local
exchange that the BE planner, and branch-4.x, do not add, for example
right above a `UNION ALL` or an external table scan. The FE now follows
the same rule as BE `StreamingAggOperatorX`.

**Background.** #66222 added
`enable_local_exchange_before_streaming_agg` (default false) so that a
streaming pre-aggregation keeps the distribution it inherits. It
replaced `enable_local_exchange_before_agg` for streaming aggregation.
In `StreamingAggOperatorX::required_data_distribution()`, the operator
asks for a hash local exchange on its partition expressions only when
the variable is on, or when
`OperatorBase::child_breaks_local_key_distribution()` (#63766) says the
child spreads rows without regard to any key: the child is serial, or
the child's own required distribution is a non-hash local exchange
(PASSTHROUGH, ADAPTIVE_PASSTHROUGH, ...).

**The problem, and what it cost.** On master the FE plans local
exchanges (`enable_local_shuffle_planner`, default true) and BE does
not. `AggregationNode.enforceAndDeriveLocalExchange()` mirrors the BE
operators, but its streaming aggregation branch still had the rule from
before #66222. It never read the new variable. It skipped the hash
requirement only when `enable_local_exchange_before_agg` was off, and
then without the child check. With the default settings it required a
hash distribution for every streaming pre-aggregation with grouping
keys, so:

- Below a streaming pre-aggregation whose child keeps the key
distribution but does not report a hash distribution, the FE inserted a
`LOCAL_EXECUTION_HASH_SHUFFLE` that BE planning does not. Examples are a
`UNION ALL` and an external table scan that is not serial. Every row was
hashed and moved once more before the pre-aggregation.
- `enable_local_exchange_before_streaming_agg=true` had no effect with
the FE planner.
- With `enable_local_exchange_before_agg=false`, the FE dropped the hash
requirement even after a child that breaks the key distribution, where
BE keeps it.

None of this changes query results: a streaming pre-aggregation only
affects performance, and the operators that need hash-distributed input
for correctness ask for it themselves.

**How this PR fixes it.** The streaming aggregation branch of
`AggregationNode.enforceAndDeriveLocalExchange()` now follows BE: after
the existing hash join passthrough rule, it keeps the base class
requirement (no exchange, or PASSTHROUGH over a serial child) unless
`enable_local_exchange_before_streaming_agg` is on or the child breaks
the key distribution. `enable_local_exchange_before_agg` no longer
affects streaming aggregation. The new
`childBreaksLocalKeyDistribution()` mirrors
`child_breaks_local_key_distribution()` with each BE operator's
`required_data_distribution()`:

| Child | BE operator's required distribution | Breaks the key
distribution |
|---|---|---|
| serial on BE (pooling scan, serial exchange, ...) | — | yes |
| nested loop join | ADAPTIVE_PASSTHROUGH, NOOP for null-aware anti,
right outer / semi / anti and full outer joins | yes, except those joins
|
| broadcast hash join | PASSTHROUGH when forced
(`enable_broadcast_join_force_passthrough`) or when the probe child is
serial | in those two cases |
| table function, assert num rows | PASSTHROUGH | yes |
| anything else (scan, exchange, union, partitioned join, ...) | NOOP or
a hash distribution | no |

`SelectNode` and `RepeatNode` take the serial flag of their child, so
the serial check already covers the base class rule (PASSTHROUGH over a
serial child) for them.

**Results.** Local exchanges in the plans of three queries on a one-BE
cluster, `parallel_pipeline_task_num=4`, non-pooling scans, taken from
the query profiles:

| Query | BE planner | FE planner before | FE planner after |
|---|---|---|---|
| `GROUP BY` over `UNION ALL` of two tables | none | LOCAL_HASH_SHUFFLE
| none |
| `GROUP BY` over a nested loop join | ADAPTIVE_PASSTHROUGH,
LOCAL_HASH_SHUFFLE | same | same |
| `GROUP BY` over an Iceberg table with 6 data files | none |
LOCAL_HASH_SHUFFLE | none |

A planner probe over 72 plans (8 query shapes × pooling / non-pooling
scans × `agg_phase` 0 / 2 × the variable off / on) found only the `UNION
ALL` plans changed with the variable off. A scan of an OLAP table, a
nested loop join, `explode`, broadcast and shuffle joins, a `LIMIT`
subquery and `GROUPING SETS` gave the same plans as before.

…r key-breaking children (#70296)

### Background

On master the FE plans the local exchanges that redistribute rows among
the pipeline tasks of one BE (`enable_local_shuffle_planner`, on by
default). Each plan node follows the rule of the BE operator it becomes,
`required_data_distribution()`, which BE applies itself when the FE
planner is off. `enable_local_exchange_before_agg` (default true)
controls the exchange before a LOCAL (first-phase) aggregation. Since
#63766, BE skips it with the switch off only while the child keeps the
grouping-key distribution; a child breaks it when it is serial or asks
for a non-hash local exchange itself, such as the probe side of a nested
loop join or a table function.

### What problem does this PR solve?

Issue Number: None
Related PR: #63366 (introduced the FE rule without the child check),
#63766 (the BE rule), #70177 (the same fix for streaming aggregation;
this PR reuses its `childBreaksLocalKeyDistribution()`), #62438 (added
the switch)
PR Type: Bugfix

Problem Summary:

With `enable_local_exchange_before_agg=false`, the FE planner dropped
the local exchange below a LOCAL `AggSink` and below a streaming
distinct pre-aggregation whatever their child was, while BE keeps it
after a child that breaks the key distribution. Queries planned by the
FE therefore got fewer local exchanges than BE planning gives, for
example no hash exchange between a nested loop join and the
pre-aggregation above it. This PR adds BE's child check to both cases.
Results do not change; most of the diff is tests.

### Details

- **Change map:** production code +15 / -7 in `AggregationNode` (two
conditions and their comments) and a `PlanNode` javadoc; tests +345 /
-34: cases in `LocalShuffleNodeCoverageTest` and
`LocalExchangePlannerTest`, and the new regression suite
`nereids_p0/local_shuffle/test_local_exchange_before_agg_breaking_child`
with its generated `.out` (+169).
- **Frequency:** Always, when `enable_local_exchange_before_agg=false`
(not the default; P0 randomizes it) and a LOCAL aggregation sits on a
child that breaks the key distribution. Only plans and speed change: the
FE always requires hash input at FIRST_MERGE and at the non-streaming
DISTINCT dedup (a deliberate divergence from BE), so results stay right.
- **Root cause:** `AggregationNode.enforceAndDeriveLocalExchange()`
mirrors the BE operators, but its DistinctStreamingAgg and AggSink LOCAL
branches fall back to the base class on `!enableLeBeforeAgg` alone,
without BE's `child_breaks_local_key_distribution()`. #63366 wrote them
after #63766 had merged, from the rule before it.
- **Why existing tests missed it:** Unit tests: the two "LE disabled"
cases in `LocalShuffleNodeCoverageTest` use a child that keeps the key
distribution, no case reached the DistinctStreamingAgg branch, and
`LocalExchangePlannerTest` had no plan with the switch off. Regression:
`query_p0/join/test_agg_after_nested_loop_join_local_exchange` (#63766's
suite) and
`nereids_p0/local_shuffle/test_enable_local_exchange_before_agg` compare
only results, which were right, and no `.out` shows local exchanges.
- **Fix:** Both branches fall back to the base class only when
`childBreaksLocalKeyDistribution()` is false. After a key-breaking
child, the AggSink LOCAL phase goes on to its hash requirement, and the
streaming distinct pre-aggregation to
`enable_distinct_streaming_agg_force_passthrough` (PASSTHROUGH, else the
base class), as on BE; it never asks for hash.
- **Other paths and branches:** Streaming aggregation got the check in
#70177; BE needs no change. One known difference from BE planning
remains, with the switch on or off: the FE lets an ADAPTIVE_PASSTHROUGH
or PASSTHROUGH child output satisfy a PASSTHROUGH requirement, where BE
`Pipeline::need_to_local_exchange()` always adds one (last row of the
table in Testing). branch-4.1 and branch-4.2 have no FE local shuffle
planner and are not affected.

### Testing

- `LocalShuffleNodeCoverageTest` and `LocalExchangePlannerTest`, two new
cases each: all four fail without the fix (for example, a serial child
got `PASSTHROUGH` instead of `LOCAL_EXECUTION_HASH_SHUFFLE`) and pass
with it. The 12 fe-core test classes that plan local exchanges or print
distributed plans pass (168 tests). Run locally.
- A planner probe over 288 plans (18 query shapes × pooling on and off ×
`agg_phase` 0 and 2 × `disable_streaming_preaggregations` on and off ×
the switch): only 32 with the switch off change. 30 of them now equal
the plan with the switch on; the other 2 keep the distribution of a
`UNION ALL`, as the switch asks.
- Local exchanges in the query profiles on a local 1 FE + 1 BE cluster
(master `4530f761fa1`; `doris-fe.jar` from `7c4b4462dcc` with #70177,
and for "after" with only `AggregationNode.class` replaced), switch off,
`parallel_pipeline_task_num=4`, `agg_phase=2`. The `AggSink` rows set
`disable_streaming_preaggregations=true`; the distinct rows are a `GROUP
BY` without aggregate functions:

| Query | BE planner | FE planner before | FE planner after |
|---|---|---|---|
| LOCAL `AggSink` over a nested loop join | ADAPTIVE_PASSTHROUGH,
LOCAL_HASH_SHUFFLE | ADAPTIVE_PASSTHROUGH | ADAPTIVE_PASSTHROUGH,
LOCAL_HASH_SHUFFLE |
| LOCAL `AggSink` over `explode` | PASSTHROUGH, LOCAL_HASH_SHUFFLE |
PASSTHROUGH | PASSTHROUGH, LOCAL_HASH_SHUFFLE |
| LOCAL `AggSink` over a one-tablet scan (serial) | PASSTHROUGH,
LOCAL_HASH_SHUFFLE | PASSTHROUGH | PASSTHROUGH, LOCAL_HASH_SHUFFLE |
| LOCAL `AggSink` over `UNION ALL` | none | none | none |
| distinct pre-aggregation over `explode` | PASSTHROUGH ×2 | PASSTHROUGH
| PASSTHROUGH ×2 |
| distinct pre-aggregation over a nested loop join |
ADAPTIVE_PASSTHROUGH, PASSTHROUGH | ADAPTIVE_PASSTHROUGH |
ADAPTIVE_PASSTHROUGH |

- The new regression suite fails on the FE without the fix at its first
plan check (`nlj: [VNESTED]`) and passes with it; its 16 result blocks
(4 queries × both planners × the switch on and off) match results
computed independently from the inserted rows.
`nereids_p0/local_shuffle` (11 suites),
`query_p0/join/test_agg_after_nested_loop_join_local_exchange` and
`nereids_syntax_p0/agg_4_phase` pass on the same cluster with the fix.

### Notes for Reviewers

- `baseClassRequire()` stays in both branches although only a non-serial
child reaches them now, since a serial child always breaks the key
distribution. It keeps the branches shaped like BE's call to the base
class, as #70177 does for streaming aggregation.
- In `LocalShuffleNodeCoverageTest`, #70177's five key-breaking children
moved into `childrenThatBreakKeyDistribution()` and its aggregate
builder into `buildLocalPhaseAgg()`, so the new cases reuse them;
#70177's case is otherwise unchanged.
- The `PlanNode#requiresShuffleForCorrectness` javadoc still named
`enable_local_exchange_before_agg` for streaming aggregation, which
reads `enable_local_exchange_before_streaming_agg` since #70177.