[fix](local shuffle) Fix hang/crash with use_serial_exchange=true in non-pooling fragments

Root cause: When use_serial_exchange=true, HASH_PARTITIONED Exchange becomes serial.
ExchangeNode.enforceAndDeriveLocalExchange() returned NOOP for serial HASH Exchange,
causing parent AggNode to insert LOCAL_EXECUTION_HASH_SHUFFLE LE. On BE, the serial
Exchange pipeline has 1 task but LE downstream has _num_instances tasks. AggSink on
instances 1+ has empty source_deps -> crash (ASAN) or hang (Release).

Fix: Distinguish pooling vs non-pooling fragments in ExchangeNode:
- Pooling fragment: serial Exchange returns NOOP (parent inserts LE for fan-out)
- Non-pooling fragment: serial Exchange reports actual distribution type
  (GLOBAL_EXECUTION_HASH_SHUFFLE / BUCKET_HASH_SHUFFLE / NOOP), preventing
  parent from inserting unnecessary LE

Also restores ExchangeNode.toThrift() to original logic (include hasSerialScanNode
check) since non-pooling fragments need this for correct serial detection.

Adds Bug 20 regression test for this hang scenario.

Author: 924060929

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

@@ -58,7 +58,6 @@ public void addLocalExchange(FragmentIdMapping<DistributedPlan> distributedPlans
         }
     }
 
