[refactor](local shuffle) Refactor FE local exchange planning: extract helper, fix root PASSTHROUGH, fix ScanNode output

1. Extract duplicated heavy-ops bottleneck avoidance code from enforceChild/
   enforceChildExchange/forceEnforceChildExchange into insertLocalExchange().
2. Fix root-level PASSTHROUGH insertion: only check newRoot.isSerialOperator()
   instead of recursive hasSerialChildren() — after enforceAndDeriveLocalExchange(),
   serial children are already handled by LocalExchangeNodes inserted during tree walk.
   This fixes 6 FE/BE consistency mismatches where FE inserted ~2x PASSTHROUGH vs BE.
3. Fix ScanNode base class: return NOOP instead of BUCKET_HASH_SHUFFLE (only
   OlapScanNode should report bucket distribution for non-pooling scans).
4. Remove ExchangeNode.toThrift() dependency on hasSerialScanNode() — with FE-planned
   local exchanges, serial scans are handled by PASSTHROUGH fan-out.
5. Fix DistributePlanner: BUCKET_SHUFFLE with local shuffle returns all instances
   (not just first-per-worker); use assignedJoinBucketIndexes for bucket mapping.
6. Cap parallelism in UnassignedScanBucketOlapTableJob when fragment has non-serial
   exchange to avoid padding instances that hang.
7. Remove dead DataStreamSink.getLocalExchangeTypeRequire() override.
8. Inline shouldUseLocalExecutionHash() constant.
9. Fix test Bug 18/19: use col_int_undef_signed (existing column) instead of
   col_varchar_10__undef_signed (non-existent).

Author: 924060929

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/DistributePlanner.java

@@ -31,6 +31,7 @@
 import org.apache.doris.nereids.trees.plans.distribute.worker.job.BucketScanSource;
 import org.apache.doris.nereids.trees.plans.distribute.worker.job.DefaultScanSource;
 import org.apache.doris.nereids.trees.plans.distribute.worker.job.LocalShuffleAssignedJob;
+import org.apache.doris.nereids.trees.plans.distribute.worker.job.LocalShuffleBucketJoinAssignedJob;
 import org.apache.doris.nereids.trees.plans.distribute.worker.job.StaticAssignedJob;
 import org.apache.doris.nereids.trees.plans.distribute.worker.job.UnassignedJob;
 import org.apache.doris.nereids.trees.plans.distribute.worker.job.UnassignedJobBuilder;
