branch-4.2: [fix](local shuffle) address all receiver instances for a non-serial exchange #65348

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

@@ -30,7 +30,6 @@
 import org.apache.doris.nereids.trees.plans.distribute.worker.job.AssignedJobBuilder;
 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;
@@ -50,6 +49,7 @@
 import org.apache.doris.thrift.TPartitionType;
 import org.apache.doris.thrift.TUniqueId;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Throwables;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.LinkedHashMultimap;
@@ -265,13 +265,20 @@ private boolean isEnableLocalShufflePlanner() {
         return connectContext != null && connectContext.getSessionVariable().isEnableLocalShufflePlanner();
     }
 
-    private List<AssignedJob> filterInstancesWhichCanReceiveDataFromRemote(
+    @VisibleForTesting
+    List<AssignedJob> filterInstancesWhichCanReceiveDataFromRemote(
             PipelineDistributedPlan receiverPlan,
             boolean enableShareHashTableForBroadcastJoin,
             ExchangeNode linkNode) {
-        boolean useLocalShuffle = receiverPlan.getInstanceJobs().stream()
-                .anyMatch(LocalShuffleAssignedJob.class::isInstance);
-        if (useLocalShuffle) {
+        // Funnel to the first instance per worker only when the exchange runs a single receiver
+        // per worker on BE, i.e. it is serial. A non-serial exchange builds a live receiver on
+        // every instance, each waiting for the full sender set, so the sender must address all of
+        // them; funneling would leave the non-first instances blocked forever on an EOS that never
+        // arrives. This must not key on LocalShuffleAssignedJob: once the FE local-shuffle planner
+        // decoupled an exchange's serial flag from the fragment's serial scan, a local-shuffle
+        // fragment can host a non-serial RANDOM/HASH exchange, and only the BUCKET_SHUFFLE path
+        // re-spreads its destinations (see getDestinationsByBuckets).
+        if (linkNode.isSerialOperatorOnBe(statementContext.getConnectContext())) {
             return getFirstInstancePerWorker(receiverPlan.getInstanceJobs());
         } else if (enableShareHashTableForBroadcastJoin && linkNode.isRightChildOfBroadcastHashJoin()) {
             return getFirstInstancePerWorker(receiverPlan.getInstanceJobs());

fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/distribute/DistributePlannerReceiverDestinationTest.java

@@ -0,0 +1,120 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.trees.plans.distribute;
+
+import org.apache.doris.nereids.StatementContext;
+import org.apache.doris.nereids.trees.plans.distribute.worker.DistributedPlanWorker;
+import org.apache.doris.nereids.trees.plans.distribute.worker.job.AssignedJob;
+import org.apache.doris.nereids.trees.plans.distribute.worker.job.LocalShuffleAssignedJob;
+import org.apache.doris.planner.ExchangeNode;
+import org.apache.doris.planner.PlanFragment;
+import org.apache.doris.qe.ConnectContext;
+
+import com.google.common.collect.Lists;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.List;
+
+/**
+ * Verifies that DistributePlanner picks the receiver instances a remote sender addresses
+ * based on whether the exchange is serial on BE -- not on whether the receiver fragment
+ * happens to use local shuffle.
+ *
+ * A non-serial exchange runs a live receiver on every instance, so the sender must address
+ * them all. Funneling such an exchange to the first instance per worker leaves the other
+ * instances waiting forever for an EOS that no sender sends (query hangs until timeout).
+ */
+public class DistributePlannerReceiverDestinationTest {
+
+    private DistributePlanner newPlanner() {
+        List<PlanFragment> noFragments = Lists.newArrayList();
+        return new DistributePlanner(Mockito.mock(StatementContext.class), noFragments, false, false);
+    }
+
+    private AssignedJob instanceOn(DistributedPlanWorker worker) {
+        LocalShuffleAssignedJob instance = Mockito.mock(LocalShuffleAssignedJob.class);
+        Mockito.when(instance.getAssignedWorker()).thenReturn(worker);
+        return instance;
+    }
+
+    private ExchangeNode exchange(boolean serialOnBe, boolean rightChildOfBroadcastJoin) {
+        ExchangeNode node = Mockito.mock(ExchangeNode.class);
+        Mockito.when(node.isSerialOperatorOnBe(Mockito.nullable(ConnectContext.class))).thenReturn(serialOnBe);
+        Mockito.when(node.isRightChildOfBroadcastHashJoin()).thenReturn(rightChildOfBroadcastJoin);
+        return node;
+    }
+
+    private PipelineDistributedPlan receiverWith(List<AssignedJob> instances) {
+        PipelineDistributedPlan plan = Mockito.mock(PipelineDistributedPlan.class);
+        Mockito.when(plan.getInstanceJobs()).thenReturn(instances);
+        return plan;
+    }
+
+    @Test
+    public void nonSerialExchangeAddressesEveryReceiverInstance() {
+        // 4 receiver instances across 2 workers (e.g. a bucket-to-hash upgraded fragment)
+        DistributedPlanWorker w0 = Mockito.mock(DistributedPlanWorker.class);
+        DistributedPlanWorker w1 = Mockito.mock(DistributedPlanWorker.class);
+        AssignedJob i0 = instanceOn(w0);
+        AssignedJob i1 = instanceOn(w0);
+        AssignedJob i2 = instanceOn(w1);
+        AssignedJob i3 = instanceOn(w1);
+        List<AssignedJob> all = Lists.newArrayList(i0, i1, i2, i3);
+
+        List<AssignedJob> destinations = newPlanner().filterInstancesWhichCanReceiveDataFromRemote(
+                receiverWith(all), false, exchange(false, false));
+
+        // Non-serial exchange: every instance owns a live receiver, so all must be addressed.
+        Assertions.assertEquals(all, destinations);
+    }
+
+    @Test
+    public void serialExchangeFunnelsToFirstInstancePerWorker() {
+        DistributedPlanWorker w0 = Mockito.mock(DistributedPlanWorker.class);
+        DistributedPlanWorker w1 = Mockito.mock(DistributedPlanWorker.class);
+        AssignedJob i0 = instanceOn(w0);
+        AssignedJob i1 = instanceOn(w0);
+        AssignedJob i2 = instanceOn(w1);
+        AssignedJob i3 = instanceOn(w1);
+        List<AssignedJob> all = Lists.newArrayList(i0, i1, i2, i3);
+
+        List<AssignedJob> destinations = newPlanner().filterInstancesWhichCanReceiveDataFromRemote(
+                receiverWith(all), false, exchange(true, false));
+
+        // Serial exchange: BE runs a single receiver per worker, funnel to the first instance.
+        Assertions.assertEquals(Lists.newArrayList(i0, i2), destinations);
+    }
+
+    @Test
+    public void broadcastShareHashTableFunnelsEvenWhenNonSerial() {
+        DistributedPlanWorker w0 = Mockito.mock(DistributedPlanWorker.class);
+        DistributedPlanWorker w1 = Mockito.mock(DistributedPlanWorker.class);
+        AssignedJob i0 = instanceOn(w0);
+        AssignedJob i1 = instanceOn(w0);
+        AssignedJob i2 = instanceOn(w1);
+        List<AssignedJob> all = Lists.newArrayList(i0, i1, i2);
+
+        List<AssignedJob> destinations = newPlanner().filterInstancesWhichCanReceiveDataFromRemote(
+                receiverWith(all), true, exchange(false, true));
+
+        // Shared-hash-table broadcast build still funnels one build per worker.
+        Assertions.assertEquals(Lists.newArrayList(i0, i2), destinations);
+    }
+}