Skip to content

Commit 665f86b

Browse files
yujun777yujun777
committed
[fix](ivm) Judge the streams of a pending baseline rebuild before it runs
The pre-step that rebuilds the partitions an IVM baseline barrier names runs before the attempt list does, and it runs the partition refresh directly. PARTITIONS FALLBACK builds [PARTITIONS, COMPLETE], which holds no IVM attempt, so the preflight in buildAttempts never judges its streams: with a partial barrier pending and a stream missing for the partitions it rebuilds, the pre-step started that rebuild, IvmFullRefreshMTMV threw STREAM_UNSUPPORTED out of it, and the task failed without reaching the COMPLETE attempt that reconciles the stream. Key changes: - Judge the streams of the partitions the pre-step rebuilds, with the predicate the partition attempt already uses, before it writes its barrier and before it commits to the rebuild - Route a fallback-authorized request to the COMPLETE attempt, which reconciles the stream and clears the barrier that stays pending; a request that may not fall back fails instead of starting a rebuild that cannot read its streams Unit Test: - MTMVTaskTest.testPendingBaselineRebuildChecksTheStreamsItsPartitionsRead: a non-empty partial barrier plus a stream missing for a table the rebuild reads goes to COMPLETE without writing a barrier, and the same request without fallback fails. Without the change the attempt list stays [PARTITIONS, COMPLETE] and the pre-step runs the rebuild inline. - Regression: mtmv_p0/ivm/test_ivm_partitions_fallback_stream_unusable and test_ivm_chained_stream_scope pass unchanged. Neither covers this path: reaching it needs a partial refresh to fail after its barrier is written, which no debug point injects, so this shape stays unit-covered.
1 parent 0148150 commit 665f86b

2 files changed

Lines changed: 85 additions & 4 deletions

File tree

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

Lines changed: 23 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -644,13 +644,32 @@ private void handlePendingIvmBaselineRebuild(MTMVRefreshContext context,
644644
mtmv.getName(), getTaskId());
645645
} else {
646646
baselinePartitions.sort(String::compareTo);
647+
// This rebuild reads the streams of the partitions it rebuilds, exactly like any other
648+
// partition refresh, so it judges them before it commits to the rebuild. A request that may
649+
// not fall back fails instead of rebuilding less than it asked for; one that may reaches the
650+
// COMPLETE attempt, which is also the only attempt that reconciles the stream this rebuild
651+
// cannot read. Judging it here rather than in buildAttempts matters for a request whose
652+
// attempt list holds no IVM attempt -- PARTITIONS FALLBACK is exactly that -- because the
653+
// pre-step runs before the attempts do.
654+
if (mtmv.isIvm()
655+
&& hasUnusableIvmStreamForPartitions(context, baselinePartitions)) {
656+
if (!request.allowFallback) {
657+
throw new JobException("IVM stream is unusable for the partitions of this refresh, mv="
658+
+ mtmv.getName());
659+
}
660+
ivmFallbackReason = IvmFailureReason.STREAM_UNSUPPORTED.name();
661+
LOG.warn("IVM stream is unusable for the partitions this baseline rebuild plans, mv={}, "
662+
+ "taskId={}. Continuing with COMPLETE refresh.", mtmv.getName(), getTaskId());
663+
attempts.clear();
664+
attempts.add(RefreshAttemptType.COMPLETE);
665+
return;
666+
}
647667
this.needRefreshPartitions = baselinePartitions;
648668
this.refreshMode = generateRefreshMode(baselinePartitions);
649669
writeIvmBaselineBarrier(RefreshMode.PARTITIONS);
650-
// An unusable stream is decided before this point, by the attempt list: it holds no
651-
// partition attempt then, so this rebuild is not reached. Anything else that fails here is
652-
// reported as it is -- leaving the barrier behind would make the IVM attempt that follows
653-
// reject the task with "baseline rebuild is pending" instead of the real reason.
670+
// Anything else that fails here is reported as it is -- leaving the barrier behind would
671+
// make the IVM attempt that follows reject the task with "baseline rebuild is pending"
672+
// instead of the real reason.
654673
executePartitionBasedRefresh(context, RefreshMode.PARTITIONS, ctx);
655674
}
656675
mtmv.releaseIvmBaselineRebuild(mtmvSchemaChangeVersion);

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

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -913,8 +913,70 @@ public void testPartitionsFallbackRebuildsPendingBaselineWithComplete() throws E
913913
Mockito.verify(mtmv, Mockito.never()).releaseIvmBaselineRebuild(Mockito.anyLong());
914914
}
915915