@@ -244,6 +245,13 @@ private List<AssignedJob> filterInstancesWhichCanReceiveDataFromRemote(
         boolean useLocalShuffle = receiverPlan.getInstanceJobs().stream()
                 .anyMatch(LocalShuffleAssignedJob.class::isInstance);
         if (useLocalShuffle) {
+            // For BUCKET_SHUFFLE with local shuffle, each instance handles specific buckets.
+            // The exchange must route data to each instance individually, not just the first
+            // per worker. Otherwise instances 1..N-1 create ExchangeSource tasks that never
+            // receive data or EOS, causing a hang.
+            if (linkNode.getPartitionType() == TPartitionType.BUCKET_SHFFULE_HASH_PARTITIONED) {
+                return receiverPlan.getInstanceJobs();
+            }
             return getFirstInstancePerWorker(receiverPlan.getInstanceJobs());
         } else if (enableShareHashTableForBroadcastJoin && linkNode.isRightChildOfBroadcastHashJoin()) {
             return getFirstInstancePerWorker(receiverPlan.getInstanceJobs());
@@ -256,8 +264,20 @@ private List<AssignedJob> sortDestinationInstancesByBuckets(
             PipelineDistributedPlan plan, List<AssignedJob> unsorted, int bucketNum) {
         AssignedJob[] instances = new AssignedJob[bucketNum];
         for (AssignedJob instanceJob : unsorted) {
-            BucketScanSource bucketScanSource = (BucketScanSource) instanceJob.getScanSource();
-            for (Integer bucketIndex : bucketScanSource.bucketIndexToScanNodeToTablets.keySet()) {
+            // For LocalShuffleBucketJoinAssignedJob, use assignedJoinBucketIndexes which
+            // gives the specific join bucket(s) this instance handles. The scanSource
+            // buckets include ALL buckets on this worker (shared via pooling scan),
+            // which would cause collisions when multiple instances on the same worker
+            // each handle different join buckets.
+            Iterable<Integer> bucketIndices;
+            if (instanceJob instanceof LocalShuffleBucketJoinAssignedJob) {
+                bucketIndices = ((LocalShuffleBucketJoinAssignedJob) instanceJob)
+                        .getAssignedJoinBucketIndexes();
+            } else {
+                BucketScanSource bucketScanSource = (BucketScanSource) instanceJob.getScanSource();
+                bucketIndices = bucketScanSource.bucketIndexToScanNodeToTablets.keySet();
+            }
+            for (Integer bucketIndex : bucketIndices) {
                 if (instances[bucketIndex] != null) {
                     throw new IllegalStateException(
                             "Multi instances scan same buckets: " + instances[bucketIndex] + " and " + instanceJob

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/UnassignedScanBucketOlapTableJob.java

@@ -539,6 +539,27 @@ protected int degreeOfParallelism(int maxParallel, boolean useLocalShuffleToAddP
         }
 
         int maxParallelism = (int) Math.max(tabletNum, fragment.getParallelExecNum());
-        return Math.min(maxParallelism, colocateMaxParallelNum);
+        int result = Math.min(maxParallelism, colocateMaxParallelNum);
+
+        // When the fragment has a non-serial exchange, the BE won't insert local exchange
+        // for the exchange pipeline (distribution matches BUCKET_HASH_SHUFFLE). Instances
+        // beyond per-BE bucket count are "padding" with no bucket assignment — they create
+        // VDataStreamRecvrs that never receive data and hang. Cap at per-BE bucket count.
+        if (useLocalShuffleToAddParallel && hasNonSerialExchangeInFragment()) {
+            result = Math.min(result, maxParallel);
+        }
+
+        return result;
+    }
+
+    private boolean hasNonSerialExchangeInFragment() {
+        List<ExchangeNode> exchanges = fragment.getPlanRoot()
+                .collectInCurrentFragment(ExchangeNode.class::isInstance);
+        for (ExchangeNode exchange : exchanges) {
+            if (!exchange.isSerialOperator()) {
+                return true;
+            }
+        }
+        return false;
     }
 }

fe/fe-core/src/main/java/org/apache/doris/planner/AddLocalExchange.java

@@ -70,17 +70,16 @@ private void addLocalExchangeForFragment(PlanFragment fragment, PlanTranslatorCo
         Pair<PlanNode, LocalExchangeType> output = root
                 .enforceAndDeriveLocalExchange(context, null, require);
         PlanNode newRoot = output.first;
-        // Mirror BE OperatorBase base class required_data_distribution():
-        // when any operator in the fragment plan is serial AND the fragment uses pooling scan
-        // (LocalShuffleAssignedJob → ignoreDataDistribution → _parallel_instances=1),
-        // serial operators reduce pipeline num_tasks to 1, and this propagates via
-        // add_pipeline() to all downstream pipelines (including the DataStreamSink pipeline).
-        // Only instance 0 creates tasks and sends EOS. Downstream receivers expect
-        // _num_instances EOSes → hang.
+        // The DataStreamSink runs in the same pipeline as the fragment root.
+        // If the root is serial in a pooling-scan fragment, the sink pipeline has 1 task
+        // and only instance 0 sends EOS → downstream receivers hang.
         // Insert PASSTHROUGH fan-out to create _num_instances sink tasks.
-        // Non-pooling fragments (regular bucket distribution) have _num_instances ==
-        // bucket_count and every instance gets a scan range, so all sink tasks run.
-        if (isLocalShuffle && (newRoot.isSerialOperator() || newRoot.hasSerialChildren())) {
+        //
+        // Only check the root node itself (not recursive hasSerialChildren), because
+        // after enforceAndDeriveLocalExchange(), any serial children deeper in the tree
+        // are already handled by LocalExchangeNodes inserted during the tree walk.
+        // Those LocalExchangeNodes create pipeline boundaries with _num_instances tasks.
+        if (isLocalShuffle && newRoot.isSerialOperator()) {
             newRoot = new LocalExchangeNode(context.nextPlanNodeId(), newRoot,
                     LocalExchangeType.PASSTHROUGH, null);
         }
@@ -116,22 +115,17 @@ public static boolean isColocated(PlanNode plan) {
 
     public static LocalExchangeType resolveExchangeType(LocalExchangeTypeRequire require,
             PlanTranslatorContext translatorContext, PlanNode parent, PlanNode child) {
-        // Only generic RequireHash adapts to LOCAL_EXECUTION_HASH_SHUFFLE based on context.
+        // Only generic RequireHash adapts to LOCAL_EXECUTION_HASH_SHUFFLE.
         // Explicit RequireSpecific (GLOBAL_EXECUTION_HASH_SHUFFLE, BUCKET_HASH_SHUFFLE, etc.)
         // must never be degraded — if they appear in an invalid context, the plan is wrong.
-        if (require instanceof RequireHash
-                && shouldUseLocalExecutionHash(translatorContext, parent, child)) {
-            return LocalExchangeType.LOCAL_EXECUTION_HASH_SHUFFLE;
-        }
-        return require.preferType();
-    }
-
-    private static boolean shouldUseLocalExecutionHash(
-            PlanTranslatorContext translatorContext, PlanNode parent, PlanNode child) {
+        //
         // Always prefer LOCAL_EXECUTION_HASH_SHUFFLE for FE-planned intra-fragment hash exchanges.
         // GLOBAL_EXECUTION_HASH_SHUFFLE requires shuffle_idx_to_instance_idx which may be empty
         // for fragments with non-hash sinks (UNPARTITIONED/MERGE). LOCAL_HASH is always safe
         // since it partitions by local instance count without needing external shuffle maps.
-        return true;
+        if (require instanceof RequireHash) {
+            return LocalExchangeType.LOCAL_EXECUTION_HASH_SHUFFLE;
+        }
+        return require.preferType();
     }
 }

fe/fe-core/src/main/java/org/apache/doris/planner/DataStreamSink.java

@@ -25,7 +25,6 @@
 import org.apache.doris.analysis.ExprToThriftVisitor;
 import org.apache.doris.analysis.ToSqlParams;
 import org.apache.doris.analysis.TupleDescriptor;
-import org.apache.doris.planner.LocalExchangeNode.LocalExchangeTypeRequire;
 import org.apache.doris.thrift.TDataSink;
 import org.apache.doris.thrift.TDataSinkType;
 import org.apache.doris.thrift.TDataStreamSink;
@@ -241,8 +240,4 @@ protected TDataSink toThrift() {
         return result;
     }
 
-    @Override
-    public LocalExchangeTypeRequire getLocalExchangeTypeRequire() {
-        return LocalExchangeTypeRequire.noRequire();
-    }
 }

fe/fe-core/src/main/java/org/apache/doris/planner/ExchangeNode.java

@@ -102,15 +102,12 @@ public void setMergeInfo(SortInfo info) {
 
     @Override
     protected void toThrift(TPlanNode msg) {
-        // If this fragment has another scan node, this exchange node is serial or not should be decided by the scan
-        // node.
-        // When FE-planned local shuffle is enabled, ExchangeNode's serial flag is preserved
-        // (same as BE-planned path). The ExchangeNode pipeline will have num_tasks=1 when serial,
-        // meaning only instance 0 creates a task. This is correct because FE's
-        // filterInstancesWhichCanReceiveDataFromRemote() ensures only the first instance per
-        // worker receives remote data. The deferred exchanger uses upstream_pipe->num_tasks()
-        // to get the correct sender_count.
-        msg.setIsSerialOperator((isSerialOperator() || fragment.hasSerialScanNode())
+        // Exchange serial status should depend only on isSerialOperator() (UNPARTITIONED or
+        // use_serial_exchange), NOT on fragment.hasSerialScanNode(). With FE-planned local
+        // exchanges, serial scans are already handled by PASSTHROUGH fan-out. Making
+        // non-UNPARTITIONED exchanges (e.g. BUCKET_SHUFFLE) serial would reduce the build-side
+        // pipeline to num_tasks=1, preventing shared state propagation to other instances.
+        msg.setIsSerialOperator(isSerialOperator()
                 && fragment.useSerialSource(ConnectContext.get()));
         msg.node_type = TPlanNodeType.EXCHANGE_NODE;
         msg.exchange_node = new TExchangeNode();

fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java

@@ -1419,6 +1419,12 @@ public long getCatalogId() {
     public Pair<PlanNode, LocalExchangeType> enforceAndDeriveLocalExchange(
             PlanTranslatorContext translatorContext, PlanNode parent,
             LocalExchangeTypeRequire parentRequire) {
-        return super.enforceAndDeriveLocalExchange(translatorContext, parent, parentRequire);
+        boolean useSerialSource = fragment != null
+                && fragment.useSerialSource(translatorContext.getConnectContext());
+        if (useSerialSource) {
+            return Pair.of(this, LocalExchangeType.NOOP);
+        }
+        // Non-pooling OlapScan has bucket distribution — each instance scans specific buckets
+        return Pair.of(this, LocalExchangeType.BUCKET_HASH_SHUFFLE);
     }
 }