Skip to content

Commit 0148150

Browse files
yujun777yujun777
committed
[fix](ivm) Judge the streams of a partition refresh by the partitions it refreshes
A partition refresh reads the stream of the PCT tables its partitions map to (they are read in RESET mode) and of the non-PCT tables the plan scans, while a PCT table that no refreshed partition's mapping names keeps its place in the plan as an ordinary scan. The preflight that decides whether a refresh can run judged every table of the plan instead, so a partition refresh that would have worked was sent to a complete rebuild whenever an unread table's stream was missing. Key changes: - Judge only the incremental attempt in buildAttempts, where the plan's tables are exactly what its rewrite scans - Let the partition attempt check its own scope once it has planned: the non-PCT tables of the plan plus the PCT tables in the refreshed partitions' mappings, which is the set the reset read mode is built from - Fall back through the existing path instead of failing, and before writing the baseline barrier, so an attempt that cannot succeed leaves nothing behind; a request that may not fall back still fails rather than refreshing less than it asked for Unit Test: - MTMVTaskTest.testPartitionRefreshChecksOnlyTheStreamsItsPartitionsRead: a PCT table that no refreshed partition's mapping names is not judged, and naming it makes its missing stream decide (the assertion fails when the whole plan is judged) - Regression: mtmv_p0/ivm/test_ivm_chained_stream_scope drops its stream with FORCE, since DROP STREAM without it is rejected in cloud mode and every suite runs there; mtmv_p0/ivm/test_ivm_partitions_fallback_stream_unusable pins the fallback end to end through the new path
1 parent 77770d7 commit 0148150

3 files changed

Lines changed: 167 additions & 17 deletions

File tree

‎fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java‎