-    /** @return true if a LOCAL_EXCHANGE_NODE was inserted in this fragment */
     private void addLocalExchangeForFragment(PlanFragment fragment, PlanTranslatorContext context,
             boolean isLocalShuffle) {
         DataSink sink = fragment.getSink();

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

@@ -102,12 +102,11 @@ public void setMergeInfo(SortInfo info) {
 
     @Override
     protected void toThrift(TPlanNode msg) {
-        // 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()
+        // BE local shuffle only supports two modes: 1 instance or all instances receiving data.
+        // hasSerialScanNode() ensures Exchange is serial when sibling scan is pooling,
+        // avoiding a middle-ground where some but not all instances receive data
+        // (EOS count mismatch → hang).
+        msg.setIsSerialOperator((isSerialOperator() || fragment.hasSerialScanNode())
                 && fragment.useSerialSource(ConnectContext.get()));
         msg.node_type = TPlanNodeType.EXCHANGE_NODE;
         msg.exchange_node = new TExchangeNode();
@@ -182,7 +181,9 @@ public Pair<PlanNode, LocalExchangeType> enforceAndDeriveLocalExchange(PlanTrans
         // Without useSerialSource() check, we'd insert PASSTHROUGH in non-pooling fragments
         // where the exchange has N tasks, corrupting broadcast join data distribution
         // (PASSTHROUGH round-robin splits complete-dataset sinks into 1/N subsets per source).
-        boolean willBeSerialOnBe = isSerialOperator()
+        // Must match toThrift: include hasSerialScanNode() — when sibling scan is pooling,
+        // Exchange becomes serial on BE even if Exchange itself isn't inherently serial.
+        boolean willBeSerialOnBe = (isSerialOperator() || fragment.hasSerialScanNode())
                 && fragment != null
                 && fragment.useSerialSource(ConnectContext.get());
         if (willBeSerialOnBe) {
@@ -195,23 +196,35 @@ public Pair<PlanNode, LocalExchangeType> enforceAndDeriveLocalExchange(PlanTrans
             if (translatorContext.hasSerialAncestorInPipeline(this)) {
                 return Pair.of(this, LocalExchangeType.NOOP);
             }
-            // Serial exchange → 1 task. Must fan out to N tasks for downstream operators.
-            // For HASH/BUCKET exchanges: return NOOP and let parent insert the appropriate
-            // redistribution (HASH_SHUFFLE or BUCKET_HASH_SHUFFLE). PASSTHROUGH round-robin
-            // would corrupt the hash/bucket distribution, and PASS_TO_ONE doesn't work for
-            // BUCKET_SHUFFLE joins (no shared hash table mechanism unlike BROADCAST).
-            // The heavy-ops bottleneck avoidance in enforceChildExchange() will automatically
-            // insert a PASSTHROUGH fan-out before the hash/bucket shuffle if needed.
+            // Serial HASH/BUCKET exchange:
+            // In pooling fragments, return NOOP so parent inserts hash/bucket LE with
+            // PASSTHROUGH fan-out (heavy-ops avoidance). Serial exchange has 1 task,
+            // LE fans out to _num_instances tasks.
+            // In non-pooling fragments, report the actual distribution type so parent's
+            // require is satisfied without inserting LE. The serial exchange reduces
+            // pipeline num_tasks to 1, matching BE-native behavior. Inserting LE in
+            // non-pooling fragments creates a pipeline split where downstream has
+            // _num_instances tasks but only 1 sender, causing shared-state mismatch.
             if (partitionType == TPartitionType.HASH_PARTITIONED
                     || partitionType == TPartitionType.BUCKET_SHFFULE_HASH_PARTITIONED) {
-                return Pair.of(this, LocalExchangeType.NOOP);
+                if (translatorContext.isLocalShuffleFragment()) {
+                    return Pair.of(this, LocalExchangeType.NOOP);
+                }
+                LocalExchangeType outputType = partitionType == TPartitionType.HASH_PARTITIONED
+                        ? LocalExchangeType.GLOBAL_EXECUTION_HASH_SHUFFLE
+                        : LocalExchangeType.BUCKET_HASH_SHUFFLE;
+                return Pair.of(this, outputType);
+            }
+            // For UNPARTITIONED (broadcast): in pooling fragments, PASSTHROUGH fan-out is
+            // needed because the exchange has 1 task but downstream needs N tasks.
+            // In non-pooling fragments, don't insert PASSTHROUGH — BE-native handles
+            // this via _plan_local_exchange which checks serial operators in the pipeline.
+            if (translatorContext.isLocalShuffleFragment()) {
+                PlanNode pt = new LocalExchangeNode(translatorContext.nextPlanNodeId(),
+                        this, LocalExchangeType.PASSTHROUGH, null);
+                return Pair.of(pt, LocalExchangeType.PASSTHROUGH);
             }
-            // For UNPARTITIONED (broadcast): PASSTHROUGH fan-out is safe because the
-            // exchange has the complete dataset. Parent nodes (HashJoin) may add PASS_TO_ONE
-            // on top for single-builder semantics (broadcast join build side).
-            PlanNode pt = new LocalExchangeNode(translatorContext.nextPlanNodeId(),
-                    this, LocalExchangeType.PASSTHROUGH, null);
-            return Pair.of(pt, LocalExchangeType.PASSTHROUGH);
+            return Pair.of(this, LocalExchangeType.NOOP);
         } else if (partitionType == TPartitionType.HASH_PARTITIONED) {
             return Pair.of(this, LocalExchangeType.GLOBAL_EXECUTION_HASH_SHUFFLE);
         } else if (partitionType == TPartitionType.BUCKET_SHFFULE_HASH_PARTITIONED) {

regression-test/suites/nereids_p0/local_shuffle/test_local_shuffle_rqg_bugs.groovy

@@ -1091,5 +1091,63 @@ suite("test_local_shuffle_rqg_bugs") {
         assertTrue(false, "Bug 19: Serial NLJ build-side source_deps crash: ${t.message}")
     }
 
+    //  Bug 20: Hang (ASAN: COREDUMP source_deps.size()=0 in AggSinkOperatorX) when
+    //  use_serial_exchange=true + RIGHT JOIN + GROUP BY in non-pooling fragment.
+    //  Root cause: serial HASH Exchange in non-pooling fragment returned NOOP, causing FE
+    //  to insert LOCAL_EXECUTION_HASH_SHUFFLE LE. On BE, serial Exchange pipeline has 1 task
+    //  but LE downstream has _num_instances tasks. AggSink on instances 1+ has empty source_deps.
+    //  Fix: ExchangeNode.enforceAndDeriveLocalExchange() returns actual distribution type
+    //  (GLOBAL_EXECUTION_HASH_SHUFFLE/BUCKET_HASH_SHUFFLE) for serial Exchange in non-pooling
+    //  fragments, preventing LE insertion.
+    //  Requires 3+ BEs to reproduce (single BE has _num_instances=1, no hang).
+    try {
+        logger.info("Bug 20: Testing serial exchange + agg hang in non-pooling fragment")
+        // Baseline uses same fuzzy vars but with planner=false (BE-native).
+        // This way we compare FE-planned vs BE-native under identical conditions,
+        // not against "correct" results — use_serial_exchange=true itself may have
+        // pre-existing BE bugs with certain pptn values.
+        def bug20_baseline_sql = { int ppt -> """
+            SELECT /*+SET_VAR(use_serial_exchange=true,
+                              parallel_pipeline_task_num=${ppt},
+                              enable_local_shuffle_planner=false,
+                              ignore_storage_data_distribution=true,
+                              enable_sql_cache=false,
+                              enable_share_hash_table_for_broadcast_join=false,
+                              disable_streaming_preaggregations=true)*/
+                table1.pk AS field1
+            FROM rqg_t1 AS table1
+            RIGHT OUTER JOIN rqg_t1 AS table2 ON table1.pk = table2.pk
+            LEFT JOIN rqg_t1 AS table3 ON table3.pk = table1.pk
+            WHERE table1.col_int_undef_signed IS NOT NULL OR table1.pk <> 10
+            GROUP BY field1
+            ORDER BY 1
+        """ }
+        for (int ppt : [3, 4, 7]) {
+            def bug20_baseline = sql bug20_baseline_sql(ppt)
+            def bug20_result = sql """
+                SELECT /*+SET_VAR(use_serial_exchange=true,
+                                  parallel_pipeline_task_num=${ppt},
+                                  enable_local_shuffle_planner=true,
+                                  ignore_storage_data_distribution=true,
+                                  enable_sql_cache=false,
+                                  enable_share_hash_table_for_broadcast_join=false,
+                                  disable_streaming_preaggregations=true)*/
+                    table1.pk AS field1
+                FROM rqg_t1 AS table1
+                RIGHT OUTER JOIN rqg_t1 AS table2 ON table1.pk = table2.pk
+                LEFT JOIN rqg_t1 AS table3 ON table3.pk = table1.pk
+                WHERE table1.col_int_undef_signed IS NOT NULL OR table1.pk <> 10
+                GROUP BY field1
+                ORDER BY 1
+            """
+            assertEquals(bug20_baseline, bug20_result,
+                "Bug 20 pptn=${ppt}: result mismatch with serial exchange + agg hang")
+        }
+        logger.info("Bug 20: PASSED (no hang, correct results)")
+    } catch (Throwable t) {
+        logger.error("Bug 20 FAILED: ${t.message}")
+        assertTrue(false, "Bug 20: Serial exchange + agg hang: ${t.message}")
+    }
+
     logger.info("=== All RQG bug reproduction tests completed ===")
 }