diff --git a/fe/fe-core/src/main/java/com/starrocks/clone/DynamicPartitionScheduler.java b/fe/fe-core/src/main/java/com/starrocks/clone/DynamicPartitionScheduler.java index 4e63c67a04d21..6694feb78f16a 100644 --- a/fe/fe-core/src/main/java/com/starrocks/clone/DynamicPartitionScheduler.java +++ b/fe/fe-core/src/main/java/com/starrocks/clone/DynamicPartitionScheduler.java @@ -34,6 +34,7 @@ package com.starrocks.clone; +import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.Maps; import com.google.common.collect.Range; import com.google.common.collect.Sets; @@ -535,6 +536,11 @@ private void initDynamicPartitionTable() { initialize = true; } + @VisibleForTesting + public void runOnceForTest() { + runAfterCatalogReady(); + } + @Override protected void runAfterCatalogReady() { if (!initialize) { diff --git a/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rule/RuleType.java b/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rule/RuleType.java index 3ff3fc2c2d429..cb0972271697c 100644 --- a/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rule/RuleType.java +++ b/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rule/RuleType.java @@ -151,6 +151,7 @@ public enum RuleType { TF_MV_ONLY_JOIN_RULE, TF_MV_FILTER_JOIN_RULE, TF_MV_AGGREGATE_SCAN_RULE, + TF_MV_AGGREGATE_AGGREGATE_SCAN_RULE, TF_MV_AGGREGATE_FILTER_SCAN_RULE, TF_MV_AGGREGATE_JOIN_RULE, TF_MV_AGGREGATE_FILTER_JOIN_RULE, diff --git a/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rule/transformation/materialization/rule/AggregateAggregateScanRule.java b/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rule/transformation/materialization/rule/AggregateAggregateScanRule.java index 1cb785b594440..6d24df340a30b 100644 --- a/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rule/transformation/materialization/rule/AggregateAggregateScanRule.java +++ b/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rule/transformation/materialization/rule/AggregateAggregateScanRule.java @@ -34,7 +34,7 @@ public class AggregateAggregateScanRule extends SingleTableRewriteBaseRule { private static AggregateAggregateScanRule INSTANCE = new AggregateAggregateScanRule(); public AggregateAggregateScanRule() { - super(RuleType.TF_MV_AGGREGATE_SCAN_RULE, Pattern.create(OperatorType.LOGICAL_AGGR) + super(RuleType.TF_MV_AGGREGATE_AGGREGATE_SCAN_RULE, Pattern.create(OperatorType.LOGICAL_AGGR) .addChildren(Pattern.create(OperatorType.LOGICAL_AGGR).addChildren(Pattern.create(OperatorType.PATTERN_SCAN)))); } diff --git a/fe/fe-core/src/test/java/com/starrocks/sql/optimizer/rule/transformation/materialization/MvRewriteOptimizationTest.java b/fe/fe-core/src/test/java/com/starrocks/sql/optimizer/rule/transformation/materialization/MvRewriteOptimizationTest.java index 97457fc5f4a32..2ba57086d1cf7 100644 --- a/fe/fe-core/src/test/java/com/starrocks/sql/optimizer/rule/transformation/materialization/MvRewriteOptimizationTest.java +++ b/fe/fe-core/src/test/java/com/starrocks/sql/optimizer/rule/transformation/materialization/MvRewriteOptimizationTest.java @@ -20,7 +20,6 @@ import com.starrocks.catalog.OlapTable; import com.starrocks.catalog.Table; import com.starrocks.common.Config; -import com.starrocks.common.TimeoutException; import com.starrocks.pseudocluster.PseudoCluster; import com.starrocks.qe.ConnectContext; import com.starrocks.scheduler.Task; @@ -44,7 +43,7 @@ public class MvRewriteOptimizationTest { @BeforeClass public static void beforeClass() throws Exception { - Config.dynamic_partition_check_interval_seconds = 1; + Config.dynamic_partition_check_interval_seconds = 10000; Config.bdbje_heartbeat_timeout_second = 60; Config.bdbje_replica_ack_timeout_second = 60; Config.bdbje_lock_timeout_second = 60; @@ -1334,6 +1333,7 @@ public void testUnionRewrite() throws Exception { cluster.runSql("test", "insert into test_base_part values(1000, 1, 2, 3)"); cluster.runSql("test", "insert into test_base_part values(2000, 1, 2, 3)"); cluster.runSql("test", "insert into test_base_part values(2500, 1, 2, 3)"); + createAndRefreshMv("test", "ttl_union_mv_1", "CREATE MATERIALIZED VIEW `ttl_union_mv_1`\n" + "COMMENT \"MATERIALIZED_VIEW\"\n" + "PARTITION BY (`c3`)\n" + @@ -1347,9 +1347,12 @@ public void testUnionRewrite() throws Exception { "AS SELECT `test_base_part`.`c1`, `test_base_part`.`c3`, sum(`test_base_part`.`c2`) AS `c2`\n" + "FROM `test_base_part`\n" + "GROUP BY `test_base_part`.`c1`, `test_base_part`.`c3`;"); + MaterializedView ttlMv1 = getMv("test", "ttl_union_mv_1"); Assert.assertNotNull(ttlMv1); - waitTtl(ttlMv1, 3, 200); + GlobalStateMgr.getCurrentState().getDynamicPartitionScheduler().runOnceForTest(); + Assert.assertEquals(3, ttlMv1.getPartitions().size()); + String query4 = "select c3, sum(c2) from test_base_part group by c3"; String plan4 = getFragmentPlan(query4); PlanTestBase.assertContains(plan4, "ttl_union_mv_1", "UNION", "test_base_part"); @@ -1450,20 +1453,6 @@ public void testNestedMv() throws Exception { starRocksAssert.dropTable("nest_base_table_1"); } - private void waitTtl(MaterializedView mv, int number, int maxRound) throws InterruptedException, TimeoutException { - int round = 0; - while (true) { - if (mv.getPartitions().size() == number) { - break; - } - if (round >= maxRound) { - throw new TimeoutException("wait ttl timeout"); - } - Thread.sleep(1000); - round++; - } - } - @Test public void testPartialPartition() throws Exception { starRocksAssert.getCtx().getSessionVariable().setEnableMaterializedViewUnionRewrite(true); @@ -1670,7 +1659,8 @@ public void testPartialPartition() throws Exception { " )\n" + " AS SELECT k1, sum(v1) as sum_v1 FROM ttl_base_table group by k1;"); MaterializedView ttlMv2 = getMv("test", "ttl_mv_2"); - waitTtl(ttlMv2, 4, 100); + GlobalStateMgr.getCurrentState().getDynamicPartitionScheduler().runOnceForTest(); + Assert.assertEquals(4, ttlMv2.getPartitions().size()); String query16 = "select k1, sum(v1) FROM ttl_base_table where k1=3 group by k1"; String plan16 = getFragmentPlan(query16);