[fix](fe) Honor enable_local_exchange_before_streaming_agg in the FE local shuffle planner - #70177

…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.

…#70197)

### What problem does this PR solve?

Issue Number: None

Related PR: #69273, #70177

Problem Summary:

**In short.** Found while bringing branch-4.2 #66216 to the FE local
shuffle planner (#70177): four session variables that decide the local
exchanges of a plan are not forwarded to the master. #69273 made the
planner and executor variables travel with a statement that a follower
or an observer forwards to the master,
`enable_local_exchange_before_agg` among them, but not
`enable_local_exchange_before_streaming_agg`,
`enable_streaming_agg_hash_join_force_passthrough`,
`enable_distinct_streaming_agg_force_passthrough` and
`force_to_local_shuffle`. A session-level `SET` of one of them is
dropped on that path, and the master plans the statement with its own
global value. This PR forwards the four as well.

**Background.** A statement that has to run on the master FE (`INSERT
INTO ... SELECT`, `CREATE TABLE ... AS SELECT`, ...) and arrives at a
follower or an observer is forwarded: `FEOpExecutor` sends
`SessionVariable#getForwardVariables()`, and `ConnectProcessor` on the
master applies them with `setForwardedSessionVariables()` to a session
that starts from the master's global variables. Only variables whose
`@VarAttr` sets `needForward` (or one of the `affectQueryResult*`
attributes) are sent.

**The problem, and what it cost.** These four do not set it:

| Variable | Default | What it decides |
|---|---|---|
| `enable_local_exchange_before_streaming_agg` (#66222) | false |
whether a streaming pre-aggregation asks for a local hash exchange on
its grouping keys: BE `StreamingAggOperatorX`, and the FE planner with
#70177 |
| `enable_streaming_agg_hash_join_force_passthrough` | true | whether a
streaming pre-aggregation right above a hash join asks for a passthrough
local exchange: the FE planner (`AggregationNode`) and BE
`StreamingAggOperatorX` |
| `enable_distinct_streaming_agg_force_passthrough` | true | the same
for a distinct streaming pre-aggregation: the FE planner and BE
`DistinctStreamingAggOperatorX` |
| `force_to_local_shuffle` | false | makes every scan serial
(`ScanNode#isSerialNode`), so the FE planner pools the scan and adds a
local shuffle above it |

So a session that sets one of them gets that plan only for the
statements its own FE runs. A forwarded statement, such as an `INSERT
INTO ... SELECT` or a CTAS issued through a follower or an observer, is
planned with the master's global value, and `EXPLAIN` on the follower
shows a plan that the master does not run. Query results stay the same,
since these variables only decide how rows move between the pipeline
tasks of a BE; what differs is the plan, and with it the speed of the
statement.

**How this PR fixes it.** Set `needForward = true` on the four, as
#69273 does for the others, and add them to the list in
`SessionVariablesTest#testForwardPlanAndExecutionKnobs`.

**Results.**

| Statement forwarded from a follower after `SET <variable> =
<non-default value>` | Before | After |
|---|---|---|
| Value the master plans it with | the master's global value | the
session's value |

branch-4.1 and branch-4.2 have the same gap, and #69273 is not on them
either.