…efore_agg

Covers #62438. Tests 10 agg query patterns with both
enable_local_exchange_before_agg=true (FE vs BE planned LE) and
enable_local_exchange_before_agg=false (verify no crash/hang).

…_before_agg

Adapt FE AggregationNode.enforceAndDeriveLocalExchange to match
#62438: when enable_local_exchange_before_agg=true (default)
and !needsFinalize, all 3 BE agg operators request HASH_SHUFFLE.

- Extract common !needsFinalize && !enableLeBeforeAgg early return
- StreamingAgg: force_passthrough_after_join takes highest priority
- DistinctStreamingAgg: add missing needsFinalize + emptyGroupKeys → NOOP
- AggSink: non-finalize with enableLeBeforeAgg → requireHash;
  finalize preserves hasPartitionExprs heuristic

Reduces FE/BE consistency mismatches from 8 to 4 (3 known NLJ + 1
union_all_followed_by_agg over-insertion).

…equires shuffle for correctness

PR #62438 introduced a semantic split for required_data_distribution=HASH:
  - "I'm a shuffled operator, downstream depends on hash for correctness"
    (finalize AggSink, HashJoin partitioned, Intersect/Except)
  - "I'd like hash distribution for performance optimization"
    (StreamingAgg pre-agg with enable_local_exchange_before_agg=true)

