[BugFix] isLocalBucketShuffleJoin return wrong result (#51954)

Signed-off-by: satanson <[email protected]>
(cherry picked from commit 1077a6d)

fe/fe-core/src/main/java/com/starrocks/qe/scheduler/dag/ExecutionFragment.java

@@ -388,19 +388,19 @@ private boolean isLocalBucketShuffleJoin(PlanNode root) {
             return false;
         }
 
+        boolean hasBucketShuffle = false;
         if (root instanceof JoinNode) {
             JoinNode joinNode = (JoinNode) root;
             if (joinNode.isLocalHashBucket()) {
-                isRightOrFullBucketShuffle = joinNode.getJoinOp().isFullOuterJoin() || joinNode.getJoinOp().isRightJoin();
-                return true;
+                hasBucketShuffle = true;
+                isRightOrFullBucketShuffle |= joinNode.getJoinOp().isFullOuterJoin() || joinNode.getJoinOp().isRightJoin();
             }
         }
 
-        boolean childHasBucketShuffle = false;
         for (PlanNode child : root.getChildren()) {
-            childHasBucketShuffle |= isLocalBucketShuffleJoin(child);
+            hasBucketShuffle |= isLocalBucketShuffleJoin(child);
         }
 
-        return childHasBucketShuffle;
+        return hasBucketShuffle;
     }
 }

fe/fe-core/src/test/java/com/starrocks/qe/TestBucketShuffleRightJoin.java