916+
@Test
917+
public void testPendingBaselineRebuildChecksTheStreamsItsPartitionsRead() throws Exception {
918+
// A partial barrier left by an earlier failed refresh. The pre-step rebuilds those partitions
919+
// before the attempts run, and that rebuild reads their streams, so a stream missing for them
920+
// decides the request just as it does for the partition attempt -- and PARTITIONS FALLBACK
921+
// reaches this pre-step without an IVM attempt for buildAttempts to have judged.
922+
Mockito.when(mtmv.isIvm()).thenReturn(true);
923+
Mockito.when(mtmv.getName()).thenReturn("test_mv");
924+
Mockito.when(mtmv.getId()).thenReturn(7L);
925+
Mockito.when(mtmv.getExcludedTriggerTables()).thenReturn(Collections.emptySet());
926+
IvmInfo ivmInfo = new IvmInfo();
927+
ivmInfo.addPendingBaselineRebuildPartitions(Sets.newHashSet(poneName));
928+
Mockito.when(mtmv.getIvmInfo()).thenReturn(ivmInfo);
929+
OlapTable t1 = mockBaseTable("t1");
930+
BaseTableInfo t1Info = Mockito.mock(BaseTableInfo.class);
931+
mtmvUtilStatic.when(() -> MTMVUtil.getTable(t1Info)).thenReturn(t1);
932+
// t1 is not a PCT table, so every partition this rebuild refreshes reads it through its stream,
933+
// and the MV's database holds no stream for it.
934+
Mockito.when(mtmv.getDatabase()).thenReturn(Mockito.mock(Database.class));
935+
MTMVRefreshContext context = Mockito.mock(MTMVRefreshContext.class);
936+
Mockito.when(context.getByPartitionName(Mockito.anyString())).thenReturn(Maps.newHashMap());
916937

938+
MTMVRelation relation = new MTMVRelation(Sets.newHashSet(t1Info), Sets.newHashSet(t1Info),
939+
Sets.newHashSet(t1Info), Sets.newHashSet(), Sets.newHashSet());
940+
MTMVTask task = new MTMVTask(mtmv, relation, MTMVTaskContext.of(
941+
MTMVTaskTriggerMode.MANUAL, null, RefreshMode.PARTITIONS, true, null));
942+
Object request = Deencapsulation.invoke(task, "resolveRefreshRequest");
943+
List<Object> attempts = Lists.newArrayList();
944+
attempts.addAll(Deencapsulation.invoke(task, "buildAttempts", request, false));
945+
Assertions.assertEquals("[PARTITIONS, COMPLETE]", attempts.toString());
946+
947+
try {
948+
Deencapsulation.invoke(task, "handlePendingIvmBaselineRebuild", context, request,
949+
new ConnectContext(), attempts);
950+
} catch (Exception expected) {
951+
// Without the stream check the pre-step rebuilds inline, and how far that rebuild gets
952+
// against these mocks is not what this test is about; the attempts it leaves behind are.
953+
}
954+
955+
// The rebuild that cannot read its streams is skipped rather than attempted: its own barrier
956+
// would have guarded nothing but the data it never wrote, and the COMPLETE attempt left in the
957+
// list reconciles the stream and clears the barrier that is already pending.
958+
Assertions.assertEquals("[COMPLETE]", attempts.toString());
959+
Assertions.assertEquals(IvmFailureReason.STREAM_UNSUPPORTED.name(),
960+
Deencapsulation.getField(task, "ivmFallbackReason"));
961+
Mockito.verify(mtmv, Mockito.never()).persistIvmBaselineGuard(Mockito.any(), Mockito.anySet(),
962+
Mockito.anyLong());
963+
Mockito.verify(mtmv, Mockito.never()).releaseIvmBaselineRebuild(Mockito.anyLong());
964+
965+
// The same rebuild without fallback is not covered by a COMPLETE attempt, so it fails here
966+
// rather than starting a rebuild that cannot read its streams.
967+
MTMVTask strictTask = new MTMVTask(mtmv, relation, MTMVTaskContext.of(
968+
MTMVTaskTriggerMode.MANUAL, null, RefreshMode.PARTITIONS, false, null));
969+
Object strictRequest = Deencapsulation.invoke(strictTask, "resolveRefreshRequest");
970+
List<Object> strictAttempts = Lists.newArrayList();
971+
strictAttempts.addAll(Deencapsulation.invoke(strictTask, "buildAttempts", strictRequest, false));
972+
Assertions.assertEquals("[PARTITIONS]", strictAttempts.toString());
973+
974+
JobException exception = Assertions.assertThrows(JobException.class,
975+
() -> Deencapsulation.invoke(strictTask, "handlePendingIvmBaselineRebuild", context,
976+
strictRequest, new ConnectContext(), strictAttempts));
917977

978+
Assertions.assertTrue(exception.getMessage().contains("IVM stream is unusable"));
979+
}
918980

919981
@Test
920982
public void testCompleteAttemptWritesTheBarrierBeforeReconcilingStreams() throws Exception {

0 commit comments

Comments
 (0)