Old FE UnionNode used parentRequire.preferType().isHashShuffle() as the
trigger to propagate hash to its inputs. This conflated the two intents —
once StreamingAgg started requesting HASH (PR #62438), every union under
a streaming preagg over-inserted HASH LE on each input branch.

Mirror BE's _followed_by_shuffled_operator propagation:
  - PlanNode.requiresShuffleForCorrectness() — default false, override in
    AggregationNode (finalize+keys), HashJoinNode (partitioned/bucket-shuffle/
    colocate), IntersectNode/ExceptNode (always).
  - PlanTranslatorContext.shuffledAncestorMap — per-node flag, set by parent
    in enforceRequire before recursing.
  - SetOperationNode reads the flag instead of parentRequire type.

Reduces FE/BE consistency mismatches from 4 to 3 (only pre-existing NLJ
optimization differences remain).

PR #62438 introduced a semantic split for required_data_distribution=HASH:
  - "I'm a shuffled operator, downstream depends on hash for correctness"
    (finalize AggSink, HashJoin partitioned, Intersect/Except)
  - "I'd like hash distribution for performance optimization"
    (StreamingAgg pre-agg with enable_local_exchange_before_agg=true)

Old FE UnionNode used parentRequire.preferType().isHashShuffle() as the
trigger to propagate hash to its inputs. This conflated the two intents —
once StreamingAgg started requesting HASH (PR #62438), every union under
a streaming preagg over-inserted HASH LE on each input branch.

Mirror BE's _followed_by_shuffled_operator propagation:
  - PlanNode.requiresShuffleForCorrectness() — default false, override in
    AggregationNode (finalize+keys), HashJoinNode (partitioned/bucket-shuffle/
    colocate), IntersectNode/ExceptNode (always).
  - PlanTranslatorContext.shuffledAncestorMap — per-node flag, set by parent
    in enforceRequire before recursing.
  - SetOperationNode reads the flag instead of parentRequire type.

Refine the AggregationNode branches with phase-aware HASH requirement (DORIS-25413):

BE's three agg operators all guard with `!_needs_finalize &&
!enable_local_exchange_before_agg → base`, but the semantics differ:

  - AggSink: `!finalize && hasKeys` covers BOTH LOCAL preagg (FIRST / SECOND
    distinct local, performance-only) AND FIRST_MERGE dedup (correctness).
    BE's flag-gated early return wrongly catches FIRST_MERGE, leading to
    PASSTHROUGH-over-serial-child → wrong result.

  - DistinctStreaming: `!finalize` covers BOTH streaming preagg
    (useStreamingPreagg=true, performance) AND non-streaming dedup
    (useStreamingPreagg=false, correctness).  Same family of bug.

  - StreamingAgg: only ever streaming preagg, flag-gated path is fine.

FE fix:
  - AggSink: restrict the flag-gated base path to non-merge LOCAL phases
    (`!isMerge()`).  FIRST_MERGE always goes to HASH regardless of the flag.
  - DistinctStreaming: restrict to streaming preagg (`useStreamingPreagg=true`).
    Non-streaming dedup always goes to HASH.
  - requiresShuffleForCorrectness() mirrors BE's is_shuffled_operator() exactly
    (`finalize && hasKeys`). For multi-phase chains, the merge phases above
    propagate the flag down via inheritedShuffled — no need to widen here.

Reduces FE/BE consistency mismatches from 4 to 3 (only pre-existing NLJ
optimization differences remain).

Previously, local exchange (LE) nodes were inserted exclusively by the
BE's `_plan_local_exchange` at pipeline build time.  The FE had no
visibility into which operators needed a fan-out or shuffle before
execution, making it impossible to validate, optimize, or override LE
decisions at planning time.

This PR introduces a full FE-side local exchange planner that mirrors
BE semantics, brings several correctness fixes, and leaves the legacy
BE path fully intact behind a feature flag.  See "Current architecture
notes" at the bottom for what the FE planner does and does not own.

A new `AddLocalExchange` pass runs after normal fragment assignment.
It walks each fragment's plan tree bottom-up, calling the polymorphic
`PlanNode.enforceAndDeriveLocalExchange()` on every node.  Nodes
declare what distribution they require of their children; the framework
inserts `LocalExchangeNode` where needed.

`LocalExchangeNode` represents intra-fragment data redistribution and
supports the full set of exchange types: PASSTHROUGH, HASH_SHUFFLE,
BUCKET_HASH_SHUFFLE, GLOBAL_EXECUTION_HASH_SHUFFLE, BROADCAST,
PASS_TO_ONE, ADAPTIVE_PASSTHROUGH, LOCAL_MERGE_SORT, and NOOP.

The pass is guarded by `enable_local_shuffle_planner` (default true).
When disabled, BE continues to run its own `_plan_local_exchange` as
before, keeping the old path fully intact.

`maxPerBeInstances` (max pipeline instances assigned to any single BE)
is used instead of a global `instanceCount`.  Planning is a no-op when
`maxPerBeInstances == 1` — inserting LE on a single-threaded pipeline
would cause task-count mismatches and pipeline starvation.

When a serial operator (e.g. OlapScanNode with a single tablet bucket)
feeds a non-serial parent without an intermediate LE, downstream tasks
starve waiting for data that never arrives.  The framework detects this
case and inserts a PASSTHROUGH LE to restore N-task parallelism, exactly
matching BE's `required_data_distribution()` serial → PASSTHROUGH rule.

`LocalExchangeTypeRequire` abstracts two strategies:
- `RequireHash` — always resolves to `LOCAL_EXECUTION_HASH_SHUFFLE`
  (safe for intra-fragment hash partitioning).
- `RequireSpecific` — preserves BUCKET_HASH_SHUFFLE /
  GLOBAL_EXECUTION_HASH_SHUFFLE without degradation.

PR #62438 added `enable_local_exchange_before_agg`, but its BE guard
`!_needs_finalize && !enable_local_exchange_before_agg → base` conflated
two semantically different cases in AggSink and DistinctStreamingAgg:

- **AggSink**: `!finalize && hasKeys` covered both LOCAL preagg
  (performance-only) and FIRST_MERGE dedup (correctness-critical).
  The flag-gated early-return wrongly skipped HASH for FIRST_MERGE,
  producing PASSTHROUGH-over-serial-child → wrong aggregation results.

- **DistinctStreamingAgg**: `!finalize` covered both streaming preagg
  (`useStreamingPreagg=true`, performance) and non-streaming dedup
  (`useStreamingPreagg=false`, correctness).  Same class of bug.

FE fix:
- AggSink: restrict the flag-gated base path to `!isMerge()` LOCAL
  phases.  FIRST_MERGE always emits HASH regardless of the flag.
- DistinctStreamingAgg: restrict to `useStreamingPreagg=true`.
  Non-streaming dedup always emits HASH.

Also add `requiresShuffleForCorrectness()` to mirror BE's
`is_shuffled_operator()`, so SetOperationNode propagates the
"downstream depends on hash" flag correctly instead of using the
coarser `parentRequire.preferType().isHashShuffle()` check that
over-inserted HASH LE on every union branch under a streaming preagg.

These fixes reduce FE/BE consistency mismatches from 8 to 3
(only pre-existing NLJ optimization differences remain).

- `enable_local_shuffle_planner` — use FE planner (default true)
- `enable_local_shuffle` — master switch for local shuffle
- `enable_local_exchange_before_agg` — HASH LE before non-final agg
  (default true, mirrors #62438)

`validateNoSerialWithoutLocalExchange()` walks the final plan tree and
logs a warning whenever a serial operator feeds a non-serial parent
without an intermediate LocalExchangeNode, catching planning gaps
before execution.

- `test_enable_local_exchange_before_agg.groovy` — 10 agg patterns
  with the flag on and off; covers the FIRST_MERGE and
  DistinctStreamingAgg correctness fixes.
- `test_local_shuffle_fe_be_consistency.groovy` — runs the same SQL
  with `enable_local_shuffle_planner=true` and `=false` across the
  full operator matrix (Agg, Sort, Analytic, HashJoin, NLJ, Set, Union,
  TableFunction, AssertNumRows, RQG-derived corner cases) and asserts
  result rows are identical.  Only data correctness is asserted — the
  two planners legitimately differ on the exact exchange counts/types
  they emit, so plan-shape equality is intentionally not checked.
- `test_local_shuffle_rqg_bugs.groovy` — reproduces 20+ RQG-found
  crashes and wrong-result cases.
- `test_old_coordinator_local_shuffle.groovy` — verifies the old
  coordinator path is unaffected.
- `test_multilevel_join_agg_local_shuffle.groovy` — multi-level join
  and aggregation plan shapes.

- `multi_version.h`: replace `atomic_load/atomic_store` (deprecated in
  libstdc++ C++20 / LLVM 20) with `std::shared_mutex`-based RW locking.
- `memory.cpp`: fix `std::max` type mismatch (`long` vs `int64_t`)
  on macOS.
- `bucketed_aggregation_sink_operator.h`: fix `ExchangeType::NOOP` →
  `TLocalPartitionType::NOOP` after thrift enum rename.

This PR puts the FE planner in the driver's seat for LE insertion but
intentionally does NOT remove the BE-side machinery — readers should be
aware of three pieces the FE planner shares with or defers to BE:

1. **`is_serial_operator` is computed on both sides.**  FE computes the
   flag and writes it into Thrift, but BE's
   `OperatorBase::is_serial_operator()` is still overridden per operator
   in C++ and used for BE-side runtime decisions.  Any future change to
   the BE override needs to be mirrored on the FE side (and vice versa)
   to keep the planner's view consistent with execution.

2. **The legacy BE planner stays as a fallback.**
   `pipeline_fragment_context.cpp::_plan_local_exchange` is preserved
   and gated by `runtime_state.h::plan_local_shuffle()`: when
   `enable_local_shuffle_planner=false`, BE plans LE itself, exactly as
   before.  The two paths are mutually exclusive, never both running on
   the same query.

3. **`_propagate_local_exchange_num_tasks` is kept as a runtime safety
   net.**  The two propagation passes in
   `pipeline_fragment_context.cpp` fix up paired pipelines whose
   `num_tasks` end up mismatched (e.g. when AGG/SORT/JOIN pipeline
   splits leave a serial Exchange feeding an N-task sink).  FE's
   framework-level serial→non-serial fan-out (`enforceRequire` step 3)
   and the `validateNoSerialWithoutLocalExchange` check aim to make
   these mismatches impossible by construction, but the BE-side fixup
   remains as a defensive guard.

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

Previously, local exchange (LE) nodes were inserted exclusively by the
BE's `_plan_local_exchange` at pipeline build time.  The FE had no
visibility into which operators needed a fan-out or shuffle before
execution, making it impossible to validate, optimize, or override LE
decisions at planning time.

This PR introduces a full FE-side local exchange planner that mirrors
BE semantics, brings several correctness fixes, and leaves the legacy
BE path fully intact behind a feature flag.  See "Current architecture
notes" at the bottom for what the FE planner does and does not own.

A new `AddLocalExchange` pass runs after normal fragment assignment.
It walks each fragment's plan tree bottom-up, calling the polymorphic
`PlanNode.enforceAndDeriveLocalExchange()` on every node.  Nodes
declare what distribution they require of their children; the framework
inserts `LocalExchangeNode` where needed.

`LocalExchangeNode` represents intra-fragment data redistribution and
supports the full set of exchange types: PASSTHROUGH, HASH_SHUFFLE,
BUCKET_HASH_SHUFFLE, GLOBAL_EXECUTION_HASH_SHUFFLE, BROADCAST,
PASS_TO_ONE, ADAPTIVE_PASSTHROUGH, LOCAL_MERGE_SORT, and NOOP.

The pass is guarded by `enable_local_shuffle_planner` (default true).
When disabled, BE continues to run its own `_plan_local_exchange` as
before, keeping the old path fully intact.

`maxPerBeInstances` (max pipeline instances assigned to any single BE)
is used instead of a global `instanceCount`.  Planning is a no-op when
`maxPerBeInstances == 1` — inserting LE on a single-threaded pipeline
would cause task-count mismatches and pipeline starvation.

When a serial operator (e.g. OlapScanNode with a single tablet bucket)
feeds a non-serial parent without an intermediate LE, downstream tasks
starve waiting for data that never arrives.  The framework detects this
case and inserts a PASSTHROUGH LE to restore N-task parallelism, exactly
matching BE's `required_data_distribution()` serial → PASSTHROUGH rule.

`LocalExchangeTypeRequire` abstracts two strategies:
- `RequireHash` — always resolves to `LOCAL_EXECUTION_HASH_SHUFFLE`
  (safe for intra-fragment hash partitioning).
- `RequireSpecific` — preserves BUCKET_HASH_SHUFFLE /
  GLOBAL_EXECUTION_HASH_SHUFFLE without degradation.

PR #62438 added `enable_local_exchange_before_agg`, but its BE guard
`!_needs_finalize && !enable_local_exchange_before_agg → base` conflated
two semantically different cases in AggSink and DistinctStreamingAgg:

- **AggSink**: `!finalize && hasKeys` covered both LOCAL preagg
  (performance-only) and FIRST_MERGE dedup (correctness-critical).
  The flag-gated early-return wrongly skipped HASH for FIRST_MERGE,
  producing PASSTHROUGH-over-serial-child → wrong aggregation results.

- **DistinctStreamingAgg**: `!finalize` covered both streaming preagg
  (`useStreamingPreagg=true`, performance) and non-streaming dedup
  (`useStreamingPreagg=false`, correctness).  Same class of bug.

FE fix:
- AggSink: restrict the flag-gated base path to `!isMerge()` LOCAL
  phases.  FIRST_MERGE always emits HASH regardless of the flag.
- DistinctStreamingAgg: restrict to `useStreamingPreagg=true`.
  Non-streaming dedup always emits HASH.

Also add `requiresShuffleForCorrectness()` to mirror BE's
`is_shuffled_operator()`, so SetOperationNode propagates the
"downstream depends on hash" flag correctly instead of using the
coarser `parentRequire.preferType().isHashShuffle()` check that
over-inserted HASH LE on every union branch under a streaming preagg.

These fixes reduce FE/BE consistency mismatches from 8 to 3
(only pre-existing NLJ optimization differences remain).

- `enable_local_shuffle_planner` — use FE planner (default true)
- `enable_local_shuffle` — master switch for local shuffle
- `enable_local_exchange_before_agg` — HASH LE before non-final agg
  (default true, mirrors #62438)

`validateNoSerialWithoutLocalExchange()` walks the final plan tree and
logs a warning whenever a serial operator feeds a non-serial parent
without an intermediate LocalExchangeNode, catching planning gaps
before execution.

- `test_enable_local_exchange_before_agg.groovy` — 10 agg patterns
  with the flag on and off; covers the FIRST_MERGE and
  DistinctStreamingAgg correctness fixes.
- `test_local_shuffle_fe_be_consistency.groovy` — runs the same SQL
  with `enable_local_shuffle_planner=true` and `=false` across the
  full operator matrix (Agg, Sort, Analytic, HashJoin, NLJ, Set, Union,
  TableFunction, AssertNumRows, RQG-derived corner cases) and asserts
  result rows are identical.  Only data correctness is asserted — the
  two planners legitimately differ on the exact exchange counts/types
  they emit, so plan-shape equality is intentionally not checked.
- `test_local_shuffle_rqg_bugs.groovy` — reproduces 20+ RQG-found
  crashes and wrong-result cases.
- `test_old_coordinator_local_shuffle.groovy` — verifies the old
  coordinator path is unaffected.
- `test_multilevel_join_agg_local_shuffle.groovy` — multi-level join
  and aggregation plan shapes.

- `multi_version.h`: replace `atomic_load/atomic_store` (deprecated in
  libstdc++ C++20 / LLVM 20) with `std::shared_mutex`-based RW locking.
- `memory.cpp`: fix `std::max` type mismatch (`long` vs `int64_t`)
  on macOS.
- `bucketed_aggregation_sink_operator.h`: fix `ExchangeType::NOOP` →
  `TLocalPartitionType::NOOP` after thrift enum rename.

This PR puts the FE planner in the driver's seat for LE insertion but
intentionally does NOT remove the BE-side machinery — readers should be
aware of three pieces the FE planner shares with or defers to BE:

1. **`is_serial_operator` is computed on both sides.**  FE computes the
   flag and writes it into Thrift, but BE's
   `OperatorBase::is_serial_operator()` is still overridden per operator
   in C++ and used for BE-side runtime decisions.  Any future change to
   the BE override needs to be mirrored on the FE side (and vice versa)
   to keep the planner's view consistent with execution.

2. **The legacy BE planner stays as a fallback.**
   `pipeline_fragment_context.cpp::_plan_local_exchange` is preserved
   and gated by `runtime_state.h::plan_local_shuffle()`: when
   `enable_local_shuffle_planner=false`, BE plans LE itself, exactly as
   before.  The two paths are mutually exclusive, never both running on
   the same query.

3. **`_propagate_local_exchange_num_tasks` is kept as a runtime safety
   net.**  The two propagation passes in
   `pipeline_fragment_context.cpp` fix up paired pipelines whose
   `num_tasks` end up mismatched (e.g. when AGG/SORT/JOIN pipeline
   splits leave a serial Exchange feeding an N-task sink).  FE's
   framework-level serial→non-serial fan-out (`enforceRequire` step 3)
   and the `validateNoSerialWithoutLocalExchange` check aim to make
   these mismatches impossible by construction, but the BE-side fixup
   remains as a defensive guard.

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

Previously, local exchange (LE) nodes were inserted exclusively by the
BE's `_plan_local_exchange` at pipeline build time.  The FE had no
visibility into which operators needed a fan-out or shuffle before
execution, making it impossible to validate, optimize, or override LE
decisions at planning time.

This PR introduces a full FE-side local exchange planner that mirrors
BE semantics, brings several correctness fixes, and leaves the legacy
BE path fully intact behind a feature flag.  See "Current architecture
notes" at the bottom for what the FE planner does and does not own.

A new `AddLocalExchange` pass runs after normal fragment assignment.
It walks each fragment's plan tree bottom-up, calling the polymorphic
`PlanNode.enforceAndDeriveLocalExchange()` on every node.  Nodes
declare what distribution they require of their children; the framework
inserts `LocalExchangeNode` where needed.

`LocalExchangeNode` represents intra-fragment data redistribution and
supports the full set of exchange types: PASSTHROUGH, HASH_SHUFFLE,
BUCKET_HASH_SHUFFLE, GLOBAL_EXECUTION_HASH_SHUFFLE, BROADCAST,
PASS_TO_ONE, ADAPTIVE_PASSTHROUGH, LOCAL_MERGE_SORT, and NOOP.

The pass is guarded by `enable_local_shuffle_planner` (default true).
When disabled, BE continues to run its own `_plan_local_exchange` as
before, keeping the old path fully intact.

`maxPerBeInstances` (max pipeline instances assigned to any single BE)
is used instead of a global `instanceCount`.  Planning is a no-op when
`maxPerBeInstances == 1` — inserting LE on a single-threaded pipeline
would cause task-count mismatches and pipeline starvation.

When a serial operator (e.g. OlapScanNode with a single tablet bucket)
feeds a non-serial parent without an intermediate LE, downstream tasks
starve waiting for data that never arrives.  The framework detects this
case and inserts a PASSTHROUGH LE to restore N-task parallelism, exactly
matching BE's `required_data_distribution()` serial → PASSTHROUGH rule.

`LocalExchangeTypeRequire` abstracts two strategies:
- `RequireHash` — always resolves to `LOCAL_EXECUTION_HASH_SHUFFLE`
  (safe for intra-fragment hash partitioning).
- `RequireSpecific` — preserves BUCKET_HASH_SHUFFLE /
  GLOBAL_EXECUTION_HASH_SHUFFLE without degradation.

PR #62438 added `enable_local_exchange_before_agg`, but its BE guard
`!_needs_finalize && !enable_local_exchange_before_agg → base` conflated
two semantically different cases in AggSink and DistinctStreamingAgg:

- **AggSink**: `!finalize && hasKeys` covered both LOCAL preagg
  (performance-only) and FIRST_MERGE dedup (correctness-critical).
  The flag-gated early-return wrongly skipped HASH for FIRST_MERGE,
  producing PASSTHROUGH-over-serial-child → wrong aggregation results.

- **DistinctStreamingAgg**: `!finalize` covered both streaming preagg
  (`useStreamingPreagg=true`, performance) and non-streaming dedup
  (`useStreamingPreagg=false`, correctness).  Same class of bug.

FE fix:
- AggSink: restrict the flag-gated base path to `!isMerge()` LOCAL
  phases.  FIRST_MERGE always emits HASH regardless of the flag.
- DistinctStreamingAgg: restrict to `useStreamingPreagg=true`.
  Non-streaming dedup always emits HASH.

Also add `requiresShuffleForCorrectness()` to mirror BE's
`is_shuffled_operator()`, so SetOperationNode propagates the
"downstream depends on hash" flag correctly instead of using the
coarser `parentRequire.preferType().isHashShuffle()` check that
over-inserted HASH LE on every union branch under a streaming preagg.

These fixes reduce FE/BE consistency mismatches from 8 to 3
(only pre-existing NLJ optimization differences remain).

- `enable_local_shuffle_planner` — use FE planner (default true)
- `enable_local_shuffle` — master switch for local shuffle
- `enable_local_exchange_before_agg` — HASH LE before non-final agg
  (default true, mirrors #62438)

`validateNoSerialWithoutLocalExchange()` walks the final plan tree and
logs a warning whenever a serial operator feeds a non-serial parent
without an intermediate LocalExchangeNode, catching planning gaps
before execution.

- `test_enable_local_exchange_before_agg.groovy` — 10 agg patterns
  with the flag on and off; covers the FIRST_MERGE and
  DistinctStreamingAgg correctness fixes.
- `test_local_shuffle_fe_be_consistency.groovy` — runs the same SQL
  with `enable_local_shuffle_planner=true` and `=false` across the
  full operator matrix (Agg, Sort, Analytic, HashJoin, NLJ, Set, Union,
  TableFunction, AssertNumRows, RQG-derived corner cases) and asserts
  result rows are identical.  Only data correctness is asserted — the
  two planners legitimately differ on the exact exchange counts/types
  they emit, so plan-shape equality is intentionally not checked.
- `test_local_shuffle_rqg_bugs.groovy` — reproduces 20+ RQG-found
  crashes and wrong-result cases.
- `test_old_coordinator_local_shuffle.groovy` — verifies the old
  coordinator path is unaffected.
- `test_multilevel_join_agg_local_shuffle.groovy` — multi-level join
  and aggregation plan shapes.

- `multi_version.h`: replace `atomic_load/atomic_store` (deprecated in
  libstdc++ C++20 / LLVM 20) with `std::shared_mutex`-based RW locking.
- `memory.cpp`: fix `std::max` type mismatch (`long` vs `int64_t`)
  on macOS.
- `bucketed_aggregation_sink_operator.h`: fix `ExchangeType::NOOP` →
  `TLocalPartitionType::NOOP` after thrift enum rename.

This PR puts the FE planner in the driver's seat for LE insertion but
intentionally does NOT remove the BE-side machinery — readers should be
aware of three pieces the FE planner shares with or defers to BE:

1. **`is_serial_operator` is computed on both sides.**  FE computes the
   flag and writes it into Thrift, but BE's
   `OperatorBase::is_serial_operator()` is still overridden per operator
   in C++ and used for BE-side runtime decisions.  Any future change to
   the BE override needs to be mirrored on the FE side (and vice versa)
   to keep the planner's view consistent with execution.

2. **The legacy BE planner stays as a fallback.**
   `pipeline_fragment_context.cpp::_plan_local_exchange` is preserved
   and gated by `runtime_state.h::plan_local_shuffle()`: when
   `enable_local_shuffle_planner=false`, BE plans LE itself, exactly as
   before.  The two paths are mutually exclusive, never both running on
   the same query.

3. **`_propagate_local_exchange_num_tasks` is kept as a runtime safety
   net.**  The two propagation passes in
   `pipeline_fragment_context.cpp` fix up paired pipelines whose
   `num_tasks` end up mismatched (e.g. when AGG/SORT/JOIN pipeline
   splits leave a serial Exchange feeding an N-task sink).  FE's
   framework-level serial→non-serial fan-out (`enforceRequire` step 3)
   and the `validateNoSerialWithoutLocalExchange` check aim to make
   these mismatches impossible by construction, but the BE-side fixup
   remains as a defensive guard.

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

Previously, local exchange (LE) nodes were inserted exclusively by the
BE's `_plan_local_exchange` at pipeline build time.  The FE had no
visibility into which operators needed a fan-out or shuffle before
execution, making it impossible to validate, optimize, or override LE
decisions at planning time.

This PR introduces a full FE-side local exchange planner that mirrors
BE semantics, brings several correctness fixes, and leaves the legacy
BE path fully intact behind a feature flag.  See "Current architecture
notes" at the bottom for what the FE planner does and does not own.

A new `AddLocalExchange` pass runs after normal fragment assignment.
It walks each fragment's plan tree bottom-up, calling the polymorphic
`PlanNode.enforceAndDeriveLocalExchange()` on every node.  Nodes
declare what distribution they require of their children; the framework
inserts `LocalExchangeNode` where needed.

`LocalExchangeNode` represents intra-fragment data redistribution and
supports the full set of exchange types: PASSTHROUGH, HASH_SHUFFLE,
BUCKET_HASH_SHUFFLE, GLOBAL_EXECUTION_HASH_SHUFFLE, BROADCAST,
PASS_TO_ONE, ADAPTIVE_PASSTHROUGH, LOCAL_MERGE_SORT, and NOOP.

The pass is guarded by `enable_local_shuffle_planner` (default true).
When disabled, BE continues to run its own `_plan_local_exchange` as
before, keeping the old path fully intact.

`maxPerBeInstances` (max pipeline instances assigned to any single BE)
is used instead of a global `instanceCount`.  Planning is a no-op when
`maxPerBeInstances == 1` — inserting LE on a single-threaded pipeline
would cause task-count mismatches and pipeline starvation.

When a serial operator (e.g. OlapScanNode with a single tablet bucket)
feeds a non-serial parent without an intermediate LE, downstream tasks
starve waiting for data that never arrives.  The framework detects this
case and inserts a PASSTHROUGH LE to restore N-task parallelism, exactly
matching BE's `required_data_distribution()` serial → PASSTHROUGH rule.

`LocalExchangeTypeRequire` abstracts two strategies:
- `RequireHash` — always resolves to `LOCAL_EXECUTION_HASH_SHUFFLE`
  (safe for intra-fragment hash partitioning).
- `RequireSpecific` — preserves BUCKET_HASH_SHUFFLE /
  GLOBAL_EXECUTION_HASH_SHUFFLE without degradation.

PR #62438 added `enable_local_exchange_before_agg`, but its BE guard
`!_needs_finalize && !enable_local_exchange_before_agg → base` conflated
two semantically different cases in AggSink and DistinctStreamingAgg:

- **AggSink**: `!finalize && hasKeys` covered both LOCAL preagg
  (performance-only) and FIRST_MERGE dedup (correctness-critical).
  The flag-gated early-return wrongly skipped HASH for FIRST_MERGE,
  producing PASSTHROUGH-over-serial-child → wrong aggregation results.

- **DistinctStreamingAgg**: `!finalize` covered both streaming preagg
  (`useStreamingPreagg=true`, performance) and non-streaming dedup
  (`useStreamingPreagg=false`, correctness).  Same class of bug.

FE fix:
- AggSink: restrict the flag-gated base path to `!isMerge()` LOCAL
  phases.  FIRST_MERGE always emits HASH regardless of the flag.
- DistinctStreamingAgg: restrict to `useStreamingPreagg=true`.
  Non-streaming dedup always emits HASH.

Also add `requiresShuffleForCorrectness()` to mirror BE's
`is_shuffled_operator()`, so SetOperationNode propagates the
"downstream depends on hash" flag correctly instead of using the
coarser `parentRequire.preferType().isHashShuffle()` check that
over-inserted HASH LE on every union branch under a streaming preagg.

These fixes reduce FE/BE consistency mismatches from 8 to 3
(only pre-existing NLJ optimization differences remain).

- `enable_local_shuffle_planner` — use FE planner (default true)
- `enable_local_shuffle` — master switch for local shuffle
- `enable_local_exchange_before_agg` — HASH LE before non-final agg
  (default true, mirrors #62438)

`validateNoSerialWithoutLocalExchange()` walks the final plan tree and
logs a warning whenever a serial operator feeds a non-serial parent
without an intermediate LocalExchangeNode, catching planning gaps
before execution.

- `test_enable_local_exchange_before_agg.groovy` — 10 agg patterns
  with the flag on and off; covers the FIRST_MERGE and
  DistinctStreamingAgg correctness fixes.
- `test_local_shuffle_fe_be_consistency.groovy` — runs the same SQL
  with `enable_local_shuffle_planner=true` and `=false` across the
  full operator matrix (Agg, Sort, Analytic, HashJoin, NLJ, Set, Union,
  TableFunction, AssertNumRows, RQG-derived corner cases) and asserts
  result rows are identical.  Only data correctness is asserted — the
  two planners legitimately differ on the exact exchange counts/types
  they emit, so plan-shape equality is intentionally not checked.
- `test_local_shuffle_rqg_bugs.groovy` — reproduces 20+ RQG-found
  crashes and wrong-result cases.
- `test_old_coordinator_local_shuffle.groovy` — verifies the old
  coordinator path is unaffected.
- `test_multilevel_join_agg_local_shuffle.groovy` — multi-level join
  and aggregation plan shapes.

- `multi_version.h`: replace `atomic_load/atomic_store` (deprecated in
  libstdc++ C++20 / LLVM 20) with `std::shared_mutex`-based RW locking.
- `memory.cpp`: fix `std::max` type mismatch (`long` vs `int64_t`)
  on macOS.
- `bucketed_aggregation_sink_operator.h`: fix `ExchangeType::NOOP` →
  `TLocalPartitionType::NOOP` after thrift enum rename.

This PR puts the FE planner in the driver's seat for LE insertion but
intentionally does NOT remove the BE-side machinery — readers should be
aware of three pieces the FE planner shares with or defers to BE:

1. **`is_serial_operator` is computed on both sides.**  FE computes the
   flag and writes it into Thrift, but BE's
   `OperatorBase::is_serial_operator()` is still overridden per operator
   in C++ and used for BE-side runtime decisions.  Any future change to
   the BE override needs to be mirrored on the FE side (and vice versa)
   to keep the planner's view consistent with execution.

2. **The legacy BE planner stays as a fallback.**
   `pipeline_fragment_context.cpp::_plan_local_exchange` is preserved
   and gated by `runtime_state.h::plan_local_shuffle()`: when
   `enable_local_shuffle_planner=false`, BE plans LE itself, exactly as
   before.  The two paths are mutually exclusive, never both running on
   the same query.

3. **`_propagate_local_exchange_num_tasks` is kept as a runtime safety
   net.**  The two propagation passes in
   `pipeline_fragment_context.cpp` fix up paired pipelines whose
   `num_tasks` end up mismatched (e.g. when AGG/SORT/JOIN pipeline
   splits leave a serial Exchange feeding an N-task sink).  FE's
   framework-level serial→non-serial fan-out (`enforceRequire` step 3)
   and the `validateNoSerialWithoutLocalExchange` check aim to make
   these mismatches impossible by construction, but the BE-side fixup
   remains as a defensive guard.

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

Previously, local exchange (LE) nodes were inserted exclusively by the
BE's `_plan_local_exchange` at pipeline build time.  The FE had no
visibility into which operators needed a fan-out or shuffle before
execution, making it impossible to validate, optimize, or override LE
decisions at planning time.

This PR introduces a full FE-side local exchange planner that mirrors
BE semantics, brings several correctness fixes, and leaves the legacy
BE path fully intact behind a feature flag.  See "Current architecture
notes" at the bottom for what the FE planner does and does not own.

A new `AddLocalExchange` pass runs after normal fragment assignment.
It walks each fragment's plan tree bottom-up, calling the polymorphic
`PlanNode.enforceAndDeriveLocalExchange()` on every node.  Nodes
declare what distribution they require of their children; the framework
inserts `LocalExchangeNode` where needed.

`LocalExchangeNode` represents intra-fragment data redistribution and
supports the full set of exchange types: PASSTHROUGH, HASH_SHUFFLE,
BUCKET_HASH_SHUFFLE, GLOBAL_EXECUTION_HASH_SHUFFLE, BROADCAST,
PASS_TO_ONE, ADAPTIVE_PASSTHROUGH, LOCAL_MERGE_SORT, and NOOP.

The pass is guarded by `enable_local_shuffle_planner` (default true).
When disabled, BE continues to run its own `_plan_local_exchange` as
before, keeping the old path fully intact.

`maxPerBeInstances` (max pipeline instances assigned to any single BE)
is used instead of a global `instanceCount`.  Planning is a no-op when
`maxPerBeInstances == 1` — inserting LE on a single-threaded pipeline
would cause task-count mismatches and pipeline starvation.

When a serial operator (e.g. OlapScanNode with a single tablet bucket)
feeds a non-serial parent without an intermediate LE, downstream tasks
starve waiting for data that never arrives.  The framework detects this
case and inserts a PASSTHROUGH LE to restore N-task parallelism, exactly
matching BE's `required_data_distribution()` serial → PASSTHROUGH rule.

`LocalExchangeTypeRequire` abstracts two strategies:
- `RequireHash` — always resolves to `LOCAL_EXECUTION_HASH_SHUFFLE`
  (safe for intra-fragment hash partitioning).
- `RequireSpecific` — preserves BUCKET_HASH_SHUFFLE /
  GLOBAL_EXECUTION_HASH_SHUFFLE without degradation.

PR #62438 added `enable_local_exchange_before_agg`, but its BE guard
`!_needs_finalize && !enable_local_exchange_before_agg → base` conflated
two semantically different cases in AggSink and DistinctStreamingAgg:

- **AggSink**: `!finalize && hasKeys` covered both LOCAL preagg
  (performance-only) and FIRST_MERGE dedup (correctness-critical).
  The flag-gated early-return wrongly skipped HASH for FIRST_MERGE,
  producing PASSTHROUGH-over-serial-child → wrong aggregation results.

- **DistinctStreamingAgg**: `!finalize` covered both streaming preagg
  (`useStreamingPreagg=true`, performance) and non-streaming dedup
  (`useStreamingPreagg=false`, correctness).  Same class of bug.

FE fix:
- AggSink: restrict the flag-gated base path to `!isMerge()` LOCAL
  phases.  FIRST_MERGE always emits HASH regardless of the flag.
- DistinctStreamingAgg: restrict to `useStreamingPreagg=true`.
  Non-streaming dedup always emits HASH.

Also add `requiresShuffleForCorrectness()` to mirror BE's
`is_shuffled_operator()`, so SetOperationNode propagates the
"downstream depends on hash" flag correctly instead of using the
coarser `parentRequire.preferType().isHashShuffle()` check that
over-inserted HASH LE on every union branch under a streaming preagg.

These fixes reduce FE/BE consistency mismatches from 8 to 3
(only pre-existing NLJ optimization differences remain).

- `enable_local_shuffle_planner` — use FE planner (default true)
- `enable_local_shuffle` — master switch for local shuffle
- `enable_local_exchange_before_agg` — HASH LE before non-final agg
  (default true, mirrors #62438)

`validateNoSerialWithoutLocalExchange()` walks the final plan tree and
logs a warning whenever a serial operator feeds a non-serial parent
without an intermediate LocalExchangeNode, catching planning gaps
before execution.

- `test_enable_local_exchange_before_agg.groovy` — 10 agg patterns
  with the flag on and off; covers the FIRST_MERGE and
  DistinctStreamingAgg correctness fixes.
- `test_local_shuffle_fe_be_consistency.groovy` — runs the same SQL
  with `enable_local_shuffle_planner=true` and `=false` across the
  full operator matrix (Agg, Sort, Analytic, HashJoin, NLJ, Set, Union,
  TableFunction, AssertNumRows, RQG-derived corner cases) and asserts
  result rows are identical.  Only data correctness is asserted — the
  two planners legitimately differ on the exact exchange counts/types
  they emit, so plan-shape equality is intentionally not checked.
- `test_local_shuffle_rqg_bugs.groovy` — reproduces 20+ RQG-found
  crashes and wrong-result cases.
- `test_old_coordinator_local_shuffle.groovy` — verifies the old
  coordinator path is unaffected.
- `test_multilevel_join_agg_local_shuffle.groovy` — multi-level join
  and aggregation plan shapes.

- `multi_version.h`: replace `atomic_load/atomic_store` (deprecated in
  libstdc++ C++20 / LLVM 20) with `std::shared_mutex`-based RW locking.
- `memory.cpp`: fix `std::max` type mismatch (`long` vs `int64_t`)
  on macOS.
- `bucketed_aggregation_sink_operator.h`: fix `ExchangeType::NOOP` →
  `TLocalPartitionType::NOOP` after thrift enum rename.

This PR puts the FE planner in the driver's seat for LE insertion but
intentionally does NOT remove the BE-side machinery — readers should be
aware of three pieces the FE planner shares with or defers to BE:

1. **`is_serial_operator` is computed on both sides.**  FE computes the
   flag and writes it into Thrift, but BE's
   `OperatorBase::is_serial_operator()` is still overridden per operator
   in C++ and used for BE-side runtime decisions.  Any future change to
   the BE override needs to be mirrored on the FE side (and vice versa)
   to keep the planner's view consistent with execution.

2. **The legacy BE planner stays as a fallback.**
   `pipeline_fragment_context.cpp::_plan_local_exchange` is preserved
   and gated by `runtime_state.h::plan_local_shuffle()`: when
   `enable_local_shuffle_planner=false`, BE plans LE itself, exactly as
   before.  The two paths are mutually exclusive, never both running on
   the same query.

3. **`_propagate_local_exchange_num_tasks` is kept as a runtime safety
   net.**  The two propagation passes in
   `pipeline_fragment_context.cpp` fix up paired pipelines whose
   `num_tasks` end up mismatched (e.g. when AGG/SORT/JOIN pipeline
   splits leave a serial Exchange feeding an N-task sink).  FE's
   framework-level serial→non-serial fan-out (`enforceRequire` step 3)
   and the `validateNoSerialWithoutLocalExchange` check aim to make
   these mismatches impossible by construction, but the BE-side fixup
   remains as a defensive guard.

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

Previously, local exchange (LE) nodes were inserted exclusively by the
BE's `_plan_local_exchange` at pipeline build time.  The FE had no
visibility into which operators needed a fan-out or shuffle before
execution, making it impossible to validate, optimize, or override LE
decisions at planning time.

This PR introduces a full FE-side local exchange planner that mirrors
BE semantics, brings several correctness fixes, and leaves the legacy
BE path fully intact behind a feature flag.  See "Current architecture
notes" at the bottom for what the FE planner does and does not own.

A new `AddLocalExchange` pass runs after normal fragment assignment.
It walks each fragment's plan tree bottom-up, calling the polymorphic
`PlanNode.enforceAndDeriveLocalExchange()` on every node.  Nodes
declare what distribution they require of their children; the framework
inserts `LocalExchangeNode` where needed.

`LocalExchangeNode` represents intra-fragment data redistribution and
supports the full set of exchange types: PASSTHROUGH, HASH_SHUFFLE,
BUCKET_HASH_SHUFFLE, GLOBAL_EXECUTION_HASH_SHUFFLE, BROADCAST,
PASS_TO_ONE, ADAPTIVE_PASSTHROUGH, LOCAL_MERGE_SORT, and NOOP.

The pass is guarded by `enable_local_shuffle_planner` (default true).
When disabled, BE continues to run its own `_plan_local_exchange` as
before, keeping the old path fully intact.

`maxPerBeInstances` (max pipeline instances assigned to any single BE)
is used instead of a global `instanceCount`.  Planning is a no-op when
`maxPerBeInstances == 1` — inserting LE on a single-threaded pipeline
would cause task-count mismatches and pipeline starvation.

When a serial operator (e.g. OlapScanNode with a single tablet bucket)
feeds a non-serial parent without an intermediate LE, downstream tasks
starve waiting for data that never arrives.  The framework detects this
case and inserts a PASSTHROUGH LE to restore N-task parallelism, exactly
matching BE's `required_data_distribution()` serial → PASSTHROUGH rule.

`LocalExchangeTypeRequire` abstracts two strategies:
- `RequireHash` — always resolves to `LOCAL_EXECUTION_HASH_SHUFFLE`
  (safe for intra-fragment hash partitioning).
- `RequireSpecific` — preserves BUCKET_HASH_SHUFFLE /
  GLOBAL_EXECUTION_HASH_SHUFFLE without degradation.

PR #62438 added `enable_local_exchange_before_agg`, but its BE guard
`!_needs_finalize && !enable_local_exchange_before_agg → base` conflated
two semantically different cases in AggSink and DistinctStreamingAgg:

- **AggSink**: `!finalize && hasKeys` covered both LOCAL preagg
  (performance-only) and FIRST_MERGE dedup (correctness-critical).
  The flag-gated early-return wrongly skipped HASH for FIRST_MERGE,
  producing PASSTHROUGH-over-serial-child → wrong aggregation results.

- **DistinctStreamingAgg**: `!finalize` covered both streaming preagg
  (`useStreamingPreagg=true`, performance) and non-streaming dedup
  (`useStreamingPreagg=false`, correctness).  Same class of bug.

FE fix:
- AggSink: restrict the flag-gated base path to `!isMerge()` LOCAL
  phases.  FIRST_MERGE always emits HASH regardless of the flag.
- DistinctStreamingAgg: restrict to `useStreamingPreagg=true`.
  Non-streaming dedup always emits HASH.

Also add `requiresShuffleForCorrectness()` to mirror BE's
`is_shuffled_operator()`, so SetOperationNode propagates the
"downstream depends on hash" flag correctly instead of using the
coarser `parentRequire.preferType().isHashShuffle()` check that
over-inserted HASH LE on every union branch under a streaming preagg.

These fixes reduce FE/BE consistency mismatches from 8 to 3
(only pre-existing NLJ optimization differences remain).

- `enable_local_shuffle_planner` — use FE planner (default true)
- `enable_local_shuffle` — master switch for local shuffle
- `enable_local_exchange_before_agg` — HASH LE before non-final agg
  (default true, mirrors #62438)

`validateNoSerialWithoutLocalExchange()` walks the final plan tree and
logs a warning whenever a serial operator feeds a non-serial parent
without an intermediate LocalExchangeNode, catching planning gaps
before execution.

- `test_enable_local_exchange_before_agg.groovy` — 10 agg patterns
  with the flag on and off; covers the FIRST_MERGE and
  DistinctStreamingAgg correctness fixes.
- `test_local_shuffle_fe_be_consistency.groovy` — runs the same SQL
  with `enable_local_shuffle_planner=true` and `=false` across the
  full operator matrix (Agg, Sort, Analytic, HashJoin, NLJ, Set, Union,
  TableFunction, AssertNumRows, RQG-derived corner cases) and asserts
  result rows are identical.  Only data correctness is asserted — the
  two planners legitimately differ on the exact exchange counts/types
  they emit, so plan-shape equality is intentionally not checked.
- `test_local_shuffle_rqg_bugs.groovy` — reproduces 20+ RQG-found
  crashes and wrong-result cases.
- `test_old_coordinator_local_shuffle.groovy` — verifies the old
  coordinator path is unaffected.
- `test_multilevel_join_agg_local_shuffle.groovy` — multi-level join
  and aggregation plan shapes.

- `multi_version.h`: replace `atomic_load/atomic_store` (deprecated in
  libstdc++ C++20 / LLVM 20) with `std::shared_mutex`-based RW locking.
- `memory.cpp`: fix `std::max` type mismatch (`long` vs `int64_t`)
  on macOS.
- `bucketed_aggregation_sink_operator.h`: fix `ExchangeType::NOOP` →
  `TLocalPartitionType::NOOP` after thrift enum rename.

This PR puts the FE planner in the driver's seat for LE insertion but
intentionally does NOT remove the BE-side machinery — readers should be
aware of three pieces the FE planner shares with or defers to BE:

1. **`is_serial_operator` is computed on both sides.**  FE computes the
   flag and writes it into Thrift, but BE's
   `OperatorBase::is_serial_operator()` is still overridden per operator
   in C++ and used for BE-side runtime decisions.  Any future change to
   the BE override needs to be mirrored on the FE side (and vice versa)
   to keep the planner's view consistent with execution.

2. **The legacy BE planner stays as a fallback.**
   `pipeline_fragment_context.cpp::_plan_local_exchange` is preserved
   and gated by `runtime_state.h::plan_local_shuffle()`: when
   `enable_local_shuffle_planner=false`, BE plans LE itself, exactly as
   before.  The two paths are mutually exclusive, never both running on
   the same query.

3. **`_propagate_local_exchange_num_tasks` is kept as a runtime safety
   net.**  The two propagation passes in
   `pipeline_fragment_context.cpp` fix up paired pipelines whose
   `num_tasks` end up mismatched (e.g. when AGG/SORT/JOIN pipeline
   splits leave a serial Exchange feeding an N-task sink).  FE's
   framework-level serial→non-serial fan-out (`enforceRequire` step 3)
   and the `validateNoSerialWithoutLocalExchange` check aim to make
   these mismatches impossible by construction, but the BE-side fixup
   remains as a defensive guard.

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

Previously, local exchange (LE) nodes were inserted exclusively by the
BE's `_plan_local_exchange` at pipeline build time.  The FE had no
visibility into which operators needed a fan-out or shuffle before
execution, making it impossible to validate, optimize, or override LE
decisions at planning time.

This PR introduces a full FE-side local exchange planner that mirrors
BE semantics, brings several correctness fixes, and leaves the legacy
BE path fully intact behind a feature flag.  See "Current architecture
notes" at the bottom for what the FE planner does and does not own.

A new `AddLocalExchange` pass runs after normal fragment assignment.
It walks each fragment's plan tree bottom-up, calling the polymorphic
`PlanNode.enforceAndDeriveLocalExchange()` on every node.  Nodes
declare what distribution they require of their children; the framework
inserts `LocalExchangeNode` where needed.

`LocalExchangeNode` represents intra-fragment data redistribution and
supports the full set of exchange types: PASSTHROUGH, HASH_SHUFFLE,
BUCKET_HASH_SHUFFLE, GLOBAL_EXECUTION_HASH_SHUFFLE, BROADCAST,
PASS_TO_ONE, ADAPTIVE_PASSTHROUGH, LOCAL_MERGE_SORT, and NOOP.

The pass is guarded by `enable_local_shuffle_planner` (default true).
When disabled, BE continues to run its own `_plan_local_exchange` as
before, keeping the old path fully intact.

`maxPerBeInstances` (max pipeline instances assigned to any single BE)
is used instead of a global `instanceCount`.  Planning is a no-op when
`maxPerBeInstances == 1` — inserting LE on a single-threaded pipeline
would cause task-count mismatches and pipeline starvation.

When a serial operator (e.g. OlapScanNode with a single tablet bucket)
feeds a non-serial parent without an intermediate LE, downstream tasks
starve waiting for data that never arrives.  The framework detects this
case and inserts a PASSTHROUGH LE to restore N-task parallelism, exactly
matching BE's `required_data_distribution()` serial → PASSTHROUGH rule.

`LocalExchangeTypeRequire` abstracts two strategies:
- `RequireHash` — always resolves to `LOCAL_EXECUTION_HASH_SHUFFLE`
  (safe for intra-fragment hash partitioning).
- `RequireSpecific` — preserves BUCKET_HASH_SHUFFLE /
  GLOBAL_EXECUTION_HASH_SHUFFLE without degradation.

PR #62438 added `enable_local_exchange_before_agg`, but its BE guard
`!_needs_finalize && !enable_local_exchange_before_agg → base` conflated
two semantically different cases in AggSink and DistinctStreamingAgg:

- **AggSink**: `!finalize && hasKeys` covered both LOCAL preagg
  (performance-only) and FIRST_MERGE dedup (correctness-critical).
  The flag-gated early-return wrongly skipped HASH for FIRST_MERGE,
  producing PASSTHROUGH-over-serial-child → wrong aggregation results.

- **DistinctStreamingAgg**: `!finalize` covered both streaming preagg
  (`useStreamingPreagg=true`, performance) and non-streaming dedup
  (`useStreamingPreagg=false`, correctness).  Same class of bug.

FE fix:
- AggSink: restrict the flag-gated base path to `!isMerge()` LOCAL
  phases.  FIRST_MERGE always emits HASH regardless of the flag.
- DistinctStreamingAgg: restrict to `useStreamingPreagg=true`.
  Non-streaming dedup always emits HASH.

Also add `requiresShuffleForCorrectness()` to mirror BE's
`is_shuffled_operator()`, so SetOperationNode propagates the
"downstream depends on hash" flag correctly instead of using the
coarser `parentRequire.preferType().isHashShuffle()` check that
over-inserted HASH LE on every union branch under a streaming preagg.

These fixes reduce FE/BE consistency mismatches from 8 to 3
(only pre-existing NLJ optimization differences remain).

- `enable_local_shuffle_planner` — use FE planner (default true)
- `enable_local_shuffle` — master switch for local shuffle
- `enable_local_exchange_before_agg` — HASH LE before non-final agg
  (default true, mirrors #62438)

`validateNoSerialWithoutLocalExchange()` walks the final plan tree and
logs a warning whenever a serial operator feeds a non-serial parent
without an intermediate LocalExchangeNode, catching planning gaps
before execution.

- `test_enable_local_exchange_before_agg.groovy` — 10 agg patterns
  with the flag on and off; covers the FIRST_MERGE and
  DistinctStreamingAgg correctness fixes.
- `test_local_shuffle_fe_be_consistency.groovy` — runs the same SQL
  with `enable_local_shuffle_planner=true` and `=false` across the
  full operator matrix (Agg, Sort, Analytic, HashJoin, NLJ, Set, Union,
  TableFunction, AssertNumRows, RQG-derived corner cases) and asserts
  result rows are identical.  Only data correctness is asserted — the
  two planners legitimately differ on the exact exchange counts/types
  they emit, so plan-shape equality is intentionally not checked.
- `test_local_shuffle_rqg_bugs.groovy` — reproduces 20+ RQG-found
  crashes and wrong-result cases.
- `test_old_coordinator_local_shuffle.groovy` — verifies the old
  coordinator path is unaffected.
- `test_multilevel_join_agg_local_shuffle.groovy` — multi-level join
  and aggregation plan shapes.

- `multi_version.h`: replace `atomic_load/atomic_store` (deprecated in
  libstdc++ C++20 / LLVM 20) with `std::shared_mutex`-based RW locking.
- `memory.cpp`: fix `std::max` type mismatch (`long` vs `int64_t`)
  on macOS.
- `bucketed_aggregation_sink_operator.h`: fix `ExchangeType::NOOP` →
  `TLocalPartitionType::NOOP` after thrift enum rename.

This PR puts the FE planner in the driver's seat for LE insertion but
intentionally does NOT remove the BE-side machinery — readers should be
aware of three pieces the FE planner shares with or defers to BE:

1. **`is_serial_operator` is computed on both sides.**  FE computes the
   flag and writes it into Thrift, but BE's
   `OperatorBase::is_serial_operator()` is still overridden per operator
   in C++ and used for BE-side runtime decisions.  Any future change to
   the BE override needs to be mirrored on the FE side (and vice versa)
   to keep the planner's view consistent with execution.

2. **The legacy BE planner stays as a fallback.**
   `pipeline_fragment_context.cpp::_plan_local_exchange` is preserved
   and gated by `runtime_state.h::plan_local_shuffle()`: when
   `enable_local_shuffle_planner=false`, BE plans LE itself, exactly as
   before.  The two paths are mutually exclusive, never both running on
   the same query.

3. **`_propagate_local_exchange_num_tasks` is kept as a runtime safety
   net.**  The two propagation passes in
   `pipeline_fragment_context.cpp` fix up paired pipelines whose
   `num_tasks` end up mismatched (e.g. when AGG/SORT/JOIN pipeline
   splits leave a serial Exchange feeding an N-task sink).  FE's
   framework-level serial→non-serial fan-out (`enforceRequire` step 3)
   and the `validateNoSerialWithoutLocalExchange` check aim to make
   these mismatches impossible by construction, but the BE-side fixup
   remains as a defensive guard.

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

Previously, local exchange (LE) nodes were inserted exclusively by the
BE's `_plan_local_exchange` at pipeline build time.  The FE had no
visibility into which operators needed a fan-out or shuffle before
execution, making it impossible to validate, optimize, or override LE
decisions at planning time.

This PR introduces a full FE-side local exchange planner that mirrors
BE semantics, brings several correctness fixes, and leaves the legacy
BE path fully intact behind a feature flag.  See "Current architecture
notes" at the bottom for what the FE planner does and does not own.

A new `AddLocalExchange` pass runs after normal fragment assignment.
It walks each fragment's plan tree bottom-up, calling the polymorphic
`PlanNode.enforceAndDeriveLocalExchange()` on every node.  Nodes
declare what distribution they require of their children; the framework
inserts `LocalExchangeNode` where needed.

`LocalExchangeNode` represents intra-fragment data redistribution and
supports the full set of exchange types: PASSTHROUGH, HASH_SHUFFLE,
BUCKET_HASH_SHUFFLE, GLOBAL_EXECUTION_HASH_SHUFFLE, BROADCAST,
PASS_TO_ONE, ADAPTIVE_PASSTHROUGH, LOCAL_MERGE_SORT, and NOOP.

The pass is guarded by `enable_local_shuffle_planner` (default true).
When disabled, BE continues to run its own `_plan_local_exchange` as
before, keeping the old path fully intact.

`maxPerBeInstances` (max pipeline instances assigned to any single BE)
is used instead of a global `instanceCount`.  Planning is a no-op when
`maxPerBeInstances == 1` — inserting LE on a single-threaded pipeline
would cause task-count mismatches and pipeline starvation.

When a serial operator (e.g. OlapScanNode with a single tablet bucket)
feeds a non-serial parent without an intermediate LE, downstream tasks
starve waiting for data that never arrives.  The framework detects this
case and inserts a PASSTHROUGH LE to restore N-task parallelism, exactly
matching BE's `required_data_distribution()` serial → PASSTHROUGH rule.

`LocalExchangeTypeRequire` abstracts two strategies:
- `RequireHash` — always resolves to `LOCAL_EXECUTION_HASH_SHUFFLE`
  (safe for intra-fragment hash partitioning).
- `RequireSpecific` — preserves BUCKET_HASH_SHUFFLE /
  GLOBAL_EXECUTION_HASH_SHUFFLE without degradation.

PR #62438 added `enable_local_exchange_before_agg`, but its BE guard
`!_needs_finalize && !enable_local_exchange_before_agg → base` conflated
two semantically different cases in AggSink and DistinctStreamingAgg:

- **AggSink**: `!finalize && hasKeys` covered both LOCAL preagg
  (performance-only) and FIRST_MERGE dedup (correctness-critical).
  The flag-gated early-return wrongly skipped HASH for FIRST_MERGE,
  producing PASSTHROUGH-over-serial-child → wrong aggregation results.

- **DistinctStreamingAgg**: `!finalize` covered both streaming preagg
  (`useStreamingPreagg=true`, performance) and non-streaming dedup
  (`useStreamingPreagg=false`, correctness).  Same class of bug.

FE fix:
- AggSink: restrict the flag-gated base path to `!isMerge()` LOCAL
  phases.  FIRST_MERGE always emits HASH regardless of the flag.
- DistinctStreamingAgg: restrict to `useStreamingPreagg=true`.
  Non-streaming dedup always emits HASH.

Also add `requiresShuffleForCorrectness()` to mirror BE's
`is_shuffled_operator()`, so SetOperationNode propagates the
"downstream depends on hash" flag correctly instead of using the
coarser `parentRequire.preferType().isHashShuffle()` check that
over-inserted HASH LE on every union branch under a streaming preagg.

These fixes reduce FE/BE consistency mismatches from 8 to 3
(only pre-existing NLJ optimization differences remain).

- `enable_local_shuffle_planner` — use FE planner (default true)
- `enable_local_shuffle` — master switch for local shuffle
- `enable_local_exchange_before_agg` — HASH LE before non-final agg
  (default true, mirrors #62438)

`validateNoSerialWithoutLocalExchange()` walks the final plan tree and
logs a warning whenever a serial operator feeds a non-serial parent
without an intermediate LocalExchangeNode, catching planning gaps
before execution.

- `test_enable_local_exchange_before_agg.groovy` — 10 agg patterns
  with the flag on and off; covers the FIRST_MERGE and
  DistinctStreamingAgg correctness fixes.
- `test_local_shuffle_fe_be_consistency.groovy` — runs the same SQL
  with `enable_local_shuffle_planner=true` and `=false` across the
  full operator matrix (Agg, Sort, Analytic, HashJoin, NLJ, Set, Union,
  TableFunction, AssertNumRows, RQG-derived corner cases) and asserts
  result rows are identical.  Only data correctness is asserted — the
  two planners legitimately differ on the exact exchange counts/types
  they emit, so plan-shape equality is intentionally not checked.
- `test_local_shuffle_rqg_bugs.groovy` — reproduces 20+ RQG-found
  crashes and wrong-result cases.
- `test_old_coordinator_local_shuffle.groovy` — verifies the old
  coordinator path is unaffected.
- `test_multilevel_join_agg_local_shuffle.groovy` — multi-level join
  and aggregation plan shapes.

- `multi_version.h`: replace `atomic_load/atomic_store` (deprecated in
  libstdc++ C++20 / LLVM 20) with `std::shared_mutex`-based RW locking.
- `memory.cpp`: fix `std::max` type mismatch (`long` vs `int64_t`)
  on macOS.
- `bucketed_aggregation_sink_operator.h`: fix `ExchangeType::NOOP` →
  `TLocalPartitionType::NOOP` after thrift enum rename.

This PR puts the FE planner in the driver's seat for LE insertion but
intentionally does NOT remove the BE-side machinery — readers should be
aware of three pieces the FE planner shares with or defers to BE:

1. **`is_serial_operator` is computed on both sides.**  FE computes the
   flag and writes it into Thrift, but BE's
   `OperatorBase::is_serial_operator()` is still overridden per operator
   in C++ and used for BE-side runtime decisions.  Any future change to
   the BE override needs to be mirrored on the FE side (and vice versa)
   to keep the planner's view consistent with execution.

2. **The legacy BE planner stays as a fallback.**
   `pipeline_fragment_context.cpp::_plan_local_exchange` is preserved
   and gated by `runtime_state.h::plan_local_shuffle()`: when
   `enable_local_shuffle_planner=false`, BE plans LE itself, exactly as
   before.  The two paths are mutually exclusive, never both running on
   the same query.

3. **`_propagate_local_exchange_num_tasks` is kept as a runtime safety
   net.**  The two propagation passes in
   `pipeline_fragment_context.cpp` fix up paired pipelines whose
   `num_tasks` end up mismatched (e.g. when AGG/SORT/JOIN pipeline
   splits leave a serial Exchange feeding an N-task sink).  FE's
   framework-level serial→non-serial fan-out (`enforceRequire` step 3)
   and the `validateNoSerialWithoutLocalExchange` check aim to make
   these mismatches impossible by construction, but the BE-side fixup
   remains as a defensive guard.

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

Previously, local exchange (LE) nodes were inserted exclusively by the
BE's `_plan_local_exchange` at pipeline build time.  The FE had no
visibility into which operators needed a fan-out or shuffle before
execution, making it impossible to validate, optimize, or override LE
decisions at planning time.

This PR introduces a full FE-side local exchange planner that mirrors
BE semantics, brings several correctness fixes, and leaves the legacy
BE path fully intact behind a feature flag.  See "Current architecture
notes" at the bottom for what the FE planner does and does not own.

A new `AddLocalExchange` pass runs after normal fragment assignment.
It walks each fragment's plan tree bottom-up, calling the polymorphic
`PlanNode.enforceAndDeriveLocalExchange()` on every node.  Nodes
declare what distribution they require of their children; the framework
inserts `LocalExchangeNode` where needed.

`LocalExchangeNode` represents intra-fragment data redistribution and
supports the full set of exchange types: PASSTHROUGH, HASH_SHUFFLE,
BUCKET_HASH_SHUFFLE, GLOBAL_EXECUTION_HASH_SHUFFLE, BROADCAST,
PASS_TO_ONE, ADAPTIVE_PASSTHROUGH, LOCAL_MERGE_SORT, and NOOP.

The pass is guarded by `enable_local_shuffle_planner` (default true).
When disabled, BE continues to run its own `_plan_local_exchange` as
before, keeping the old path fully intact.

`maxPerBeInstances` (max pipeline instances assigned to any single BE)
is used instead of a global `instanceCount`.  Planning is a no-op when
`maxPerBeInstances == 1` — inserting LE on a single-threaded pipeline
would cause task-count mismatches and pipeline starvation.

When a serial operator (e.g. OlapScanNode with a single tablet bucket)
feeds a non-serial parent without an intermediate LE, downstream tasks
starve waiting for data that never arrives.  The framework detects this
case and inserts a PASSTHROUGH LE to restore N-task parallelism, exactly
matching BE's `required_data_distribution()` serial → PASSTHROUGH rule.

`LocalExchangeTypeRequire` abstracts two strategies:
- `RequireHash` — always resolves to `LOCAL_EXECUTION_HASH_SHUFFLE`
  (safe for intra-fragment hash partitioning).
- `RequireSpecific` — preserves BUCKET_HASH_SHUFFLE /
  GLOBAL_EXECUTION_HASH_SHUFFLE without degradation.

PR #62438 added `enable_local_exchange_before_agg`, but its BE guard
`!_needs_finalize && !enable_local_exchange_before_agg → base` conflated
two semantically different cases in AggSink and DistinctStreamingAgg:

- **AggSink**: `!finalize && hasKeys` covered both LOCAL preagg
  (performance-only) and FIRST_MERGE dedup (correctness-critical).
  The flag-gated early-return wrongly skipped HASH for FIRST_MERGE,
  producing PASSTHROUGH-over-serial-child → wrong aggregation results.

- **DistinctStreamingAgg**: `!finalize` covered both streaming preagg
  (`useStreamingPreagg=true`, performance) and non-streaming dedup
  (`useStreamingPreagg=false`, correctness).  Same class of bug.

FE fix:
- AggSink: restrict the flag-gated base path to `!isMerge()` LOCAL
  phases.  FIRST_MERGE always emits HASH regardless of the flag.
- DistinctStreamingAgg: restrict to `useStreamingPreagg=true`.
  Non-streaming dedup always emits HASH.

Also add `requiresShuffleForCorrectness()` to mirror BE's
`is_shuffled_operator()`, so SetOperationNode propagates the
"downstream depends on hash" flag correctly instead of using the
coarser `parentRequire.preferType().isHashShuffle()` check that
over-inserted HASH LE on every union branch under a streaming preagg.

These fixes reduce FE/BE consistency mismatches from 8 to 3
(only pre-existing NLJ optimization differences remain).

- `enable_local_shuffle_planner` — use FE planner (default true)
- `enable_local_shuffle` — master switch for local shuffle
- `enable_local_exchange_before_agg` — HASH LE before non-final agg
  (default true, mirrors #62438)

`validateNoSerialWithoutLocalExchange()` walks the final plan tree and
logs a warning whenever a serial operator feeds a non-serial parent
without an intermediate LocalExchangeNode, catching planning gaps
before execution.

- `test_enable_local_exchange_before_agg.groovy` — 10 agg patterns
  with the flag on and off; covers the FIRST_MERGE and
  DistinctStreamingAgg correctness fixes.
- `test_local_shuffle_fe_be_consistency.groovy` — runs the same SQL
  with `enable_local_shuffle_planner=true` and `=false` across the
  full operator matrix (Agg, Sort, Analytic, HashJoin, NLJ, Set, Union,
  TableFunction, AssertNumRows, RQG-derived corner cases) and asserts
  result rows are identical.  Only data correctness is asserted — the
  two planners legitimately differ on the exact exchange counts/types
  they emit, so plan-shape equality is intentionally not checked.
- `test_local_shuffle_rqg_bugs.groovy` — reproduces 20+ RQG-found
  crashes and wrong-result cases.
- `test_old_coordinator_local_shuffle.groovy` — verifies the old
  coordinator path is unaffected.
- `test_multilevel_join_agg_local_shuffle.groovy` — multi-level join
  and aggregation plan shapes.

- `multi_version.h`: replace `atomic_load/atomic_store` (deprecated in
  libstdc++ C++20 / LLVM 20) with `std::shared_mutex`-based RW locking.
- `memory.cpp`: fix `std::max` type mismatch (`long` vs `int64_t`)
  on macOS.
- `bucketed_aggregation_sink_operator.h`: fix `ExchangeType::NOOP` →
  `TLocalPartitionType::NOOP` after thrift enum rename.

This PR puts the FE planner in the driver's seat for LE insertion but
intentionally does NOT remove the BE-side machinery — readers should be
aware of three pieces the FE planner shares with or defers to BE:

1. **`is_serial_operator` is computed on both sides.**  FE computes the
   flag and writes it into Thrift, but BE's
   `OperatorBase::is_serial_operator()` is still overridden per operator
   in C++ and used for BE-side runtime decisions.  Any future change to
   the BE override needs to be mirrored on the FE side (and vice versa)
   to keep the planner's view consistent with execution.

2. **The legacy BE planner stays as a fallback.**
   `pipeline_fragment_context.cpp::_plan_local_exchange` is preserved
   and gated by `runtime_state.h::plan_local_shuffle()`: when
   `enable_local_shuffle_planner=false`, BE plans LE itself, exactly as
   before.  The two paths are mutually exclusive, never both running on
   the same query.

3. **`_propagate_local_exchange_num_tasks` is kept as a runtime safety
   net.**  The two propagation passes in
   `pipeline_fragment_context.cpp` fix up paired pipelines whose
   `num_tasks` end up mismatched (e.g. when AGG/SORT/JOIN pipeline
   splits leave a serial Exchange feeding an N-task sink).  FE's
   framework-level serial→non-serial fan-out (`enforceRequire` step 3)
   and the `validateNoSerialWithoutLocalExchange` check aim to make
   these mismatches impossible by construction, but the BE-side fixup
   remains as a defensive guard.

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

Previously, local exchange (LE) nodes were inserted exclusively by the
BE's `_plan_local_exchange` at pipeline build time.  The FE had no
visibility into which operators needed a fan-out or shuffle before
execution, making it impossible to validate, optimize, or override LE
decisions at planning time.

This PR introduces a full FE-side local exchange planner that mirrors
BE semantics, brings several correctness fixes, and leaves the legacy
BE path fully intact behind a feature flag.  See "Current architecture
notes" at the bottom for what the FE planner does and does not own.

A new `AddLocalExchange` pass runs after normal fragment assignment.
It walks each fragment's plan tree bottom-up, calling the polymorphic
`PlanNode.enforceAndDeriveLocalExchange()` on every node.  Nodes
declare what distribution they require of their children; the framework
inserts `LocalExchangeNode` where needed.

`LocalExchangeNode` represents intra-fragment data redistribution and
supports the full set of exchange types: PASSTHROUGH, HASH_SHUFFLE,
BUCKET_HASH_SHUFFLE, GLOBAL_EXECUTION_HASH_SHUFFLE, BROADCAST,
PASS_TO_ONE, ADAPTIVE_PASSTHROUGH, LOCAL_MERGE_SORT, and NOOP.

The pass is guarded by `enable_local_shuffle_planner` (default true).
When disabled, BE continues to run its own `_plan_local_exchange` as
before, keeping the old path fully intact.

`maxPerBeInstances` (max pipeline instances assigned to any single BE)
is used instead of a global `instanceCount`.  Planning is a no-op when
`maxPerBeInstances == 1` — inserting LE on a single-threaded pipeline
would cause task-count mismatches and pipeline starvation.

When a serial operator (e.g. OlapScanNode with a single tablet bucket)
feeds a non-serial parent without an intermediate LE, downstream tasks
starve waiting for data that never arrives.  The framework detects this
case and inserts a PASSTHROUGH LE to restore N-task parallelism, exactly
matching BE's `required_data_distribution()` serial → PASSTHROUGH rule.

`LocalExchangeTypeRequire` abstracts two strategies:
- `RequireHash` — always resolves to `LOCAL_EXECUTION_HASH_SHUFFLE`
  (safe for intra-fragment hash partitioning).
- `RequireSpecific` — preserves BUCKET_HASH_SHUFFLE /
  GLOBAL_EXECUTION_HASH_SHUFFLE without degradation.

PR #62438 added `enable_local_exchange_before_agg`, but its BE guard
`!_needs_finalize && !enable_local_exchange_before_agg → base` conflated
two semantically different cases in AggSink and DistinctStreamingAgg:

- **AggSink**: `!finalize && hasKeys` covered both LOCAL preagg
  (performance-only) and FIRST_MERGE dedup (correctness-critical).
  The flag-gated early-return wrongly skipped HASH for FIRST_MERGE,
  producing PASSTHROUGH-over-serial-child → wrong aggregation results.

- **DistinctStreamingAgg**: `!finalize` covered both streaming preagg
  (`useStreamingPreagg=true`, performance) and non-streaming dedup
  (`useStreamingPreagg=false`, correctness).  Same class of bug.

FE fix:
- AggSink: restrict the flag-gated base path to `!isMerge()` LOCAL
  phases.  FIRST_MERGE always emits HASH regardless of the flag.
- DistinctStreamingAgg: restrict to `useStreamingPreagg=true`.
  Non-streaming dedup always emits HASH.

Also add `requiresShuffleForCorrectness()` to mirror BE's
`is_shuffled_operator()`, so SetOperationNode propagates the
"downstream depends on hash" flag correctly instead of using the
coarser `parentRequire.preferType().isHashShuffle()` check that
over-inserted HASH LE on every union branch under a streaming preagg.

These fixes reduce FE/BE consistency mismatches from 8 to 3
(only pre-existing NLJ optimization differences remain).

- `enable_local_shuffle_planner` — use FE planner (default true)
- `enable_local_shuffle` — master switch for local shuffle
- `enable_local_exchange_before_agg` — HASH LE before non-final agg
  (default true, mirrors #62438)

`validateNoSerialWithoutLocalExchange()` walks the final plan tree and
logs a warning whenever a serial operator feeds a non-serial parent
without an intermediate LocalExchangeNode, catching planning gaps
before execution.

- `test_enable_local_exchange_before_agg.groovy` — 10 agg patterns
  with the flag on and off; covers the FIRST_MERGE and
  DistinctStreamingAgg correctness fixes.
- `test_local_shuffle_fe_be_consistency.groovy` — runs the same SQL
  with `enable_local_shuffle_planner=true` and `=false` across the
  full operator matrix (Agg, Sort, Analytic, HashJoin, NLJ, Set, Union,
  TableFunction, AssertNumRows, RQG-derived corner cases) and asserts
  result rows are identical.  Only data correctness is asserted — the
  two planners legitimately differ on the exact exchange counts/types
  they emit, so plan-shape equality is intentionally not checked.
- `test_local_shuffle_rqg_bugs.groovy` — reproduces 20+ RQG-found
  crashes and wrong-result cases.
- `test_old_coordinator_local_shuffle.groovy` — verifies the old
  coordinator path is unaffected.
- `test_multilevel_join_agg_local_shuffle.groovy` — multi-level join
  and aggregation plan shapes.

- `multi_version.h`: replace `atomic_load/atomic_store` (deprecated in
  libstdc++ C++20 / LLVM 20) with `std::shared_mutex`-based RW locking.
- `memory.cpp`: fix `std::max` type mismatch (`long` vs `int64_t`)
  on macOS.
- `bucketed_aggregation_sink_operator.h`: fix `ExchangeType::NOOP` →
  `TLocalPartitionType::NOOP` after thrift enum rename.

This PR puts the FE planner in the driver's seat for LE insertion but
intentionally does NOT remove the BE-side machinery — readers should be
aware of three pieces the FE planner shares with or defers to BE:

1. **`is_serial_operator` is computed on both sides.**  FE computes the
   flag and writes it into Thrift, but BE's
   `OperatorBase::is_serial_operator()` is still overridden per operator
   in C++ and used for BE-side runtime decisions.  Any future change to
   the BE override needs to be mirrored on the FE side (and vice versa)
   to keep the planner's view consistent with execution.

2. **The legacy BE planner stays as a fallback.**
   `pipeline_fragment_context.cpp::_plan_local_exchange` is preserved
   and gated by `runtime_state.h::plan_local_shuffle()`: when
   `enable_local_shuffle_planner=false`, BE plans LE itself, exactly as
   before.  The two paths are mutually exclusive, never both running on
   the same query.

3. **`_propagate_local_exchange_num_tasks` is kept as a runtime safety
   net.**  The two propagation passes in
   `pipeline_fragment_context.cpp` fix up paired pipelines whose
   `num_tasks` end up mismatched (e.g. when AGG/SORT/JOIN pipeline
   splits leave a serial Exchange feeding an N-task sink).  FE's
   framework-level serial→non-serial fan-out (`enforceRequire` step 3)
   and the `validateNoSerialWithoutLocalExchange` check aim to make
   these mismatches impossible by construction, but the BE-side fixup
   remains as a defensive guard.

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

Problem Summary: PR #62438 added `enable_local_exchange_before_agg` to
allow skipping local exchange before non-finalizing aggregation. That
optimization used `!_needs_finalize` as the condition, but
non-finalizing aggregation includes both first-phase update/serialize
aggregation and merge/serialize aggregation.

When `experimental_use_serial_exchange` is enabled and
`enable_local_exchange_before_agg` is disabled, a serial exchange source
can be followed by the default `PASSTHROUGH` local exchange before a
non-finalizing merge aggregation. For DISTINCT aggregation, the merge
aggregation is the stage that deduplicates distinct keys after hash
exchange. `PASSTHROUGH` only restores local parallelism and does not
preserve key distribution, so duplicate distinct keys can be processed
by different local tasks and later partial sums can produce incorrect
results.

This PR keeps the knob behavior for aggregation stages that can safely
skip the local hash exchange, but excludes merge aggregation with a
serial child from the skip path. That case falls through to the existing
`HASH_SHUFFLE` / `BUCKET_HASH_SHUFFLE` distribution requirement. The PR
also computes the merge flag during `AggSinkOperatorX::init()` because
local exchange planning runs before `prepare()`.

### Release note

Fix occasional incorrect DISTINCT aggregate results when serial exchange
is enabled.

### What problem does this PR solve?

Issue Number: None

Related PR: apache#63529, apache#62438

Problem Summary: A hash-distributed aggregation can receive input after a non-hash local exchange, such as the adaptive passthrough exchanges planned around nested loop joins. The downstream aggregation still relies on key distribution for distinct aggregation, but the non-hash local exchange can split identical keys across local aggregation instances. This caused wrong count distinct results. This change lets local exchange sources expose their actual exchange distribution and lets aggregations force a new local hash shuffle when their child breaks local key distribution.

### Release note

Fix wrong results for distinct aggregation after non-hash local exchange.

### Check List (For Author)

- Test: Regression test / Unit Test / Manual test
    - ./run-be-ut.sh --run --filter=AggOperatorRequiredDistributionTest.*
    - ./run-regression-test.sh --run -d query_p0/join -s test_agg_after_nested_loop_join_local_exchange
    - Manual reproduction from output/ddl.txt returned 10 5070261
- Behavior changed: Yes. Aggregation now preserves required hash distribution when its child local exchange uses a non-hash distribution.
- Does this need documentation: No

Co-authored-by: Copilot <[email protected]>

Related PR: #63529, #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 #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.

Previously, local exchange (LE) nodes were inserted exclusively by the
BE's `_plan_local_exchange` at pipeline build time.  The FE had no
visibility into which operators needed a fan-out or shuffle before
execution, making it impossible to validate, optimize, or override LE
decisions at planning time.

This PR introduces a full FE-side local exchange planner that mirrors
BE semantics, brings several correctness fixes, and leaves the legacy
BE path fully intact behind a feature flag.  See "Current architecture
notes" at the bottom for what the FE planner does and does not own.

A new `AddLocalExchange` pass runs after normal fragment assignment.
It walks each fragment's plan tree bottom-up, calling the polymorphic
`PlanNode.enforceAndDeriveLocalExchange()` on every node.  Nodes
declare what distribution they require of their children; the framework
inserts `LocalExchangeNode` where needed.

`LocalExchangeNode` represents intra-fragment data redistribution and
supports the full set of exchange types: PASSTHROUGH, HASH_SHUFFLE,
BUCKET_HASH_SHUFFLE, GLOBAL_EXECUTION_HASH_SHUFFLE, BROADCAST,
PASS_TO_ONE, ADAPTIVE_PASSTHROUGH, LOCAL_MERGE_SORT, and NOOP.

The pass is guarded by `enable_local_shuffle_planner` (default true).
When disabled, BE continues to run its own `_plan_local_exchange` as
before, keeping the old path fully intact.

`maxPerBeInstances` (max pipeline instances assigned to any single BE)
is used instead of a global `instanceCount`.  Planning is a no-op when
`maxPerBeInstances == 1` — inserting LE on a single-threaded pipeline
would cause task-count mismatches and pipeline starvation.

When a serial operator (e.g. OlapScanNode with a single tablet bucket)
feeds a non-serial parent without an intermediate LE, downstream tasks
starve waiting for data that never arrives.  The framework detects this
case and inserts a PASSTHROUGH LE to restore N-task parallelism, exactly
matching BE's `required_data_distribution()` serial → PASSTHROUGH rule.

`LocalExchangeTypeRequire` abstracts two strategies:
- `RequireHash` — always resolves to `LOCAL_EXECUTION_HASH_SHUFFLE`
  (safe for intra-fragment hash partitioning).
- `RequireSpecific` — preserves BUCKET_HASH_SHUFFLE /
  GLOBAL_EXECUTION_HASH_SHUFFLE without degradation.

PR #62438 added `enable_local_exchange_before_agg`, but its BE guard
`!_needs_finalize && !enable_local_exchange_before_agg → base` conflated
two semantically different cases in AggSink and DistinctStreamingAgg:

- **AggSink**: `!finalize && hasKeys` covered both LOCAL preagg
  (performance-only) and FIRST_MERGE dedup (correctness-critical).
  The flag-gated early-return wrongly skipped HASH for FIRST_MERGE,
  producing PASSTHROUGH-over-serial-child → wrong aggregation results.

- **DistinctStreamingAgg**: `!finalize` covered both streaming preagg
  (`useStreamingPreagg=true`, performance) and non-streaming dedup
  (`useStreamingPreagg=false`, correctness).  Same class of bug.

FE fix:
- AggSink: restrict the flag-gated base path to `!isMerge()` LOCAL
  phases.  FIRST_MERGE always emits HASH regardless of the flag.
- DistinctStreamingAgg: restrict to `useStreamingPreagg=true`.
  Non-streaming dedup always emits HASH.

Also add `requiresShuffleForCorrectness()` to mirror BE's
`is_shuffled_operator()`, so SetOperationNode propagates the
"downstream depends on hash" flag correctly instead of using the
coarser `parentRequire.preferType().isHashShuffle()` check that
over-inserted HASH LE on every union branch under a streaming preagg.

These fixes reduce FE/BE consistency mismatches from 8 to 3
(only pre-existing NLJ optimization differences remain).

- `enable_local_shuffle_planner` — use FE planner (default true)
- `enable_local_shuffle` — master switch for local shuffle
- `enable_local_exchange_before_agg` — HASH LE before non-final agg
  (default true, mirrors #62438)

`validateNoSerialWithoutLocalExchange()` walks the final plan tree and
logs a warning whenever a serial operator feeds a non-serial parent
without an intermediate LocalExchangeNode, catching planning gaps
before execution.

- `test_enable_local_exchange_before_agg.groovy` — 10 agg patterns
  with the flag on and off; covers the FIRST_MERGE and
  DistinctStreamingAgg correctness fixes.
- `test_local_shuffle_fe_be_consistency.groovy` — runs the same SQL
  with `enable_local_shuffle_planner=true` and `=false` across the
  full operator matrix (Agg, Sort, Analytic, HashJoin, NLJ, Set, Union,
  TableFunction, AssertNumRows, RQG-derived corner cases) and asserts
  result rows are identical.  Only data correctness is asserted — the
  two planners legitimately differ on the exact exchange counts/types
  they emit, so plan-shape equality is intentionally not checked.
- `test_local_shuffle_rqg_bugs.groovy` — reproduces 20+ RQG-found
  crashes and wrong-result cases.
- `test_old_coordinator_local_shuffle.groovy` — verifies the old
  coordinator path is unaffected.
- `test_multilevel_join_agg_local_shuffle.groovy` — multi-level join
  and aggregation plan shapes.

- `multi_version.h`: replace `atomic_load/atomic_store` (deprecated in
  libstdc++ C++20 / LLVM 20) with `std::shared_mutex`-based RW locking.
- `memory.cpp`: fix `std::max` type mismatch (`long` vs `int64_t`)
  on macOS.
- `bucketed_aggregation_sink_operator.h`: fix `ExchangeType::NOOP` →
  `TLocalPartitionType::NOOP` after thrift enum rename.

This PR puts the FE planner in the driver's seat for LE insertion but
intentionally does NOT remove the BE-side machinery — readers should be
aware of three pieces the FE planner shares with or defers to BE:

1. **`is_serial_operator` is computed on both sides.**  FE computes the
   flag and writes it into Thrift, but BE's
   `OperatorBase::is_serial_operator()` is still overridden per operator
   in C++ and used for BE-side runtime decisions.  Any future change to
   the BE override needs to be mirrored on the FE side (and vice versa)
   to keep the planner's view consistent with execution.

2. **The legacy BE planner stays as a fallback.**
   `pipeline_fragment_context.cpp::_plan_local_exchange` is preserved
   and gated by `runtime_state.h::plan_local_shuffle()`: when
   `enable_local_shuffle_planner=false`, BE plans LE itself, exactly as
   before.  The two paths are mutually exclusive, never both running on
   the same query.

3. **`_propagate_local_exchange_num_tasks` is kept as a runtime safety
   net.**  The two propagation passes in
   `pipeline_fragment_context.cpp` fix up paired pipelines whose
   `num_tasks` end up mismatched (e.g. when AGG/SORT/JOIN pipeline
   splits leave a serial Exchange feeding an N-task sink).  FE's
   framework-level serial→non-serial fan-out (`enforceRequire` step 3)
   and the `validateNoSerialWithoutLocalExchange` check aim to make
   these mismatches impossible by construction, but the BE-side fixup
   remains as a defensive guard.

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

Problem Summary: PR apache#62438 added `enable_local_exchange_before_agg` to
allow skipping local exchange before non-finalizing aggregation. That
optimization used `!_needs_finalize` as the condition, but
non-finalizing aggregation includes both first-phase update/serialize
aggregation and merge/serialize aggregation.

When `experimental_use_serial_exchange` is enabled and
`enable_local_exchange_before_agg` is disabled, a serial exchange source
can be followed by the default `PASSTHROUGH` local exchange before a
non-finalizing merge aggregation. For DISTINCT aggregation, the merge
aggregation is the stage that deduplicates distinct keys after hash
exchange. `PASSTHROUGH` only restores local parallelism and does not
preserve key distribution, so duplicate distinct keys can be processed
by different local tasks and later partial sums can produce incorrect
results.

This PR keeps the knob behavior for aggregation stages that can safely
skip the local hash exchange, but excludes merge aggregation with a
serial child from the skip path. That case falls through to the existing
`HASH_SHUFFLE` / `BUCKET_HASH_SHUFFLE` distribution requirement. The PR
also computes the merge flag during `AggSinkOperatorX::init()` because
local exchange planning runs before `prepare()`.

### Release note

Fix occasional incorrect DISTINCT aggregate results when serial exchange
is enabled.

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

Previously, local exchange (LE) nodes were inserted exclusively by the
BE's `_plan_local_exchange` at pipeline build time.  The FE had no
visibility into which operators needed a fan-out or shuffle before
execution, making it impossible to validate, optimize, or override LE
decisions at planning time.

This PR introduces a full FE-side local exchange planner that mirrors
BE semantics, brings several correctness fixes, and leaves the legacy
BE path fully intact behind a feature flag.  See "Current architecture
notes" at the bottom for what the FE planner does and does not own.

A new `AddLocalExchange` pass runs after normal fragment assignment.
It walks each fragment's plan tree bottom-up, calling the polymorphic
`PlanNode.enforceAndDeriveLocalExchange()` on every node.  Nodes
declare what distribution they require of their children; the framework
inserts `LocalExchangeNode` where needed.

`LocalExchangeNode` represents intra-fragment data redistribution and
supports the full set of exchange types: PASSTHROUGH, HASH_SHUFFLE,
BUCKET_HASH_SHUFFLE, GLOBAL_EXECUTION_HASH_SHUFFLE, BROADCAST,
PASS_TO_ONE, ADAPTIVE_PASSTHROUGH, LOCAL_MERGE_SORT, and NOOP.

The pass is guarded by `enable_local_shuffle_planner` (default true).
When disabled, BE continues to run its own `_plan_local_exchange` as
before, keeping the old path fully intact.

`maxPerBeInstances` (max pipeline instances assigned to any single BE)
is used instead of a global `instanceCount`.  Planning is a no-op when
`maxPerBeInstances == 1` — inserting LE on a single-threaded pipeline
would cause task-count mismatches and pipeline starvation.

When a serial operator (e.g. OlapScanNode with a single tablet bucket)
feeds a non-serial parent without an intermediate LE, downstream tasks
starve waiting for data that never arrives.  The framework detects this
case and inserts a PASSTHROUGH LE to restore N-task parallelism, exactly
matching BE's `required_data_distribution()` serial → PASSTHROUGH rule.

`LocalExchangeTypeRequire` abstracts two strategies:
- `RequireHash` — always resolves to `LOCAL_EXECUTION_HASH_SHUFFLE`
  (safe for intra-fragment hash partitioning).
- `RequireSpecific` — preserves BUCKET_HASH_SHUFFLE /
  GLOBAL_EXECUTION_HASH_SHUFFLE without degradation.

PR #62438 added `enable_local_exchange_before_agg`, but its BE guard
`!_needs_finalize && !enable_local_exchange_before_agg → base` conflated
two semantically different cases in AggSink and DistinctStreamingAgg:

- **AggSink**: `!finalize && hasKeys` covered both LOCAL preagg
  (performance-only) and FIRST_MERGE dedup (correctness-critical).
  The flag-gated early-return wrongly skipped HASH for FIRST_MERGE,
  producing PASSTHROUGH-over-serial-child → wrong aggregation results.

- **DistinctStreamingAgg**: `!finalize` covered both streaming preagg
  (`useStreamingPreagg=true`, performance) and non-streaming dedup
  (`useStreamingPreagg=false`, correctness).  Same class of bug.

FE fix:
- AggSink: restrict the flag-gated base path to `!isMerge()` LOCAL
  phases.  FIRST_MERGE always emits HASH regardless of the flag.
- DistinctStreamingAgg: restrict to `useStreamingPreagg=true`.
  Non-streaming dedup always emits HASH.

Also add `requiresShuffleForCorrectness()` to mirror BE's
`is_shuffled_operator()`, so SetOperationNode propagates the
"downstream depends on hash" flag correctly instead of using the
coarser `parentRequire.preferType().isHashShuffle()` check that
over-inserted HASH LE on every union branch under a streaming preagg.

These fixes reduce FE/BE consistency mismatches from 8 to 3
(only pre-existing NLJ optimization differences remain).

- `enable_local_shuffle_planner` — use FE planner (default true)
- `enable_local_shuffle` — master switch for local shuffle
- `enable_local_exchange_before_agg` — HASH LE before non-final agg
  (default true, mirrors #62438)

`validateNoSerialWithoutLocalExchange()` walks the final plan tree and
logs a warning whenever a serial operator feeds a non-serial parent
without an intermediate LocalExchangeNode, catching planning gaps
before execution.

- `test_enable_local_exchange_before_agg.groovy` — 10 agg patterns
  with the flag on and off; covers the FIRST_MERGE and
  DistinctStreamingAgg correctness fixes.
- `test_local_shuffle_fe_be_consistency.groovy` — runs the same SQL
  with `enable_local_shuffle_planner=true` and `=false` across the
  full operator matrix (Agg, Sort, Analytic, HashJoin, NLJ, Set, Union,
  TableFunction, AssertNumRows, RQG-derived corner cases) and asserts
  result rows are identical.  Only data correctness is asserted — the
  two planners legitimately differ on the exact exchange counts/types
  they emit, so plan-shape equality is intentionally not checked.
- `test_local_shuffle_rqg_bugs.groovy` — reproduces 20+ RQG-found
  crashes and wrong-result cases.
- `test_old_coordinator_local_shuffle.groovy` — verifies the old
  coordinator path is unaffected.
- `test_multilevel_join_agg_local_shuffle.groovy` — multi-level join
  and aggregation plan shapes.

- `multi_version.h`: replace `atomic_load/atomic_store` (deprecated in
  libstdc++ C++20 / LLVM 20) with `std::shared_mutex`-based RW locking.
- `memory.cpp`: fix `std::max` type mismatch (`long` vs `int64_t`)
  on macOS.
- `bucketed_aggregation_sink_operator.h`: fix `ExchangeType::NOOP` →
  `TLocalPartitionType::NOOP` after thrift enum rename.

This PR puts the FE planner in the driver's seat for LE insertion but
intentionally does NOT remove the BE-side machinery — readers should be
aware of three pieces the FE planner shares with or defers to BE:

1. **`is_serial_operator` is computed on both sides.**  FE computes the
   flag and writes it into Thrift, but BE's
   `OperatorBase::is_serial_operator()` is still overridden per operator
   in C++ and used for BE-side runtime decisions.  Any future change to
   the BE override needs to be mirrored on the FE side (and vice versa)
   to keep the planner's view consistent with execution.

2. **The legacy BE planner stays as a fallback.**
   `pipeline_fragment_context.cpp::_plan_local_exchange` is preserved
   and gated by `runtime_state.h::plan_local_shuffle()`: when
   `enable_local_shuffle_planner=false`, BE plans LE itself, exactly as
   before.  The two paths are mutually exclusive, never both running on
   the same query.

3. **`_propagate_local_exchange_num_tasks` is kept as a runtime safety
   net.**  The two propagation passes in
   `pipeline_fragment_context.cpp` fix up paired pipelines whose
   `num_tasks` end up mismatched (e.g. when AGG/SORT/JOIN pipeline
   splits leave a serial Exchange feeding an N-task sink).  FE's
   framework-level serial→non-serial fan-out (`enforceRequire` step 3)
   and the `validateNoSerialWithoutLocalExchange` check aim to make
   these mismatches impossible by construction, but the BE-side fixup
   remains as a defensive guard.

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

Previously, local exchange (LE) nodes were inserted exclusively by the
BE's `_plan_local_exchange` at pipeline build time.  The FE had no
visibility into which operators needed a fan-out or shuffle before
execution, making it impossible to validate, optimize, or override LE
decisions at planning time.

This PR introduces a full FE-side local exchange planner that mirrors
BE semantics, brings several correctness fixes, and leaves the legacy
BE path fully intact behind a feature flag.  See "Current architecture
notes" at the bottom for what the FE planner does and does not own.

A new `AddLocalExchange` pass runs after normal fragment assignment.
It walks each fragment's plan tree bottom-up, calling the polymorphic
`PlanNode.enforceAndDeriveLocalExchange()` on every node.  Nodes
declare what distribution they require of their children; the framework
inserts `LocalExchangeNode` where needed.

`LocalExchangeNode` represents intra-fragment data redistribution and
supports the full set of exchange types: PASSTHROUGH, HASH_SHUFFLE,
BUCKET_HASH_SHUFFLE, GLOBAL_EXECUTION_HASH_SHUFFLE, BROADCAST,
PASS_TO_ONE, ADAPTIVE_PASSTHROUGH, LOCAL_MERGE_SORT, and NOOP.

The pass is guarded by `enable_local_shuffle_planner` (default true).
When disabled, BE continues to run its own `_plan_local_exchange` as
before, keeping the old path fully intact.

`maxPerBeInstances` (max pipeline instances assigned to any single BE)
is used instead of a global `instanceCount`.  Planning is a no-op when
`maxPerBeInstances == 1` — inserting LE on a single-threaded pipeline
would cause task-count mismatches and pipeline starvation.

When a serial operator (e.g. OlapScanNode with a single tablet bucket)
feeds a non-serial parent without an intermediate LE, downstream tasks
starve waiting for data that never arrives.  The framework detects this
case and inserts a PASSTHROUGH LE to restore N-task parallelism, exactly
matching BE's `required_data_distribution()` serial → PASSTHROUGH rule.

`LocalExchangeTypeRequire` abstracts two strategies:
- `RequireHash` — always resolves to `LOCAL_EXECUTION_HASH_SHUFFLE`
  (safe for intra-fragment hash partitioning).
- `RequireSpecific` — preserves BUCKET_HASH_SHUFFLE /
  GLOBAL_EXECUTION_HASH_SHUFFLE without degradation.

PR #62438 added `enable_local_exchange_before_agg`, but its BE guard
`!_needs_finalize && !enable_local_exchange_before_agg → base` conflated
two semantically different cases in AggSink and DistinctStreamingAgg:

- **AggSink**: `!finalize && hasKeys` covered both LOCAL preagg
  (performance-only) and FIRST_MERGE dedup (correctness-critical).
  The flag-gated early-return wrongly skipped HASH for FIRST_MERGE,
  producing PASSTHROUGH-over-serial-child → wrong aggregation results.

- **DistinctStreamingAgg**: `!finalize` covered both streaming preagg
  (`useStreamingPreagg=true`, performance) and non-streaming dedup
  (`useStreamingPreagg=false`, correctness).  Same class of bug.

FE fix:
- AggSink: restrict the flag-gated base path to `!isMerge()` LOCAL
  phases.  FIRST_MERGE always emits HASH regardless of the flag.
- DistinctStreamingAgg: restrict to `useStreamingPreagg=true`.
  Non-streaming dedup always emits HASH.

Also add `requiresShuffleForCorrectness()` to mirror BE's
`is_shuffled_operator()`, so SetOperationNode propagates the
"downstream depends on hash" flag correctly instead of using the
coarser `parentRequire.preferType().isHashShuffle()` check that
over-inserted HASH LE on every union branch under a streaming preagg.

These fixes reduce FE/BE consistency mismatches from 8 to 3
(only pre-existing NLJ optimization differences remain).

- `enable_local_shuffle_planner` — use FE planner (default true)
- `enable_local_shuffle` — master switch for local shuffle
- `enable_local_exchange_before_agg` — HASH LE before non-final agg
  (default true, mirrors #62438)

`validateNoSerialWithoutLocalExchange()` walks the final plan tree and
logs a warning whenever a serial operator feeds a non-serial parent
without an intermediate LocalExchangeNode, catching planning gaps
before execution.

- `test_enable_local_exchange_before_agg.groovy` — 10 agg patterns
  with the flag on and off; covers the FIRST_MERGE and
  DistinctStreamingAgg correctness fixes.
- `test_local_shuffle_fe_be_consistency.groovy` — runs the same SQL
  with `enable_local_shuffle_planner=true` and `=false` across the
  full operator matrix (Agg, Sort, Analytic, HashJoin, NLJ, Set, Union,
  TableFunction, AssertNumRows, RQG-derived corner cases) and asserts
  result rows are identical.  Only data correctness is asserted — the
  two planners legitimately differ on the exact exchange counts/types
  they emit, so plan-shape equality is intentionally not checked.
- `test_local_shuffle_rqg_bugs.groovy` — reproduces 20+ RQG-found
  crashes and wrong-result cases.
- `test_old_coordinator_local_shuffle.groovy` — verifies the old
  coordinator path is unaffected.
- `test_multilevel_join_agg_local_shuffle.groovy` — multi-level join
  and aggregation plan shapes.

- `multi_version.h`: replace `atomic_load/atomic_store` (deprecated in
  libstdc++ C++20 / LLVM 20) with `std::shared_mutex`-based RW locking.
- `memory.cpp`: fix `std::max` type mismatch (`long` vs `int64_t`)
  on macOS.
- `bucketed_aggregation_sink_operator.h`: fix `ExchangeType::NOOP` →
  `TLocalPartitionType::NOOP` after thrift enum rename.

This PR puts the FE planner in the driver's seat for LE insertion but
intentionally does NOT remove the BE-side machinery — readers should be
aware of three pieces the FE planner shares with or defers to BE:

1. **`is_serial_operator` is computed on both sides.**  FE computes the
   flag and writes it into Thrift, but BE's
   `OperatorBase::is_serial_operator()` is still overridden per operator
   in C++ and used for BE-side runtime decisions.  Any future change to
   the BE override needs to be mirrored on the FE side (and vice versa)
   to keep the planner's view consistent with execution.

2. **The legacy BE planner stays as a fallback.**
   `pipeline_fragment_context.cpp::_plan_local_exchange` is preserved
   and gated by `runtime_state.h::plan_local_shuffle()`: when
   `enable_local_shuffle_planner=false`, BE plans LE itself, exactly as
   before.  The two paths are mutually exclusive, never both running on
   the same query.

3. **`_propagate_local_exchange_num_tasks` is kept as a runtime safety
   net.**  The two propagation passes in
   `pipeline_fragment_context.cpp` fix up paired pipelines whose
   `num_tasks` end up mismatched (e.g. when AGG/SORT/JOIN pipeline
   splits leave a serial Exchange feeding an N-task sink).  FE's
   framework-level serial→non-serial fan-out (`enforceRequire` step 3)
   and the `validateNoSerialWithoutLocalExchange` check aim to make
   these mismatches impossible by construction, but the BE-side fixup
   remains as a defensive guard.

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

Previously, local exchange (LE) nodes were inserted exclusively by the
BE's `_plan_local_exchange` at pipeline build time.  The FE had no
visibility into which operators needed a fan-out or shuffle before
execution, making it impossible to validate, optimize, or override LE
decisions at planning time.

This PR introduces a full FE-side local exchange planner that mirrors
BE semantics, brings several correctness fixes, and leaves the legacy
BE path fully intact behind a feature flag.  See "Current architecture
notes" at the bottom for what the FE planner does and does not own.

A new `AddLocalExchange` pass runs after normal fragment assignment.
It walks each fragment's plan tree bottom-up, calling the polymorphic
`PlanNode.enforceAndDeriveLocalExchange()` on every node.  Nodes
declare what distribution they require of their children; the framework
inserts `LocalExchangeNode` where needed.

`LocalExchangeNode` represents intra-fragment data redistribution and
supports the full set of exchange types: PASSTHROUGH, HASH_SHUFFLE,
BUCKET_HASH_SHUFFLE, GLOBAL_EXECUTION_HASH_SHUFFLE, BROADCAST,
PASS_TO_ONE, ADAPTIVE_PASSTHROUGH, LOCAL_MERGE_SORT, and NOOP.

The pass is guarded by `enable_local_shuffle_planner` (default true).
When disabled, BE continues to run its own `_plan_local_exchange` as
before, keeping the old path fully intact.

`maxPerBeInstances` (max pipeline instances assigned to any single BE)
is used instead of a global `instanceCount`.  Planning is a no-op when
`maxPerBeInstances == 1` — inserting LE on a single-threaded pipeline
would cause task-count mismatches and pipeline starvation.

When a serial operator (e.g. OlapScanNode with a single tablet bucket)
feeds a non-serial parent without an intermediate LE, downstream tasks
starve waiting for data that never arrives.  The framework detects this
case and inserts a PASSTHROUGH LE to restore N-task parallelism, exactly
matching BE's `required_data_distribution()` serial → PASSTHROUGH rule.

`LocalExchangeTypeRequire` abstracts two strategies:
- `RequireHash` — always resolves to `LOCAL_EXECUTION_HASH_SHUFFLE`
  (safe for intra-fragment hash partitioning).
- `RequireSpecific` — preserves BUCKET_HASH_SHUFFLE /
  GLOBAL_EXECUTION_HASH_SHUFFLE without degradation.

PR #62438 added `enable_local_exchange_before_agg`, but its BE guard
`!_needs_finalize && !enable_local_exchange_before_agg → base` conflated
two semantically different cases in AggSink and DistinctStreamingAgg:

- **AggSink**: `!finalize && hasKeys` covered both LOCAL preagg
  (performance-only) and FIRST_MERGE dedup (correctness-critical).
  The flag-gated early-return wrongly skipped HASH for FIRST_MERGE,
  producing PASSTHROUGH-over-serial-child → wrong aggregation results.

- **DistinctStreamingAgg**: `!finalize` covered both streaming preagg
  (`useStreamingPreagg=true`, performance) and non-streaming dedup
  (`useStreamingPreagg=false`, correctness).  Same class of bug.

FE fix:
- AggSink: restrict the flag-gated base path to `!isMerge()` LOCAL
  phases.  FIRST_MERGE always emits HASH regardless of the flag.
- DistinctStreamingAgg: restrict to `useStreamingPreagg=true`.
  Non-streaming dedup always emits HASH.

Also add `requiresShuffleForCorrectness()` to mirror BE's
`is_shuffled_operator()`, so SetOperationNode propagates the
"downstream depends on hash" flag correctly instead of using the
coarser `parentRequire.preferType().isHashShuffle()` check that
over-inserted HASH LE on every union branch under a streaming preagg.

These fixes reduce FE/BE consistency mismatches from 8 to 3
(only pre-existing NLJ optimization differences remain).

- `enable_local_shuffle_planner` — use FE planner (default true)
- `enable_local_shuffle` — master switch for local shuffle
- `enable_local_exchange_before_agg` — HASH LE before non-final agg
  (default true, mirrors #62438)

`validateNoSerialWithoutLocalExchange()` walks the final plan tree and
logs a warning whenever a serial operator feeds a non-serial parent
without an intermediate LocalExchangeNode, catching planning gaps
before execution.

- `test_enable_local_exchange_before_agg.groovy` — 10 agg patterns
  with the flag on and off; covers the FIRST_MERGE and
  DistinctStreamingAgg correctness fixes.
- `test_local_shuffle_fe_be_consistency.groovy` — runs the same SQL
  with `enable_local_shuffle_planner=true` and `=false` across the
  full operator matrix (Agg, Sort, Analytic, HashJoin, NLJ, Set, Union,
  TableFunction, AssertNumRows, RQG-derived corner cases) and asserts
  result rows are identical.  Only data correctness is asserted — the
  two planners legitimately differ on the exact exchange counts/types
  they emit, so plan-shape equality is intentionally not checked.
- `test_local_shuffle_rqg_bugs.groovy` — reproduces 20+ RQG-found
  crashes and wrong-result cases.
- `test_old_coordinator_local_shuffle.groovy` — verifies the old
  coordinator path is unaffected.
- `test_multilevel_join_agg_local_shuffle.groovy` — multi-level join
  and aggregation plan shapes.

- `multi_version.h`: replace `atomic_load/atomic_store` (deprecated in
  libstdc++ C++20 / LLVM 20) with `std::shared_mutex`-based RW locking.
- `memory.cpp`: fix `std::max` type mismatch (`long` vs `int64_t`)
  on macOS.
- `bucketed_aggregation_sink_operator.h`: fix `ExchangeType::NOOP` →
  `TLocalPartitionType::NOOP` after thrift enum rename.

This PR puts the FE planner in the driver's seat for LE insertion but
intentionally does NOT remove the BE-side machinery — readers should be
aware of three pieces the FE planner shares with or defers to BE:

1. **`is_serial_operator` is computed on both sides.**  FE computes the
   flag and writes it into Thrift, but BE's
   `OperatorBase::is_serial_operator()` is still overridden per operator
   in C++ and used for BE-side runtime decisions.  Any future change to
   the BE override needs to be mirrored on the FE side (and vice versa)
   to keep the planner's view consistent with execution.

2. **The legacy BE planner stays as a fallback.**
   `pipeline_fragment_context.cpp::_plan_local_exchange` is preserved
   and gated by `runtime_state.h::plan_local_shuffle()`: when
   `enable_local_shuffle_planner=false`, BE plans LE itself, exactly as
   before.  The two paths are mutually exclusive, never both running on
   the same query.

3. **`_propagate_local_exchange_num_tasks` is kept as a runtime safety
   net.**  The two propagation passes in
   `pipeline_fragment_context.cpp` fix up paired pipelines whose
   `num_tasks` end up mismatched (e.g. when AGG/SORT/JOIN pipeline
   splits leave a serial Exchange feeding an N-task sink).  FE's
   framework-level serial→non-serial fan-out (`enforceRequire` step 3)
   and the `validateNoSerialWithoutLocalExchange` check aim to make
   these mismatches impossible by construction, but the BE-side fixup
   remains as a defensive guard.

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

Previously, local exchange (LE) nodes were inserted exclusively by the
BE's `_plan_local_exchange` at pipeline build time.  The FE had no
visibility into which operators needed a fan-out or shuffle before
execution, making it impossible to validate, optimize, or override LE
decisions at planning time.

This PR introduces a full FE-side local exchange planner that mirrors
BE semantics, brings several correctness fixes, and leaves the legacy
BE path fully intact behind a feature flag.  See "Current architecture
notes" at the bottom for what the FE planner does and does not own.

A new `AddLocalExchange` pass runs after normal fragment assignment.
It walks each fragment's plan tree bottom-up, calling the polymorphic
`PlanNode.enforceAndDeriveLocalExchange()` on every node.  Nodes
declare what distribution they require of their children; the framework
inserts `LocalExchangeNode` where needed.

`LocalExchangeNode` represents intra-fragment data redistribution and
supports the full set of exchange types: PASSTHROUGH, HASH_SHUFFLE,
BUCKET_HASH_SHUFFLE, GLOBAL_EXECUTION_HASH_SHUFFLE, BROADCAST,
PASS_TO_ONE, ADAPTIVE_PASSTHROUGH, LOCAL_MERGE_SORT, and NOOP.

The pass is guarded by `enable_local_shuffle_planner` (default true).
When disabled, BE continues to run its own `_plan_local_exchange` as
before, keeping the old path fully intact.

`maxPerBeInstances` (max pipeline instances assigned to any single BE)
is used instead of a global `instanceCount`.  Planning is a no-op when
`maxPerBeInstances == 1` — inserting LE on a single-threaded pipeline
would cause task-count mismatches and pipeline starvation.

When a serial operator (e.g. OlapScanNode with a single tablet bucket)
feeds a non-serial parent without an intermediate LE, downstream tasks
starve waiting for data that never arrives.  The framework detects this
case and inserts a PASSTHROUGH LE to restore N-task parallelism, exactly
matching BE's `required_data_distribution()` serial → PASSTHROUGH rule.

`LocalExchangeTypeRequire` abstracts two strategies:
- `RequireHash` — always resolves to `LOCAL_EXECUTION_HASH_SHUFFLE`
  (safe for intra-fragment hash partitioning).
- `RequireSpecific` — preserves BUCKET_HASH_SHUFFLE /
  GLOBAL_EXECUTION_HASH_SHUFFLE without degradation.

PR #62438 added `enable_local_exchange_before_agg`, but its BE guard
`!_needs_finalize && !enable_local_exchange_before_agg → base` conflated
two semantically different cases in AggSink and DistinctStreamingAgg:

- **AggSink**: `!finalize && hasKeys` covered both LOCAL preagg
  (performance-only) and FIRST_MERGE dedup (correctness-critical).
  The flag-gated early-return wrongly skipped HASH for FIRST_MERGE,
  producing PASSTHROUGH-over-serial-child → wrong aggregation results.

- **DistinctStreamingAgg**: `!finalize` covered both streaming preagg
  (`useStreamingPreagg=true`, performance) and non-streaming dedup
  (`useStreamingPreagg=false`, correctness).  Same class of bug.

FE fix:
- AggSink: restrict the flag-gated base path to `!isMerge()` LOCAL
  phases.  FIRST_MERGE always emits HASH regardless of the flag.
- DistinctStreamingAgg: restrict to `useStreamingPreagg=true`.
  Non-streaming dedup always emits HASH.

Also add `requiresShuffleForCorrectness()` to mirror BE's
`is_shuffled_operator()`, so SetOperationNode propagates the
"downstream depends on hash" flag correctly instead of using the
coarser `parentRequire.preferType().isHashShuffle()` check that
over-inserted HASH LE on every union branch under a streaming preagg.

These fixes reduce FE/BE consistency mismatches from 8 to 3
(only pre-existing NLJ optimization differences remain).

- `enable_local_shuffle_planner` — use FE planner (default true)
- `enable_local_shuffle` — master switch for local shuffle
- `enable_local_exchange_before_agg` — HASH LE before non-final agg
  (default true, mirrors #62438)

`validateNoSerialWithoutLocalExchange()` walks the final plan tree and
logs a warning whenever a serial operator feeds a non-serial parent
without an intermediate LocalExchangeNode, catching planning gaps
before execution.

- `test_enable_local_exchange_before_agg.groovy` — 10 agg patterns
  with the flag on and off; covers the FIRST_MERGE and
  DistinctStreamingAgg correctness fixes.
- `test_local_shuffle_fe_be_consistency.groovy` — runs the same SQL
  with `enable_local_shuffle_planner=true` and `=false` across the
  full operator matrix (Agg, Sort, Analytic, HashJoin, NLJ, Set, Union,
  TableFunction, AssertNumRows, RQG-derived corner cases) and asserts
  result rows are identical.  Only data correctness is asserted — the
  two planners legitimately differ on the exact exchange counts/types
  they emit, so plan-shape equality is intentionally not checked.
- `test_local_shuffle_rqg_bugs.groovy` — reproduces 20+ RQG-found
  crashes and wrong-result cases.
- `test_old_coordinator_local_shuffle.groovy` — verifies the old
  coordinator path is unaffected.
- `test_multilevel_join_agg_local_shuffle.groovy` — multi-level join
  and aggregation plan shapes.

- `multi_version.h`: replace `atomic_load/atomic_store` (deprecated in
  libstdc++ C++20 / LLVM 20) with `std::shared_mutex`-based RW locking.
- `memory.cpp`: fix `std::max` type mismatch (`long` vs `int64_t`)
  on macOS.
- `bucketed_aggregation_sink_operator.h`: fix `ExchangeType::NOOP` →
  `TLocalPartitionType::NOOP` after thrift enum rename.

This PR puts the FE planner in the driver's seat for LE insertion but
intentionally does NOT remove the BE-side machinery — readers should be
aware of three pieces the FE planner shares with or defers to BE:

1. **`is_serial_operator` is computed on both sides.**  FE computes the
   flag and writes it into Thrift, but BE's
   `OperatorBase::is_serial_operator()` is still overridden per operator
   in C++ and used for BE-side runtime decisions.  Any future change to
   the BE override needs to be mirrored on the FE side (and vice versa)
   to keep the planner's view consistent with execution.

2. **The legacy BE planner stays as a fallback.**
   `pipeline_fragment_context.cpp::_plan_local_exchange` is preserved
   and gated by `runtime_state.h::plan_local_shuffle()`: when
   `enable_local_shuffle_planner=false`, BE plans LE itself, exactly as
   before.  The two paths are mutually exclusive, never both running on
   the same query.

3. **`_propagate_local_exchange_num_tasks` is kept as a runtime safety
   net.**  The two propagation passes in
   `pipeline_fragment_context.cpp` fix up paired pipelines whose
   `num_tasks` end up mismatched (e.g. when AGG/SORT/JOIN pipeline
   splits leave a serial Exchange feeding an N-task sink).  FE's
   framework-level serial→non-serial fan-out (`enforceRequire` step 3)
   and the `validateNoSerialWithoutLocalExchange` check aim to make
   these mismatches impossible by construction, but the BE-side fixup
   remains as a defensive guard.

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

…63366)

Problem Summary:

Move local exchange (LE) planning from BE's `_plan_local_exchange`
(pipeline build time) to a new FE-side planner. The FE planner mirrors
BE semantics, brings several correctness fixes, and is gated by a
session variable so the legacy BE path stays available as a fallback.

**Core design**

- New `AddLocalExchange` pass runs after `DistributePlanner`, walking
each fragment's plan tree bottom-up via the polymorphic
`PlanNode.enforceAndDeriveLocalExchange()`. Each node declares what
distribution it requires of its children; the framework inserts
`LocalExchangeNode` where needed.
- `LocalExchangeNode` represents intra-fragment data redistribution and
supports PASSTHROUGH, GLOBAL/LOCAL/BUCKET HASH_SHUFFLE, BROADCAST,
PASS_TO_ONE, ADAPTIVE_PASSTHROUGH, LOCAL_MERGE_SORT, NOOP.
- Per-BE instance semantics: `maxPerBeInstances` (max pipeline instances
assigned to any single BE) is used instead of global instance count to
match BE's `_num_instances` check. Planning is a no-op when
`maxPerBeInstances == 1`.
- Serial → non-serial fan-out: when a serial operator feeds a non-serial
parent without an intermediate LE, the framework inserts a PASSTHROUGH
LE to restore N-task parallelism, matching BE's
`required_data_distribution()` rule.
- Requirement-based exchange type resolution via
`LocalExchangeTypeRequire`: `RequireHash` adapts to any hash flavour,
`RequireSpecific` preserves the exact requested type.

**AggregationNode correctness fixes**

PR #62438 introduced a semantic split for
`required_data_distribution=HASH` (correctness-required vs
performance-only). BE's `!_needs_finalize &&
!enable_local_exchange_before_agg → base` early-return conflates both
intents in `AggSinkOperatorX` and `DistinctStreamingAggOperatorX`,
wrongly catching FIRST_MERGE (correctness) / non-streaming dedup
(correctness) and producing PASSTHROUGH-over-serial-child → wrong
aggregation results. The FE planner adds the missing `!isMerge()` /
`useStreamingPreagg=true` guards so FIRST_MERGE and non-streaming dedup
always emit HASH, regardless of the flag. Also adds
`requiresShuffleForCorrectness()` (mirrors BE's
`is_shuffled_operator()`) so SetOperationNode propagates the "downstream
depends on hash" flag correctly through chains.

**Session variables**

- `enable_local_shuffle_planner` (default true) — use FE planner; when
false, BE plans LE itself via the legacy path.
- `enable_local_shuffle` — master switch.
- `enable_local_exchange_before_agg` — mirrors #62438.

**Architectural notes**

This PR puts the FE planner in the driver's seat for LE insertion but
intentionally keeps BE-side machinery as a fallback:

1. `is_serial_operator` is still computed on both sides — any future
change to BE's per-operator C++ override must be mirrored in FE.
2. Legacy BE planner
(`pipeline_fragment_context.cpp::_plan_local_exchange`) is preserved and
gated by `runtime_state.h::plan_local_shuffle()`; the two paths are
mutually exclusive.
3. `_propagate_local_exchange_num_tasks` is kept as a runtime safety net
for paired-pipeline num_tasks mismatches.

**Thrift enum rename**

The intra-fragment exchange enum is renamed `ExchangeType` →
`TLocalPartitionType` (and `HASH_SHUFFLE` →
`GLOBAL_EXECUTION_HASH_SHUFFLE`) for clarity; the BE operator headers
are updated mechanically. This accounts for the otherwise-mechanical BE
`.h` churn.

### Release note

Add session variable `enable_local_shuffle_planner` (default true) to
control whether local exchange nodes are planned in FE (new path) or in
BE (legacy `_plan_local_exchange`). The two paths are mutually
exclusive; the legacy path remains intact behind this flag.

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

Add a knob to control local exchange before agg

Add a knob to control local exchange before agg

## Proposed changes

Backport the requested changes to `branch-4.1` in master merge order,
skipping changes already present in this PR:

1. #62438
2. #65329 (merged prerequisite for the nested-schema cases)
3. #65960
4. #65965
5. #65972
6. #65998
7. #66002
8. #65992
9. #66021
10. #66036
11. #66008
12. #66073
13. #66056 (explicitly requested; current open-PR head, appended after
the merged sequence)

The branch-specific compatibility commits preserve the selected master
behavior on `branch-4.1`, including master wire IDs for file formats and
the merged Paimon test helper prerequisite. No regression expected
output, test assertion, or test input was changed to make validation
pass.

## Validation

- Full BE ASAN build passed.
- Full FE build passed.
- Targeted BE ASAN unit tests: 332 tests from 13 suites passed.
- Targeted FE Iceberg unit tests: 41 passed, 0 failed.
- Iceberg write regression: 20 suites, 0 failed, 0 fatal.
- `PaimonScanNodeTest`: all 16 test bodies completed with 0 assertion
failures; the class reports one Mockito teardown error because #66008
left four now-unused stubs on master. The still-open #65867 contains the
upstream test-only cleanup commit. This PR intentionally does not alter
those test cases.
- Final rebase against the latest `branch-4.1` completed; the branch was
already up to date.
- Working-tree, formatting, and sensitive-information audits completed.
Existing EOF blank lines in picked regression output files are
preserved.

---------

Co-authored-by: daidai <[email protected]>
Co-authored-by: Mingyu Chen (Rayner) <[email protected]>

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

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