@@ -0,0 +1,88 @@
+// Copyright 2021-present StarRocks, Inc. All rights reserved.
+//
+// Licensed 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
+//
+//     https://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 com.starrocks.qe;
+
+import com.starrocks.common.FeConstants;
+import com.starrocks.common.Pair;
+import com.starrocks.qe.scheduler.dag.ExecutionFragment;
+import com.starrocks.sql.plan.ExecPlan;
+import com.starrocks.statistic.StatsConstants;
+import com.starrocks.utframe.StarRocksAssert;
+import com.starrocks.utframe.UtFrameUtils;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.Arrays;
+
+import static com.starrocks.sql.optimizer.statistics.CachedStatisticStorageTest.DEFAULT_CREATE_TABLE_TEMPLATE;
+
+public class TestBucketShuffleRightJoin {
+
+    private static StarRocksAssert starRocksAssert;
+
+    @BeforeClass
+    public static void setUp() {
+        UtFrameUtils.createMinStarRocksCluster();
+        ConnectContext ctx = UtFrameUtils.createDefaultCtx();
+        FeConstants.runningUnitTest = true;
+        starRocksAssert = new StarRocksAssert(ctx);
+        try {
+            starRocksAssert.withDatabase(StatsConstants.STATISTICS_DB_NAME)
+                    .useDatabase(StatsConstants.STATISTICS_DB_NAME)
+                    .withTable(DEFAULT_CREATE_TABLE_TEMPLATE);
+
+            starRocksAssert.withDatabase("test").useDatabase("test");
+            String tableFmt = "CREATE TABLE `%s` (\n" +
+                    "  `c0` bigint(20) NOT NULL COMMENT \"\",\n" +
+                    "  `c1` bigint(20) NOT NULL COMMENT \"\",\n" +
+                    "  `c2` bigint(20) NOT NULL COMMENT \"\"\n" +
+                    ") ENGINE=OLAP \n" +
+                    "DUPLICATE KEY(`c0`)\n" +
+                    "COMMENT \"OLAP\"\n" +
+                    "DISTRIBUTED BY HASH(`c0`) BUCKETS 9 \n" +
+                    "PROPERTIES (\n" +
+                    "\"compression\" = \"LZ4\",\n" +
+                    "\"replication_num\" = \"1\"\n" +
+                    ");";
+            for (String tableName : Arrays.asList("t0", "t1", "t2")) {
+                String createTblSql = String.format(tableFmt, tableName);
+                starRocksAssert.withTable(createTblSql);
+            }
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+        FeConstants.runningUnitTest = true;
+    }
+
+    @Test
+    public void test() throws Exception {
+        ConnectContext ctx = starRocksAssert.getCtx();
+        String sql = "select t0.c0,t0.c1,t0.c2,t1.c2 as ab from " +
+                "(select * from t1 where c0 in (8)) t1 right outer join[bucket] " +
+                "(select if(murmur_hash3_32(c0)=0,c0,NULL) as c0,c1,c2 from t0) t0 on t0.c0 = t1.c0 " +
+                "left join[bucket] t2 on t2.c0 = t1.c0";
+        Pair<String, ExecPlan> explainAndExecPlan = UtFrameUtils.getPlanAndFragment(ctx, sql);
+        String plan = explainAndExecPlan.first;
+        ExecPlan execPlan = explainAndExecPlan.second;
+        DefaultCoordinator coord = new DefaultCoordinator.Factory().createQueryScheduler(
+                ctx, execPlan.getFragments(), execPlan.getScanNodes(), execPlan.getDescTbl().toThrift());
+        coord.prepareExec();
+
+        boolean bucketShuffleRightJoinPresent = coord.getExecutionDAG().getFragmentsInPreorder()
+                .stream().anyMatch(ExecutionFragment::isRightOrFullBucketShuffle);
+        Assert.assertTrue(plan, bucketShuffleRightJoinPresent);
+    }
+}

test/sql/test_bucket_shuffle_right_join/R/test_bucket_shuffle_right_join

@@ -0,0 +1,270 @@
+-- name: test_bucket_shuffle_right_join
+ DROP TABLE if exists t0;
+-- result:
+-- !result
+ CREATE TABLE if not exists t0
+ (
+ c0 BIGINT NOT NULL,
+ c1 BIGINT NOT NULL,
+ c2 BIGINT NOT NULL 
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`c0` )
+ COMMENT "OLAP"
+ DISTRIBUTED BY HASH(`c0` ) BUCKETS 9
+ PROPERTIES(
+ "replication_num" = "1",
+ "in_memory" = "false",
+ "storage_format" = "default" 
+ );
+-- result:
+-- !result
+ DROP TABLE if exists t1;
+-- result:
+-- !result
+ CREATE TABLE if not exists t1
+ (
+ c0 BIGINT NOT NULL,
+ c1 BIGINT NOT NULL,
+ c2 BIGINT NOT NULL 
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`c0` )
+ COMMENT "OLAP"
+ DISTRIBUTED BY HASH(`c0` ) BUCKETS 9
+ PROPERTIES(
+ "replication_num" = "1",
+ "in_memory" = "false",
+ "storage_format" = "default" 
+ );
+-- result:
+-- !result
+ DROP TABLE if exists t2;
+-- result:
+-- !result
+ CREATE TABLE if not exists t2
+ (
+ c0 BIGINT NOT NULL,
+ c1 BIGINT NOT NULL,
+ c2 BIGINT NOT NULL 
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`c0` )
+ COMMENT "OLAP"
+ DISTRIBUTED BY HASH(`c0` ) BUCKETS 9
+ PROPERTIES(
+ "replication_num" = "1",
+ "in_memory" = "false",
+ "storage_format" = "default" 
+ );
+-- result:
+-- !result
+INSERT INTO t0
+(c0, c1, c2)
+VALUES
+('0', '0', '0'),
+('1', '1', '1'),
+('2', '2', '2'),
+('3', '3', '3'),
+('4', '4', '4'),
+('5', '5', '5'),
+('6', '6', '6'),
+('7', '7', '7'),
+('8', '8', '8'),
+('9', '9', '9');
+-- result:
+-- !result
+INSERT INTO t1
+(c0, c1, c2)
+select * from t0;
+-- result:
+-- !result
+INSERT INTO t2
+(c0, c1, c2)
+select * from t0;
+-- result:
+-- !result
+ DROP TABLE if exists r;
+-- result:
+-- !result
+ CREATE TABLE if not exists r
+ (
+ fp BIGINT  NULL 
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`fp` )
+ COMMENT "OLAP"
+ DISTRIBUTED BY HASH(`fp` ) BUCKETS 1
+ PROPERTIES(
+ "replication_num" = "1",
+ "in_memory" = "false",
+ "storage_format" = "default" 
+ );
+-- result:
+-- !result
+truncate table r;
+-- result:
+-- !result
+INSERT INTO r
+(fp)
+select  (sum(murmur_hash3_32(ifnull(c0,0))+murmur_hash3_32(ifnull(c1,0))+murmur_hash3_32(ifnull(c2,0))+murmur_hash3_32(ifnull(ab,0)))) as fingerprint from (select t0.c0,t0.c1,t0.c2,t1.c2 as ab from (select * from t1 where c0 in (0)) t1 right outer join[bucket] (select if(murmur_hash3_32(c0)=0,c0,NULL) as c0,c1,c2 from t0) t0 on t0.c0 = t1.c0 left join[bucket] t2 on t2.c0 = t1.c0) as t;
+-- result:
+-- !result
+INSERT INTO r
+(fp)
+select  (sum(murmur_hash3_32(ifnull(c0,0))+murmur_hash3_32(ifnull(c1,0))+murmur_hash3_32(ifnull(c2,0))+murmur_hash3_32(ifnull(ab,0)))) as fingerprint from (select t0.c0,t0.c1,t0.c2,t1.c2 as ab from (select * from t1 where c0 in (0)) t1 right outer join[shuffle] (select if(murmur_hash3_32(c0)=0,c0,NULL) as c0,c1,c2 from t0) t0 on t0.c0 = t1.c0 left join[shuffle] t2 on t2.c0 = t1.c0) as t;
+-- result:
+-- !result
+select assert_true(count(fp)=2), assert_true(count(distinct fp)=1) from r;
+-- result:
+1	1
+-- !result
+truncate table r;
+-- result:
+-- !result
+INSERT INTO r
+(fp)
+select  (sum(murmur_hash3_32(ifnull(c0,0))+murmur_hash3_32(ifnull(c1,0))+murmur_hash3_32(ifnull(c2,0))+murmur_hash3_32(ifnull(ab,0)))) as fingerprint from (select t0.c0,t0.c1,t0.c2,t1.c2 as ab from (select * from t1 where c0 in (1)) t1 right outer join[bucket] (select if(murmur_hash3_32(c0)=0,c0,NULL) as c0,c1,c2 from t0) t0 on t0.c0 = t1.c0 left join[bucket] t2 on t2.c0 = t1.c0) as t;
+-- result:
+-- !result
+INSERT INTO r
+(fp)
+select  (sum(murmur_hash3_32(ifnull(c0,0))+murmur_hash3_32(ifnull(c1,0))+murmur_hash3_32(ifnull(c2,0))+murmur_hash3_32(ifnull(ab,0)))) as fingerprint from (select t0.c0,t0.c1,t0.c2,t1.c2 as ab from (select * from t1 where c0 in (1)) t1 right outer join[shuffle] (select if(murmur_hash3_32(c0)=0,c0,NULL) as c0,c1,c2 from t0) t0 on t0.c0 = t1.c0 left join[shuffle] t2 on t2.c0 = t1.c0) as t;
+-- result:
+-- !result
+select assert_true(count(fp)=2), assert_true(count(distinct fp)=1) from r;
+-- result:
+1	1
+-- !result
+truncate table r;
+-- result:
+-- !result
+INSERT INTO r
+(fp)
+select  (sum(murmur_hash3_32(ifnull(c0,0))+murmur_hash3_32(ifnull(c1,0))+murmur_hash3_32(ifnull(c2,0))+murmur_hash3_32(ifnull(ab,0)))) as fingerprint from (select t0.c0,t0.c1,t0.c2,t1.c2 as ab from (select * from t1 where c0 in (2)) t1 right outer join[bucket] (select if(murmur_hash3_32(c0)=0,c0,NULL) as c0,c1,c2 from t0) t0 on t0.c0 = t1.c0 left join[bucket] t2 on t2.c0 = t1.c0) as t;
+-- result:
+-- !result
+INSERT INTO r
+(fp)
+select  (sum(murmur_hash3_32(ifnull(c0,0))+murmur_hash3_32(ifnull(c1,0))+murmur_hash3_32(ifnull(c2,0))+murmur_hash3_32(ifnull(ab,0)))) as fingerprint from (select t0.c0,t0.c1,t0.c2,t1.c2 as ab from (select * from t1 where c0 in (2)) t1 right outer join[shuffle] (select if(murmur_hash3_32(c0)=0,c0,NULL) as c0,c1,c2 from t0) t0 on t0.c0 = t1.c0 left join[shuffle] t2 on t2.c0 = t1.c0) as t;
+-- result:
+-- !result
+select assert_true(count(fp)=2), assert_true(count(distinct fp)=1) from r;
+-- result:
+1	1
+-- !result
+truncate table r;
+-- result:
+-- !result
+INSERT INTO r
+(fp)
+select  (sum(murmur_hash3_32(ifnull(c0,0))+murmur_hash3_32(ifnull(c1,0))+murmur_hash3_32(ifnull(c2,0))+murmur_hash3_32(ifnull(ab,0)))) as fingerprint from (select t0.c0,t0.c1,t0.c2,t1.c2 as ab from (select * from t1 where c0 in (3)) t1 right outer join[bucket] (select if(murmur_hash3_32(c0)=0,c0,NULL) as c0,c1,c2 from t0) t0 on t0.c0 = t1.c0 left join[bucket] t2 on t2.c0 = t1.c0) as t;
+-- result:
+-- !result
+INSERT INTO r
+(fp)
+select  (sum(murmur_hash3_32(ifnull(c0,0))+murmur_hash3_32(ifnull(c1,0))+murmur_hash3_32(ifnull(c2,0))+murmur_hash3_32(ifnull(ab,0)))) as fingerprint from (select t0.c0,t0.c1,t0.c2,t1.c2 as ab from (select * from t1 where c0 in (3)) t1 right outer join[shuffle] (select if(murmur_hash3_32(c0)=0,c0,NULL) as c0,c1,c2 from t0) t0 on t0.c0 = t1.c0 left join[shuffle] t2 on t2.c0 = t1.c0) as t;
+-- result:
+-- !result
+select assert_true(count(fp)=2), assert_true(count(distinct fp)=1) from r;
+-- result:
+1	1
+-- !result
+truncate table r;
+-- result:
+-- !result
+INSERT INTO r
+(fp)
+select  (sum(murmur_hash3_32(ifnull(c0,0))+murmur_hash3_32(ifnull(c1,0))+murmur_hash3_32(ifnull(c2,0))+murmur_hash3_32(ifnull(ab,0)))) as fingerprint from (select t0.c0,t0.c1,t0.c2,t1.c2 as ab from (select * from t1 where c0 in (4)) t1 right outer join[bucket] (select if(murmur_hash3_32(c0)=0,c0,NULL) as c0,c1,c2 from t0) t0 on t0.c0 = t1.c0 left join[bucket] t2 on t2.c0 = t1.c0) as t;
+-- result:
+-- !result
+INSERT INTO r
+(fp)
+select  (sum(murmur_hash3_32(ifnull(c0,0))+murmur_hash3_32(ifnull(c1,0))+murmur_hash3_32(ifnull(c2,0))+murmur_hash3_32(ifnull(ab,0)))) as fingerprint from (select t0.c0,t0.c1,t0.c2,t1.c2 as ab from (select * from t1 where c0 in (4)) t1 right outer join[shuffle] (select if(murmur_hash3_32(c0)=0,c0,NULL) as c0,c1,c2 from t0) t0 on t0.c0 = t1.c0 left join[shuffle] t2 on t2.c0 = t1.c0) as t;
+-- result:
+-- !result
+select assert_true(count(fp)=2), assert_true(count(distinct fp)=1) from r;
+-- result:
+1	1
+-- !result
+truncate table r;
+-- result:
+-- !result
+INSERT INTO r
+(fp)
+select  (sum(murmur_hash3_32(ifnull(c0,0))+murmur_hash3_32(ifnull(c1,0))+murmur_hash3_32(ifnull(c2,0))+murmur_hash3_32(ifnull(ab,0)))) as fingerprint from (select t0.c0,t0.c1,t0.c2,t1.c2 as ab from (select * from t1 where c0 in (5)) t1 right outer join[bucket] (select if(murmur_hash3_32(c0)=0,c0,NULL) as c0,c1,c2 from t0) t0 on t0.c0 = t1.c0 left join[bucket] t2 on t2.c0 = t1.c0) as t;
+-- result:
+-- !result
+INSERT INTO r
+(fp)
+select  (sum(murmur_hash3_32(ifnull(c0,0))+murmur_hash3_32(ifnull(c1,0))+murmur_hash3_32(ifnull(c2,0))+murmur_hash3_32(ifnull(ab,0)))) as fingerprint from (select t0.c0,t0.c1,t0.c2,t1.c2 as ab from (select * from t1 where c0 in (5)) t1 right outer join[shuffle] (select if(murmur_hash3_32(c0)=0,c0,NULL) as c0,c1,c2 from t0) t0 on t0.c0 = t1.c0 left join[shuffle] t2 on t2.c0 = t1.c0) as t;
+-- result:
+-- !result
+select assert_true(count(fp)=2), assert_true(count(distinct fp)=1) from r;
+-- result:
+1	1
+-- !result
+truncate table r;
+-- result:
+-- !result
+INSERT INTO r
+(fp)
+select  (sum(murmur_hash3_32(ifnull(c0,0))+murmur_hash3_32(ifnull(c1,0))+murmur_hash3_32(ifnull(c2,0))+murmur_hash3_32(ifnull(ab,0)))) as fingerprint from (select t0.c0,t0.c1,t0.c2,t1.c2 as ab from (select * from t1 where c0 in (6)) t1 right outer join[bucket] (select if(murmur_hash3_32(c0)=0,c0,NULL) as c0,c1,c2 from t0) t0 on t0.c0 = t1.c0 left join[bucket] t2 on t2.c0 = t1.c0) as t;
+-- result:
+-- !result
+INSERT INTO r
+(fp)
+select  (sum(murmur_hash3_32(ifnull(c0,0))+murmur_hash3_32(ifnull(c1,0))+murmur_hash3_32(ifnull(c2,0))+murmur_hash3_32(ifnull(ab,0)))) as fingerprint from (select t0.c0,t0.c1,t0.c2,t1.c2 as ab from (select * from t1 where c0 in (6)) t1 right outer join[shuffle] (select if(murmur_hash3_32(c0)=0,c0,NULL) as c0,c1,c2 from t0) t0 on t0.c0 = t1.c0 left join[shuffle] t2 on t2.c0 = t1.c0) as t;
+-- result:
+-- !result
+select assert_true(count(fp)=2), assert_true(count(distinct fp)=1) from r;
+-- result:
+1	1
+-- !result
+truncate table r;
+-- result:
+-- !result
+INSERT INTO r
+(fp)
+select  (sum(murmur_hash3_32(ifnull(c0,0))+murmur_hash3_32(ifnull(c1,0))+murmur_hash3_32(ifnull(c2,0))+murmur_hash3_32(ifnull(ab,0)))) as fingerprint from (select t0.c0,t0.c1,t0.c2,t1.c2 as ab from (select * from t1 where c0 in (7)) t1 right outer join[bucket] (select if(murmur_hash3_32(c0)=0,c0,NULL) as c0,c1,c2 from t0) t0 on t0.c0 = t1.c0 left join[bucket] t2 on t2.c0 = t1.c0) as t;
+-- result:
+-- !result
+INSERT INTO r
+(fp)
+select  (sum(murmur_hash3_32(ifnull(c0,0))+murmur_hash3_32(ifnull(c1,0))+murmur_hash3_32(ifnull(c2,0))+murmur_hash3_32(ifnull(ab,0)))) as fingerprint from (select t0.c0,t0.c1,t0.c2,t1.c2 as ab from (select * from t1 where c0 in (7)) t1 right outer join[shuffle] (select if(murmur_hash3_32(c0)=0,c0,NULL) as c0,c1,c2 from t0) t0 on t0.c0 = t1.c0 left join[shuffle] t2 on t2.c0 = t1.c0) as t;
+-- result:
+-- !result
+select assert_true(count(fp)=2), assert_true(count(distinct fp)=1) from r;
+-- result:
+1	1
+-- !result
+truncate table r;
+-- result:
+-- !result
+INSERT INTO r
+(fp)
+select  (sum(murmur_hash3_32(ifnull(c0,0))+murmur_hash3_32(ifnull(c1,0))+murmur_hash3_32(ifnull(c2,0))+murmur_hash3_32(ifnull(ab,0)))) as fingerprint from (select t0.c0,t0.c1,t0.c2,t1.c2 as ab from (select * from t1 where c0 in (8)) t1 right outer join[bucket] (select if(murmur_hash3_32(c0)=0,c0,NULL) as c0,c1,c2 from t0) t0 on t0.c0 = t1.c0 left join[bucket] t2 on t2.c0 = t1.c0) as t;
+-- result:
+-- !result
+INSERT INTO r
+(fp)
+select  (sum(murmur_hash3_32(ifnull(c0,0))+murmur_hash3_32(ifnull(c1,0))+murmur_hash3_32(ifnull(c2,0))+murmur_hash3_32(ifnull(ab,0)))) as fingerprint from (select t0.c0,t0.c1,t0.c2,t1.c2 as ab from (select * from t1 where c0 in (8)) t1 right outer join[shuffle] (select if(murmur_hash3_32(c0)=0,c0,NULL) as c0,c1,c2 from t0) t0 on t0.c0 = t1.c0 left join[shuffle] t2 on t2.c0 = t1.c0) as t;
+-- result:
+-- !result
+select assert_true(count(fp)=2), assert_true(count(distinct fp)=1) from r;
+-- result:
+1	1
+-- !result
+truncate table r;
+-- result:
+-- !result
+INSERT INTO r
+(fp)
+select  (sum(murmur_hash3_32(ifnull(c0,0))+murmur_hash3_32(ifnull(c1,0))+murmur_hash3_32(ifnull(c2,0))+murmur_hash3_32(ifnull(ab,0)))) as fingerprint from (select t0.c0,t0.c1,t0.c2,t1.c2 as ab from (select * from t1 where c0 in (9)) t1 right outer join[bucket] (select if(murmur_hash3_32(c0)=0,c0,NULL) as c0,c1,c2 from t0) t0 on t0.c0 = t1.c0 left join[bucket] t2 on t2.c0 = t1.c0) as t;
+-- result:
+-- !result
+INSERT INTO r
+(fp)
+select  (sum(murmur_hash3_32(ifnull(c0,0))+murmur_hash3_32(ifnull(c1,0))+murmur_hash3_32(ifnull(c2,0))+murmur_hash3_32(ifnull(ab,0)))) as fingerprint from (select t0.c0,t0.c1,t0.c2,t1.c2 as ab from (select * from t1 where c0 in (9)) t1 right outer join[shuffle] (select if(murmur_hash3_32(c0)=0,c0,NULL) as c0,c1,c2 from t0) t0 on t0.c0 = t1.c0 left join[shuffle] t2 on t2.c0 = t1.c0) as t;
+-- result:
+-- !result
+select assert_true(count(fp)=2), assert_true(count(distinct fp)=1) from r;
+-- result:
+1	1
+-- !result