Lines changed: 105 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -445,17 +445,6 @@ private List<RefreshAttemptType> buildAttempts(RefreshRequest request, boolean c
445445
if (shouldUseCompleteForInitialIvmRefresh(containsOneRowRelation)) {
446446
return Lists.newArrayList(RefreshAttemptType.COMPLETE);
447447
}
448-
// A base table the plan scans without a usable stream makes every other attempt fail: the
449-
// incremental rewrite reads the stream of every table it scans, and a partition refresh reads
450-
// those of the PCT tables it realigns as well. Only the COMPLETE attempt reconciles streams, so
451-
// it is the one that has to run. Deciding it here keeps the attempt that cannot succeed -- and
452-
// the baseline barrier a partition refresh writes before it starts -- out of the way altogether.
453-
if (request.allowFallback && mtmv.isIvm() && hasUnusableIvmStream()) {
454-
ivmFallbackReason = IvmFailureReason.STREAM_UNSUPPORTED.name();
455-
LOG.warn("IVM stream is unusable, mv={}, taskId={}. Continuing with COMPLETE refresh.",
456-
mtmv.getName(), getTaskId());
457-
return Lists.newArrayList(RefreshAttemptType.COMPLETE);
458-
}
459448
List<RefreshAttemptType> attempts = Lists.newArrayList();
460449
switch (request.refreshMode) {
461450
case AUTO:
@@ -500,6 +489,24 @@ private List<RefreshAttemptType> buildAttempts(RefreshRequest request, boolean c
500489
default:
501490
throw new IllegalStateException("Unsupported refresh mode: " + request.refreshMode);
502491
}
492+
// A base table the plan scans without a usable stream makes the incremental attempt fail: its
493+
// rewrite reads the stream of every table it scans. Only COMPLETE reconciles streams, so a request
494+
// that would try the incremental path first goes there directly, and the attempt that cannot
495+
// succeed -- with the baseline barrier it writes before it starts -- stays out of the way.
496+
//
497+
// Only that attempt is judged here. A partition refresh reads a narrower set, and how narrow
498+
// depends on the partitions it plans: a PCT table that no refreshed partition's mapping names
499+
// keeps its place in the plan as an ordinary scan, and an excluded trigger table is never read
500+
// through a stream. It checks its own scope once it has planned (see
501+
// hasUnusableIvmStreamForPartitions), so judging it here by the whole plan would send a partition
502+
// refresh that would have worked to COMPLETE.
503+
if (request.allowFallback && mtmv.isIvm() && attempts.contains(RefreshAttemptType.IVM)
504+
&& hasUnusableIvmStream()) {
505+
ivmFallbackReason = IvmFailureReason.STREAM_UNSUPPORTED.name();
506+
LOG.warn("IVM stream is unusable, mv={}, taskId={}. Continuing with COMPLETE refresh.",
507+
mtmv.getName(), getTaskId());
508+
return Lists.newArrayList(RefreshAttemptType.COMPLETE);
509+
}
503510
return attempts;
504511
}
505512

@@ -810,6 +817,22 @@ private boolean executePartitionBasedRefresh(MTMVRefreshContext refreshContext,
810817
}
811818
this.needRefreshPartitions = partitionPlan.partitions;
812819
this.refreshMode = generateRefreshMode(needRefreshPartitions);
820+
// This attempt now knows which partitions it refreshes, and with them which streams it reads.
821+
// Judged here rather than in buildAttempts because only the plan knows that scope, and judged
822+
// before the NOT_REFRESH return below so that a fallback-capable request still reaches the only
823+
// attempt that reconciles streams. Falling back here continues to COMPLETE, which is what repairs
824+
// them; a request that may not fall back fails instead of quietly refreshing less than it asked.
825+
if (mtmv.isIvm()
826+
&& hasUnusableIvmStreamForPartitions(partitionPlan.context, needRefreshPartitions)) {
827+
if (!request.allowFallback) {
828+
throw new JobException("IVM stream is unusable for the partitions of this refresh, mv="
829+
+ mtmv.getName());
830+
}
831+
ivmFallbackReason = IvmFailureReason.STREAM_UNSUPPORTED.name();
832+
LOG.warn("IVM stream is unusable for the partitions this refresh plans, mv={}, taskId={}. "
833+
+ "Continuing with COMPLETE refresh.", mtmv.getName(), getTaskId());
834+
return false;
835+
}
813836
if (refreshMode == MTMVTaskRefreshMode.NOT_REFRESH) {
814837
return true;
815838
}
@@ -909,24 +932,90 @@ private boolean hasUnusableIvmStream() {
909932
// rewriter and the full refresh take the streams of the plan's scans -- so judging them would
910933
// rebuild an MV whose refresh had nothing wrong with it.
911934
for (BaseTableInfo baseTableInfo : relation.getBaseTablesOneLevelAndFromView()) {
912-
OlapTable baseTable;
913-
try {
914-
baseTable = (OlapTable) MTMVUtil.getTable(baseTableInfo);
915-
} catch (Exception e) {
916-
LOG.warn("Cannot resolve base table {} of mv={}", baseTableInfo, mtmv.getName(), e);
935+
OlapTable baseTable = resolveIvmBaseTable(baseTableInfo);
936+
if (baseTable == null) {
937+
continue;
938+
}
939+
if (MTMVPartitionUtil.isTableExcluded(excluded,
940+
new TableNameInfo(baseTable.getFullQualifiers()))) {
941+
continue;
942+
}
943+
if (usableIvmStream(mvDb, baseTable) == null) {
944+
return true;
945+
}
946+
}
947+
return false;
948+
}
949+
950+
/**
951+
* Whether a stream this partition refresh will read is missing or unusable.
952+
*
953+
* <p>The set is narrower than the plan, and which tables are in it depends on the partitions being
954+
* refreshed: a PCT table that no refreshed partition's mapping names keeps its place in the plan as
955+
* an ordinary scan, so its stream is never read, and a table outside the plan's own tables is not
956+
* scanned at all. Judging the whole plan instead sends a partition refresh that would have worked to
957+
* a full rebuild whenever such an unread stream is missing.
958+
*/
959+
private boolean hasUnusableIvmStreamForPartitions(MTMVRefreshContext context,
960+
List<String> mvPartitionNames) {
961+
Database mvDb = (Database) mtmv.getDatabase();
962+
if (mvDb == null) {
963+
// Nothing to look the streams up in, so there is nothing to decide here.
964+
return false;
965+
}
966+
Set<TableNameInfo> excluded = mtmv.getExcludedTriggerTables();
967+
Set<OlapTable> streamedTables = Sets.newLinkedHashSet();
968+
Set<BaseTableInfo> pctTableInfos = Sets.newHashSet();
969+
for (BaseColInfo pctInfo : mtmv.getMvPartitionInfo().getPctInfos()) {
970+
pctTableInfos.add(pctInfo.getTableInfo());
971+
}
972+
// Every table of the plan that is not a PCT table is read through its stream: the partition
973+
// refresh gives those tables a read mode (IvmRewriteContext.full).
974+
for (BaseTableInfo baseTableInfo : relation.getBaseTablesOneLevelAndFromView()) {
975+
if (pctTableInfos.contains(baseTableInfo)) {
917976
continue;
918977
}
978+
OlapTable baseTable = resolveIvmBaseTable(baseTableInfo);
979+
if (baseTable != null) {
980+
streamedTables.add(baseTable);
981+
}
982+
}
983+
// A PCT table is read through its stream only while the mapping of a refreshed partition names
984+
// it, which is exactly the mapping the reset read mode is built from.
985+
for (String mvPartitionName : mvPartitionNames) {
986+
for (MTMVRelatedTableIf relatedTable : context.getByPartitionName(mvPartitionName).keySet()) {
987+
if (relatedTable instanceof OlapTable) {
988+
streamedTables.add((OlapTable) relatedTable);
989+
}
990+
}
991+
}
992+
for (OlapTable baseTable : streamedTables) {
919993
if (MTMVPartitionUtil.isTableExcluded(excluded,
920994
new TableNameInfo(baseTable.getFullQualifiers()))) {
921995
continue;
922996
}
923997
if (usableIvmStream(mvDb, baseTable) == null) {
998+
LOG.warn("IVM stream is unusable for the partitions this refresh plans, mv={}, baseTable={}",
999+
mtmv.getName(), baseTable.getName());
9241000
return true;
9251001
}
9261002
}
9271003
return false;
9281004
}
9291005

1006+
/**
1007+
* A base table of the plan, or null when it cannot be resolved -- which says nothing about its stream,
1008+
* so a caller that is only judging streams skips it.
1009+
*/
1010+
private OlapTable resolveIvmBaseTable(BaseTableInfo baseTableInfo) {
1011+
try {
1012+
return (OlapTable) MTMVUtil.getTable(baseTableInfo);
1013+
} catch (Exception e) {
1014+
LOG.warn("Cannot resolve base table {} of mv={}", baseTableInfo, mtmv.getName(), e);
1015+
return null;
1016+
}
1017+
}
1018+
9301019
/** The MV's stream for the base table, or null when it is missing or cannot be used. */
9311020
private BaseTableStream usableIvmStream(Database mvDb, OlapTable baseTable) {
9321021
TableIf stream = mvDb.getTableNullable(

‎fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTaskTest.java‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -72,6 +72,7 @@
7272

7373
import com.google.common.collect.ImmutableMap;
7474
import com.google.common.collect.Lists;
75+
import com.google.common.collect.Maps;
7576
import com.google.common.collect.Sets;
7677
import org.apache.commons.collections4.CollectionUtils;
7778
import org.junit.jupiter.api.AfterEach;
@@ -323,6 +324,66 @@ public void testBuildAttemptsIgnoresAStreamOnlyTheClosureCarries() throws Except
323324
Assertions.assertNull(Deencapsulation.getField(task, "ivmFallbackReason"));
324325
}
325326

327+
@Test
328+
public void testPartitionRefreshChecksOnlyTheStreamsItsPartitionsRead() throws Exception {
329+
// t1 UNION ALL t2, both PCT tables of the MV, each backing one of its partitions. Refreshing the
330+
// t1-backed partition reads t1's stream and leaves t2 to an ordinary scan, so t2's stream is not
331+
// part of this refresh and its absence must not send it to COMPLETE.
332+
Mockito.when(mtmv.isIvm()).thenReturn(true);
333+
Mockito.when(mtmv.getName()).thenReturn("test_mv");
334+
Mockito.when(mtmv.getId()).thenReturn(7L);
335+
Mockito.when(mtmv.getExcludedTriggerTables()).thenReturn(Collections.emptySet());
336+
OlapTable t1 = mockBaseTable("t1");
337+
OlapTable t2 = mockBaseTable("t2");
338+
BaseTableInfo t1Info = Mockito.mock(BaseTableInfo.class);
339+
BaseTableInfo t2Info = Mockito.mock(BaseTableInfo.class);
340+
mtmvUtilStatic.when(() -> MTMVUtil.getTable(t1Info)).thenReturn(t1);
341+
mtmvUtilStatic.when(() -> MTMVUtil.getTable(t2Info)).thenReturn(t2);
342+
OlapTableStream t1Stream = usableStreamFor(t1);
343+
Database mvDb = Mockito.mock(Database.class);
344+
Mockito.when(mvDb.getTableNullable(IvmUtil.streamName(7L, t1.getFullQualifiers())))
345+
.thenReturn(t1Stream);
346+
Mockito.when(mtmv.getDatabase()).thenReturn(mvDb);
347+
Mockito.when(mtmvPartitionInfo.getPctInfos()).thenReturn(Lists.newArrayList(
348+
new BaseColInfo("dt", t1Info), new BaseColInfo("dt", t2Info)));
349+
350+
// The partition this refresh plans reads t1, and t2 keeps its place in the plan as a plain scan.
351+
Map<MTMVRelatedTableIf, Set<String>> mapping = Maps.newHashMap();
352+
mapping.put(t1, Sets.newHashSet("p1"));
353+
MTMVRefreshContext context = Mockito.mock(MTMVRefreshContext.class);
354+
Mockito.when(context.getByPartitionName(Mockito.anyString())).thenReturn(mapping);
355+
356+
MTMVRelation relation = new MTMVRelation(Sets.newHashSet(t1Info, t2Info), Sets.newHashSet(t1Info, t2Info),
357+
Sets.newHashSet(t1Info, t2Info), Sets.newHashSet(), Sets.newHashSet());
358+
MTMVTask task = new MTMVTask(mtmv, relation, new MTMVTaskContext(MTMVTaskTriggerMode.MANUAL));
359+
360+
Assertions.assertFalse((Boolean) Deencapsulation.invoke(task, "hasUnusableIvmStreamForPartitions",
361+
context, Lists.newArrayList("p_t1")));
362+
363+
// Once a refreshed partition's mapping names t2, its stream is read, and its absence decides.
364+
mapping.put(t2, Sets.newHashSet("p2"));
365+
Assertions.assertTrue((Boolean) Deencapsulation.invoke(task, "hasUnusableIvmStreamForPartitions",
366+
context, Lists.newArrayList("p_t1")));
367+
}
368+
369+
private OlapTable mockBaseTable(String name) {
370+
OlapTable baseTable = Mockito.mock(OlapTable.class);
371+
Mockito.when(baseTable.getName()).thenReturn(name);
372+
Mockito.when(baseTable.getFullQualifiers()).thenReturn(Lists.newArrayList("internal", "db", name));
373+
return baseTable;
374+
}
375+
376+
/** A stream that {@code IvmUtil.isIvmStreamUsable} accepts for the given base table. */
377+
private OlapTableStream usableStreamFor(OlapTable baseTable) {
378+
List<String> qualifiers = baseTable.getFullQualifiers();
379+
OlapTableStream stream = Mockito.mock(OlapTableStream.class);
380+
Mockito.when(stream.getBaseTableFullQualifiers()).thenReturn(qualifiers);
381+
Mockito.when(stream.isDisabled()).thenReturn(false);
382+
Mockito.when(stream.isStale()).thenReturn(false);
383+
Mockito.when(stream.getBaseTableNullable()).thenReturn(baseTable);
384+
return stream;
385+
}
386+
326387
private MTMVRelation relationWithOneBaseTable() {
327388
BaseTableInfo baseTable = Mockito.mock(BaseTableInfo.class);
328389
// A table of the query is in the plan, in the first level of the query, and in the closure.

‎regression-test/suites/mtmv_p0/ivm/test_ivm_chained_stream_scope.groovy‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -141,7 +141,7 @@ suite("test_ivm_chained_stream_scope") {
141141
// base table, and the MV was created with a stream for it.
142142
def unusedStreams = streamNamesOf(downstreamMv, baseTable)
143143
assertEquals(1, unusedStreams.size())
144-
sql """DROP STREAM ${context.dbName}.${unusedStreams[0]}"""
144+
sql """DROP STREAM ${context.dbName}.${unusedStreams[0]} FORCE"""
145145
assertEquals(0, streamNamesOf(downstreamMv, baseTable).size())
146146

147147
// One change reaches the downstream MV through the upstream MV, so the refresh below has something

0 commit comments

Comments
 (0)