(doris) branch fe_local_shuffle updated: [fix](fe) Guard PASSTHROUGH insertion with useSerialSource() check

[hu...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

huajianlan pushed a commit to branch fe_local_shuffle
in repository https://gitbox.apache.org/repos/asf/doris.git


The following commit(s) were added to refs/heads/fe_local_shuffle by this push:
     new d9524ee3965 [fix](fe) Guard PASSTHROUGH insertion with useSerialSource() check
d9524ee3965 is described below

commit d9524ee3965ab89a0baacd2972f345886747eb23
Author: 924060929 <la...@selectdb.com>
AuthorDate: Wed Apr 8 21:55:45 2026 +0800

    [fix](fe) Guard PASSTHROUGH insertion with useSerialSource() check
    
    ### What problem does this PR solve?
    
    Problem Summary: The previous PASSTHROUGH insertion in ExchangeNode.enforceAndDeriveLocalExchange()
    only checked isSerialOperator() to decide whether to insert a PASSTHROUGH local exchange.
    However, the BE only treats an exchange as serial when BOTH isSerialOperator() AND
    fragment.useSerialSource() are true. Without the useSerialSource() guard, PASSTHROUGH was
    incorrectly inserted 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).
    
    The fix adds fragment.useSerialSource() check to match the BE's serial condition in toThrift().
    
    ### Release note
    
    None
    
    ### Check List (For Author)
    
    - Test: Manual EXPLAIN verification for UNION, broadcast join, and BUCKET_SHUFFLE join
    - Behavior changed: No
    - Does this need documentation: No
    
    Co-authored-by: Copilot <22...@users.noreply.github.com>
---
 .../org/apache/doris/planner/ExchangeNode.java     | 23 ++++++++++++++--------
 1 file changed, 15 insertions(+), 8 deletions(-)

diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/ExchangeNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/ExchangeNode.java
index 508788717c4..59c89f701bb 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/ExchangeNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/ExchangeNode.java
@@ -180,14 +180,21 @@ public class ExchangeNode extends PlanNode {
     @Override
     public Pair<PlanNode, LocalExchangeType> enforceAndDeriveLocalExchange(PlanTranslatorContext translatorContext,
             PlanNode parent, LocalExchangeTypeRequire parentRequire) {
-        // Mirror BE's ExchangeSourceOperatorX::required_data_distribution():
-        // serial Exchange returns NOOP. This covers both:
-        // 1. UNPARTITIONED exchange (naturally serial) → already NOOP in else branch
-        // 2. HASH/BUCKET exchange made serial by use_serial_exchange=true → override to NOOP
-        // Mirror BE ExchangeSourceOperatorX::required_data_distribution():
-        // serial → NOOP; else HASH/BUCKET/NOOP based on partition type.
-        if (isSerialOperator()) {
-            return Pair.of(this, LocalExchangeType.NOOP);
+        // Must match the BE serial condition in toThrift(): isSerialOperator() && useSerialSource().
+        // Only insert PASSTHROUGH when the exchange will actually be serial on the BE.
+        // 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()
+                && fragment != null
+                && fragment.useSerialSource(ConnectContext.get());
+        if (willBeSerialOnBe) {
+            // Serial exchange → 1 task. Insert PASSTHROUGH to fan out to N tasks.
+            // Parent nodes (HashJoin, etc.) may add PASS_TO_ONE on top if they need
+            // single-builder semantics (e.g., broadcast join build side).
+            PlanNode pt = new LocalExchangeNode(translatorContext.nextPlanNodeId(),
+                    this, LocalExchangeType.PASSTHROUGH, null);
+            return Pair.of(pt, LocalExchangeType.PASSTHROUGH);
         } else if (partitionType == TPartitionType.HASH_PARTITIONED) {
             return Pair.of(this, LocalExchangeType.GLOBAL_EXECUTION_HASH_SHUFFLE);
         } else if (partitionType == TPartitionType.BUCKET_SHFFULE_HASH_PARTITIONED) {


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org