Skip to content

Commit ae2ba6f

Browse files
924060929924060929
authored
[fix](coordinator) fix legacy coordinator use 1 instance for shuffle fragment when children has multiple instances (#59295)
fix legacy coordinator use 1 instance for shuffle fragment when children has multiple instances, introduced by #39999
1 parent f57e8f3 commit ae2ba6f

2 files changed

Lines changed: 109 additions & 5 deletions

File tree

‎fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java‎

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1988,15 +1988,17 @@ protected void computeFragmentHosts() throws Exception {
19881988
private int findMaxParallelFragmentIndex(PlanFragment fragment) {
19891989
Preconditions.checkState(!fragment.getChildren().isEmpty(), "fragment has no children");
19901990

1991-
// exclude broadcast join right side's child fragments
1992-
List<PlanFragment> childFragmentCandidates = fragment.getChildren().stream()
1993-
.filter(e -> e.getOutputPartition() != DataPartition.UNPARTITIONED)
1994-
.collect(Collectors.toList());
1991+
List<PlanFragment> childFragmentCandidates = fragment.getChildren();
19951992

19961993
int maxParallelism = 0;
19971994
int maxParaIndex = 0;
19981995
for (int i = 0; i < childFragmentCandidates.size(); i++) {
1999-
PlanFragmentId childFragmentId = childFragmentCandidates.get(i).getFragmentId();
1996+
PlanFragment planFragment = childFragmentCandidates.get(i);
1997+
// exclude broadcast join right side's child fragments
1998+
if (planFragment.getOutputPartition() == DataPartition.UNPARTITIONED) {
1999+
continue;
2000+
}
2001+
PlanFragmentId childFragmentId = planFragment.getFragmentId();
20002002
int currentChildFragmentParallelism = fragmentExecParamsMap.get(childFragmentId).instanceExecParams.size();
20012003
if (currentChildFragmentParallelism > maxParallelism) {
20022004
maxParallelism = currentChildFragmentParallelism;

‎fe/fe-core/src/test/java/org/apache/doris/qe/OldCoordinatorTest.java‎

Lines changed: 102 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,102 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
package org.apache.doris.qe;
19+
20+
import org.apache.doris.nereids.rules.RuleType;
21+
import org.apache.doris.planner.OlapScanNode;
22+
import org.apache.doris.planner.PlanFragment;
23+
import org.apache.doris.planner.PlanFragmentId;
24+
import org.apache.doris.planner.PlanNode;
25+
import org.apache.doris.utframe.TestWithFeService;
26+
27+
import org.junit.jupiter.api.Assertions;
28+
import org.junit.jupiter.api.Test;
29+
30+
import java.util.List;
31+
import java.util.Map;
32+
import java.util.concurrent.atomic.AtomicBoolean;
33+
34+
public class OldCoordinatorTest extends TestWithFeService {
35+
@Override
36+
protected void runBeforeAll() throws Exception {
37+
createDatabase("test");
38+
createTable("create table test.tbl(id int, value int) distributed by hash(id) buckets 1 properties('replication_num'='1')");
39+
createTable("create table test.tbl2(id int, value int) distributed by hash(id) buckets 1 properties('replication_num'='1')");
40+
createTable("create table test.tbl3(id int, value int) distributed by hash(id) buckets 10 properties('replication_num'='1')");
41+
}
42+
43+
@Test
44+
public void test() throws Exception {
45+
connectContext.getSessionVariable().setDisableNereidsRules(RuleType.PRUNE_EMPTY_PARTITION.name());
46+
connectContext.getSessionVariable().setDisableJoinReorder(true);
47+
connectContext.getSessionVariable().parallelPipelineTaskNum = 2;
48+
connectContext.getSessionVariable().setEnableLocalShuffle(false);
49+
StmtExecutor stmtExecutor = getSqlStmtExecutor(
50+
"select *\n"
51+
+ "from (\n" // left most has exchange trigger shuffle instance logic
52+
+ " select a.value\n"
53+
+ " from (select value from test.tbl group by value)a\n"
54+
+ " join[broadcast] test.tbl2\n"
55+
+ " on a.value=tbl2.id\n"
56+
+ ")b\n"
57+
+ "join[shuffle] test.tbl3 c\n" // 2 instances, should use as the shuffle instances
58+
+ "on b.value = c.id\n"
59+
+ "join[shuffle] (\n" // 1 instance, skip
60+
+ " select a.value\n"
61+
+ " from (select value from test.tbl group by value)a\n"
62+
+ " join[broadcast] test.tbl2\n"
63+
+ " on a.value=tbl2.id\n"
64+
+ ")d\n"
65+
+ "on c.id = d.value\n"
66+
+ "join[shuffle] test.tbl e\n" // 1 instance, skip
67+
+ "on d.value=e.id");
68+
69+
AtomicBoolean shuffleFragmentHasMultiInstances = new AtomicBoolean(false);
70+
new Coordinator(connectContext, stmtExecutor.planner()) {
71+
public void test() throws Exception {
72+
super.processFragmentAssignmentAndParams();
73+
74+
Map<PlanFragmentId, FragmentExecParams> fragmentExecParamsMap = getFragmentExecParamsMap();
75+
PlanFragmentId scanTbl3FragmentId = null;
76+
for (FragmentExecParams fragmentExecParams : fragmentExecParamsMap.values()) {
77+
PlanNode planRoot = fragmentExecParams.fragment.getPlanRoot();
78+
if (planRoot instanceof OlapScanNode && ((OlapScanNode) planRoot).getOlapTable().getName()
79+
.equals("tbl3")) {
80+
scanTbl3FragmentId = fragmentExecParams.fragment.getId();
81+
break;
82+
}
83+
}
84+
85+
for (FragmentExecParams fragmentExecParams : fragmentExecParamsMap.values()) {
86+
List<FInstanceExecParam> instances = fragmentExecParams.instanceExecParams;
87+
boolean childScanTbl3 = false;
88+
for (PlanFragment child : fragmentExecParams.fragment.getChildren()) {
89+
if (child.getFragmentId().equals(scanTbl3FragmentId)) {
90+
childScanTbl3 = true;
91+
break;
92+
}
93+
}
94+
if (childScanTbl3 && instances.size() >= 2) {
95+
shuffleFragmentHasMultiInstances.set(true);
96+
}
97+
}
98+
}
99+
}.test();
100+
Assertions.assertTrue(shuffleFragmentHasMultiInstances.get());
101+
}
102+
}

0 commit comments

Comments
 (0)