This is an automated email from the ASF dual-hosted git repository.
924060929 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new b747c34914a [fix](local shuffle) address all receiver instances for a
non-serial exchange (#65348)
b747c34914a is described below
commit b747c34914afd25a853c76e97fe8a531df6f5b93
Author: 924060929 <[email protected]>
AuthorDate: Wed Jul 8 14:37:41 2026 +0800
[fix](local shuffle) address all receiver instances for a non-serial
exchange (#65348)
Related PR: #64793
Problem Summary:
PR #64793 decoupled an exchange's serial flag from the fragment's serial
scan under the FE local-shuffle planner, so a local-shuffle fragment can
now host a **non-serial** RANDOM/HASH exchange whose BE receiver runs on
every instance and waits for the full sender set.
However `DistributePlanner.filterInstancesWhichCanReceiveDataFromRemote`
still funnels the sender's destinations to the first instance per worker
whenever the receiver uses local shuffle (keyed on
`LocalShuffleAssignedJob`), and #64793 only added destination spreading
for `BUCKET_SHUFFLE`, not for RANDOM/HASH. So the sender addresses only
instance 0 while the non-first instances each build a receiver expecting
the full sender set — they wait forever for an EOS that never arrives,
and a query such as an MV partition-union rewrite (a RANDOM exchange
feeding a UNION) hangs until query timeout.
This PR gates the funnel on `ExchangeNode.isSerialOperatorOnBe` instead
of `LocalShuffleAssignedJob`: only a serial exchange runs a single
receiver per worker and may funnel to it; a non-serial exchange must
address every receiver instance. `BUCKET_SHUFFLE` is unchanged (it
re-spreads by bucket in `getDestinationsByBuckets`). This mirrors the
`!isSerialOperatorOnBe` gate already used there.
Verified end-to-end on a cloud single-BE cluster: the co-located `UNION
<- BUCKETED AGGREGATE <- OlapScan` plan hangs at query timeout before
the fix and returns in ~1s after it. Added
`DistributePlannerReceiverDestinationTest` covering the serial /
non-serial / shared-broadcast-hash-table cases.
### Release note
Fix a query hang (until timeout) where a non-serial RANDOM/HASH exchange
feeding a local-shuffle receiver fragment addressed only the first
instance per worker, leaving the other instances waiting forever for
sender EOS.
---
.../trees/plans/distribute/DistributePlanner.java | 17 ++-
.../DistributePlannerReceiverDestinationTest.java | 120 +++++++++++++++++++++
2 files changed, 132 insertions(+), 5 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/DistributePlanner.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/DistributePlanner.java
index fd1ff52f5f9..5568d332f23 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/DistributePlanner.java
+++
b/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.AssignedJob;
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.qe.StmtExecutor;
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;
@@ -260,13 +260,20 @@ public class DistributePlanner {
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());
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/distribute/DistributePlannerReceiverDestinationTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/distribute/DistributePlannerReceiverDestinationTest.java
new file mode 100644
index 00000000000..220062dfb95
--- /dev/null
+++
b/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);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
