From b935fd0e7df92c35dffd75500fbb43837b77975d Mon Sep 17 00:00:00 2001 From: starocean999 <40539150+starocean999@users.noreply.github.com> Date: Fri, 23 Dec 2022 16:44:44 +0800 Subject: [PATCH] [fix](fe)fix bug of the bucket shuffle join is not recognized (#15255) * [fix](fe)fix bug of the bucket shuffle join is not recognized * use broadcast join for empty table --- .../doris/planner/DistributedPlanner.java | 2 +- .../org/apache/doris/planner/PlanNode.java | 17 ++++ .../test_bucket_shuffle_join.groovy | 81 +++++++++++++++++++ 3 files changed, 99 insertions(+), 1 deletion(-) create mode 100644 regression-test/suites/correctness_p0/test_bucket_shuffle_join.groovy diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/DistributedPlanner.java b/fe/fe-core/src/main/java/org/apache/doris/planner/DistributedPlanner.java index 6382787d1b..6c35b62006 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/planner/DistributedPlanner.java +++ b/fe/fe-core/src/main/java/org/apache/doris/planner/DistributedPlanner.java @@ -656,7 +656,7 @@ public class DistributedPlanner { continue; } - SlotRef leftSlot = lhsJoinExpr.unwrapSlotRef(); + SlotRef leftSlot = node.getChild(0).findSrcSlotRef(lhsJoinExpr.getSrcSlotRef()); if (leftSlot.getTable() instanceof OlapTable && leftScanNode.desc.getSlots().contains(leftSlot.getDesc())) { // table name in SlotRef is not the really name. `select * from test as t` diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java index 548d2fb174..f54ed59ff3 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java @@ -32,6 +32,7 @@ import org.apache.doris.analysis.SlotRef; import org.apache.doris.analysis.TupleDescriptor; import org.apache.doris.analysis.TupleId; import org.apache.doris.catalog.Function; +import org.apache.doris.catalog.OlapTable; import org.apache.doris.catalog.Type; import org.apache.doris.common.AnalysisException; import org.apache.doris.common.NotImplementedException; @@ -926,6 +927,22 @@ public abstract class PlanNode extends TreeNode implements PlanStats { return null; } + public SlotRef findSrcSlotRef(SlotRef slotRef) { + if (slotRef.getTable() instanceof OlapTable) { + return slotRef; + } + if (this instanceof HashJoinNode) { + HashJoinNode hashJoinNode = (HashJoinNode) this; + SlotRef inputSlotRef = hashJoinNode.getMappedInputSlotRef(slotRef); + if (inputSlotRef != null) { + return hashJoinNode.getChild(0).findSrcSlotRef(inputSlotRef); + } else { + return slotRef; + } + } + return slotRef; + } + protected void addRuntimeFilter(RuntimeFilter filter) { runtimeFilters.add(filter); } diff --git a/regression-test/suites/correctness_p0/test_bucket_shuffle_join.groovy b/regression-test/suites/correctness_p0/test_bucket_shuffle_join.groovy new file mode 100644 index 0000000000..febd57a353 --- /dev/null +++ b/regression-test/suites/correctness_p0/test_bucket_shuffle_join.groovy @@ -0,0 +1,81 @@ +// 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. + +suite("test_bucket_shuffle_join") { + sql """ DROP TABLE IF EXISTS `test_colo1` """ + sql """ DROP TABLE IF EXISTS `test_colo2` """ + sql """ DROP TABLE IF EXISTS `test_colo3` """ + sql """ + CREATE TABLE `test_colo1` ( + `id` varchar(64) NULL, + `name` varchar(64) NULL, + `age` int NULL + ) ENGINE=OLAP + DUPLICATE KEY(`id`,`name`) + COMMENT 'OLAP' + DISTRIBUTED BY HASH(`id`,`name`) BUCKETS 4 + PROPERTIES ( + "replication_allocation" = "tag.location.default: 1", + "in_memory" = "false", + "storage_format" = "V2", + "disable_auto_compaction" = "false" + ); + """ + sql """ + CREATE TABLE `test_colo2` ( + `id` varchar(64) NULL, + `name` varchar(64) NULL, + `age` int NULL + ) ENGINE=OLAP + DUPLICATE KEY(`id`,`name`) + COMMENT 'OLAP' + DISTRIBUTED BY HASH(`id`,`name`) BUCKETS 5 + PROPERTIES ( + "replication_allocation" = "tag.location.default: 1", + "in_memory" = "false", + "storage_format" = "V2", + "disable_auto_compaction" = "false" + ); + """ + + sql """ + CREATE TABLE `test_colo3` ( + `id` varchar(64) NULL, + `name` varchar(64) NULL, + `age` int NULL + ) ENGINE=OLAP + DUPLICATE KEY(`id`,`name`) + COMMENT 'OLAP' + DISTRIBUTED BY HASH(`id`,`name`) BUCKETS 6 + PROPERTIES ( + "replication_allocation" = "tag.location.default: 1", + "in_memory" = "false", + "storage_format" = "V2", + "disable_auto_compaction" = "false" + ); + """ + + sql """insert into test_colo1 values('1','a',12);""" + sql """insert into test_colo2 values('1','a',12);""" + sql """insert into test_colo3 values('1','a',12);""" + + explain { + sql("select a.id,a.name,b.id,b.name,c.id,c.name from test_colo1 a inner join test_colo2 b on a.id = b.id and a.name = b.name inner join test_colo3 c on a.id=c.id and a.name= c.name") + contains "4:VHASH JOIN\n | join op: INNER JOIN(BUCKET_SHUFFLE)" + contains "2:VHASH JOIN\n | join op: INNER JOIN(BUCKET_SHUFFLE)" + } +}