Skip to content

Commit

Permalink
[BugFix] deduplicate partition range for multi-table join mv (#45773)
Browse files Browse the repository at this point in the history 
Signed-off-by: Murphy <[email protected]>
  • Loading branch information
@murphyatwork
murphyatwork authored May 20, 2024
1 parent ecab48f commit 2bb8e40
Show file tree
Hide file tree
Showing 3 changed files with 78 additions and 3 deletions.
4 changes: 3 additions & 1 deletion fe/fe-core/src/main/java/com/starrocks/connector/ConnectorPartitionTraits.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -320,7 +320,9 @@ public boolean supportPartitionRefresh() {

@Override
public Map<String, Range<PartitionKey>> getPartitionKeyRange(Column partitionColumn, Expr partitionExpr) {
// TODO: check partition type
if (!((OlapTable) table).getPartitionInfo().isRangePartition()) {
throw new IllegalArgumentException("Must be range partitioned table");
}
return ((OlapTable) table).getRangePartitionMap();
}

Expand Down
15 changes: 13 additions & 2 deletions fe/fe-core/src/main/java/com/starrocks/sql/common/SyncPartitionUtils.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,9 @@
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Range;
import com.google.common.collect.RangeSet;
import com.google.common.collect.Sets;
import com.google.common.collect.TreeRangeSet;
import com.starrocks.analysis.DateLiteral;
import com.starrocks.analysis.Expr;
import com.starrocks.analysis.FunctionCallExpr;
Expand Down Expand Up @@ -97,8 +99,17 @@ public static RangePartitionDiff getRangePartitionDiffOfSlotRef(Map<String, Rang
PartitionDiffer differ) {
// This synchronization method has a one-to-one correspondence
// between the base table and the partition of the mv.
return differ != null ? differ.diff(baseRangeMap, mvRangeMap) :
PartitionDiffer.simpleDiff(baseRangeMap, mvRangeMap);
RangeSet<PartitionKey> ranges = TreeRangeSet.create();
Map<String, Range<PartitionKey>> unique = Maps.newHashMap();
for (Map.Entry<String, Range<PartitionKey>> entry : baseRangeMap.entrySet()) {
PartitionRange range = new PartitionRange(entry.getKey(), entry.getValue());
if (!ranges.encloses(entry.getValue())) {
ranges.add(entry.getValue());
unique.put(entry.getKey(), entry.getValue());
}
}
return differ != null ? differ.diff(unique, mvRangeMap) :
PartitionDiffer.simpleDiff(unique, mvRangeMap);
}

public static ListPartitionDiff getListPartitionDiff(Map<String, List<List<String>>> baseListMap,
Expand Down
62 changes: 62 additions & 0 deletions fe/fe-core/src/test/java/com/starrocks/scheduler/PartitionBasedMvRefreshTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
package com.starrocks.scheduler;

import com.google.common.collect.ImmutableList;
import com.google.common.collect.Sets;
import com.starrocks.catalog.Database;
import com.starrocks.catalog.MaterializedView;
import com.starrocks.common.util.UUIDUtil;
Expand Down Expand Up @@ -313,4 +314,65 @@ public void testUnionAllMvWithPartitionWithPartitionStartAndEnd() {
}
});
}

@Test
public void testJoinMV_SlotRef() throws Exception {
starRocksAssert.withTable("CREATE TABLE join_base_t1 (dt1 date, int1 int)\n" +
" PARTITION BY RANGE(dt1)\n" +
" (\n" +
" PARTITION p1 VALUES LESS THAN (\"2020-07-01\"),\n" +
" PARTITION p2 VALUES LESS THAN (\"2020-08-01\"),\n" +
" PARTITION p3 VALUES LESS THAN (\"2020-09-01\")\n" +
" );");
starRocksAssert.withTable("CREATE TABLE join_base_t2 (dt2 date, int2 int)\n" +
" PARTITION BY RANGE(dt2)\n" +
" (\n" +
" PARTITION p4 VALUES LESS THAN (\"2020-07-01\"),\n" +
" PARTITION p5 VALUES LESS THAN (\"2020-08-01\"),\n" +
" PARTITION p6 VALUES LESS THAN (\"2020-09-01\")\n" +
" );");
starRocksAssert.withRefreshedMaterializedView("CREATE MATERIALIZED VIEW join_mv1 " +
"PARTITION BY dt1 " +
"REFRESH MANUAL " +
"PROPERTIES (\"partition_refresh_number\"=\"3\") AS " +
"SELECT dt1,dt2,sum(int1) " +
"FROM join_base_t1 t1 " +
"JOIN join_base_t2 t2 ON t1.dt1=t2.dt2 GROUP BY dt1,dt2;");

MaterializedView mv = starRocksAssert.getMv("test", "join_mv1");
Assert.assertEquals(Sets.newHashSet("p1", "p2", "p3"), mv.getPartitionNames());
starRocksAssert.dropTable("join_base_t1");
starRocksAssert.dropTable("join_base_t2");
starRocksAssert.dropMaterializedView("join_mv1");
}

@Test
public void testJoinMV_ListPartition() throws Exception {
starRocksAssert.withTable("CREATE TABLE join_base_t1 (dt1 date, int1 int)\n" +
"PARTITION BY RANGE(dt1)\n" +
"(\n" +
" PARTITION p202006 VALUES LESS THAN (\"2020-07-01\"),\n" +
" PARTITION p202007 VALUES LESS THAN (\"2020-08-01\"),\n" +
" PARTITION p202008 VALUES LESS THAN (\"2020-09-01\")\n" +
")");
starRocksAssert.withTable("CREATE TABLE join_base_t2 (dt2 date not null, int2 int)\n" +
"PARTITION BY LIST(dt2)\n" +
"(\n" +
" PARTITION p202006 VALUES in (\"2020-06-23\"),\n" +
" PARTITION p202007 VALUES in (\"2020-07-23\"),\n" +
" PARTITION p202008 VALUES in (\"2020-08-23\")\n" +
");");
Exception e = Assert.assertThrows(IllegalArgumentException.class, () ->
starRocksAssert.withRefreshedMaterializedView("CREATE MATERIALIZED VIEW join_mv1 " +
"PARTITION BY dt1 REFRESH MANUAL PROPERTIES (\"partition_refresh_number\"=\"3\") AS \n" +
"SELECT dt1,dt2,sum(int1) " +
"FROM join_base_t1 t1 " +
"JOIN join_base_t2 t2 ON t1.dt1=t2.dt2 GROUP BY dt1,dt2")
);
Assert.assertEquals("Must be range partitioned table", e.getMessage());

starRocksAssert.dropTable("join_base_t1");
starRocksAssert.dropTable("join_base_t2");
starRocksAssert.dropMaterializedView("join_mv1");
}
}

0 comments on commit 2bb8e40

Please sign in to comment.