…he#66222)

### What

Add `enable_local_exchange_before_streaming_agg`, defaulting to `false`,
and use it to control whether streaming aggregation requests a local
hash exchange.

### Why

Using the generic aggregation switch changes streaming aggregation
whenever it is enabled. A dedicated switch preserves the existing
streaming aggregation distribution by default. The master-specific
non-hash child distribution guard remains in place so grouping keys are
reshuffled when required.

### Validation

- Not run as requested.

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

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