This is an automated email from the ASF dual-hosted git repository.
yujun777 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 01efbca40a5 [fix](ivm) Fall back to complete refresh when the IVM
stream is unusable (#68170)
01efbca40a5 is described below
commit 01efbca40a548fc266c5c387c38645e39dc0061d
Author: yujun <[email protected]>
AuthorDate: Tue Sep 22 17:18:47 2026 +0800
[fix](ivm) Fall back to complete refresh when the IVM stream is unusable
(#68170)
When an IVM incremental refresh hits a missing or
unusable base table stream, the rewrite fails with `STREAM_UNSUPPORTED`.
Previously this reason fell back to the partition-based refresh, which
reads the same broken stream again and fails the task, so the refresh
never reached the COMPLETE attempt whose reconcile step recreates the
stream. Only an explicit `REFRESH COMPLETE` could recover.
Two changes:
1. Add `STREAM_UNSUPPORTED` to
`IvmFailureReason.requiresCompleteRefresh()`, so an unusable stream
jumps directly to the COMPLETE attempt that reconciles it.
2. Move `reconcileIvmStreams` from the shared partition-based path into
`executeCompleteAttempt`. Behavior is unchanged (COMPLETE was its only
triggering mode), but it makes the complete attempt the single place
that resets stream baselines, which a stream's global baseline requires.
IVM incremental refresh now recovers automatically from an unusable base
table stream by falling back to a complete refresh instead of failing
the task.
---
.../apache/doris/job/extensions/mtmv/MTMVTask.java | 244 ++++++++++++---
.../apache/doris/mtmv/ivm/IvmFailureReason.java | 1 +
.../java/org/apache/doris/mtmv/MTMVTaskTest.java | 330 ++++++++++++++++-----
.../doris/mtmv/ivm/IvmFailureReasonTest.java | 3 +
.../mtmv_p0/ivm/test_ivm_chained_stream_scope.out | 6 +
...est_ivm_partitions_fallback_stream_unusable.out | 13 +
.../ivm/test_ivm_chained_stream_scope.groovy | 162 ++++++++++
..._ivm_partitions_fallback_stream_unusable.groovy | 187 ++++++++++++
8 files changed, 822 insertions(+), 124 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
index 8b24e11fe31..f0445240602 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
@@ -492,6 +492,24 @@ public class MTMVTask extends AbstractTask {
default:
throw new IllegalStateException("Unsupported refresh mode: " +
request.refreshMode);
}
+ // A base table the plan scans without a usable stream makes the
incremental attempt fail: its
+ // rewrite reads the stream of every table it scans. Only COMPLETE
reconciles streams, so a request
+ // that would try the incremental path first goes there directly, and
the attempt that cannot
+ // succeed -- with the baseline barrier it writes before it starts --
stays out of the way.
+ //
+ // Only that attempt is judged here. A partition refresh reads a
narrower set, and how narrow
+ // depends on the partitions it plans: a PCT table that no refreshed
partition's mapping names
+ // keeps its place in the plan as an ordinary scan, and an excluded
trigger table is never read
+ // through a stream. It checks its own scope once it has planned (see
+ // hasUnusableIvmStreamForPartitions), so judging it here by the whole
plan would send a partition
+ // refresh that would have worked to COMPLETE.
+ if (request.allowFallback && mtmv.isIvm() &&
attempts.contains(RefreshAttemptType.IVM)
+ && hasUnusableIvmStream()) {
+ ivmFallbackReason = IvmFailureReason.STREAM_UNSUPPORTED.name();
+ LOG.warn("IVM stream is unusable, mv={}, taskId={}. Continuing
with COMPLETE refresh.",
+ mtmv.getName(), getTaskId());
+ return Lists.newArrayList(RefreshAttemptType.COMPLETE);
+ }
return attempts;
}
@@ -549,6 +567,21 @@ public class MTMVTask extends AbstractTask {
}
}
+ /**
+ * Makes the barrier that says "these MV partitions must be rebuilt before
their IVM offsets may be
+ * used again" durable. Every caller writes it as soon as it has decided
the partition set and
+ * before anything that touches MV data or base table streams, so that a
crash or a rejection can
+ * only ever leave a barrier with no rebuild behind it, which merely costs
one extra rebuild, and
+ * never a rebuild with no barrier, which silently loses rows.
+ */
+ private void writeIvmBaselineBarrier(RefreshMode refreshMode) throws
JobException {
+ if (mtmv.isIvm()) {
+ // Persist the guard before the first baseline data transaction.
+ mtmv.persistIvmBaselineGuard(refreshMode,
Sets.newHashSet(needRefreshPartitions),
+ mtmvSchemaChangeVersion);
+ }
+ }
+
private void executeCompleteAttempt(MTMVRefreshContext context,
ConnectContext ctx)
throws JobException, AnalysisException {
this.needRefreshPartitions =
Lists.newArrayList(mtmv.getPartitionNames());
@@ -556,6 +589,16 @@ public class MTMVTask extends AbstractTask {
if (refreshMode == MTMVTaskRefreshMode.NOT_REFRESH) {
return;
}
+ // The barrier goes first: a stream this rebuild reconciles carries
the base table's current
+ // rows as its initial snapshot, and a later incremental refresh that
consumed it as a delta
+ // against data still built from the old baseline would double-count
them.
+ writeIvmBaselineBarrier(RefreshMode.COMPLETE);
+ // A complete rebuild resets the stream baselines, so reconcile
missing or unusable streams
+ // before reading anything. Only COMPLETE may do this: a stream
baseline is global, resetting
+ // it during a partial refresh would corrupt the partitions that
refresh does not touch.
+ if (mtmv.isIvm()) {
+ reconcileIvmStreams(ctx);
+ }
executePartitionBasedRefresh(context, RefreshMode.COMPLETE, ctx);
}
@@ -604,8 +647,32 @@ public class MTMVTask extends AbstractTask {
mtmv.getName(), getTaskId());
} else {
baselinePartitions.sort(String::compareTo);
+ // This rebuild reads the streams of the partitions it rebuilds,
exactly like any other
+ // partition refresh, so it judges them before it commits to the
rebuild. A request that may
+ // not fall back fails instead of rebuilding less than it asked
for; one that may reaches the
+ // COMPLETE attempt, which is also the only attempt that
reconciles the stream this rebuild
+ // cannot read. Judging it here rather than in buildAttempts
matters for a request whose
+ // attempt list holds no IVM attempt -- PARTITIONS FALLBACK is
exactly that -- because the
+ // pre-step runs before the attempts do.
+ if (mtmv.isIvm()
+ && hasUnusableIvmStreamForPartitions(context,
baselinePartitions)) {
+ if (!request.allowFallback) {
+ throw new JobException("IVM stream is unusable for the
partitions of this refresh, mv="
+ + mtmv.getName());
+ }
+ ivmFallbackReason = IvmFailureReason.STREAM_UNSUPPORTED.name();
+ LOG.warn("IVM stream is unusable for the partitions this
baseline rebuild plans, mv={}, "
+ + "taskId={}. Continuing with COMPLETE refresh.",
mtmv.getName(), getTaskId());
+ attempts.clear();
+ attempts.add(RefreshAttemptType.COMPLETE);
+ return;
+ }
this.needRefreshPartitions = baselinePartitions;
this.refreshMode = generateRefreshMode(baselinePartitions);
+ writeIvmBaselineBarrier(RefreshMode.PARTITIONS);
+ // Anything else that fails here is reported as it is -- leaving
the barrier behind would
+ // make the IVM attempt that follows reject the task with
"baseline rebuild is pending"
+ // instead of the real reason.
executePartitionBasedRefresh(context, RefreshMode.PARTITIONS, ctx);
}
mtmv.releaseIvmBaselineRebuild(mtmvSchemaChangeVersion);
@@ -776,9 +843,26 @@ public class MTMVTask extends AbstractTask {
}
this.needRefreshPartitions = partitionPlan.partitions;
this.refreshMode = generateRefreshMode(needRefreshPartitions);
+ // This attempt now knows which partitions it refreshes, and with them
which streams it reads.
+ // Judged here rather than in buildAttempts because only the plan
knows that scope, and judged
+ // before the NOT_REFRESH return below so that a fallback-capable
request still reaches the only
+ // attempt that reconciles streams. Falling back here continues to
COMPLETE, which is what repairs
+ // them; a request that may not fall back fails instead of quietly
refreshing less than it asked.
+ if (mtmv.isIvm()
+ && hasUnusableIvmStreamForPartitions(partitionPlan.context,
needRefreshPartitions)) {
+ if (!request.allowFallback) {
+ throw new JobException("IVM stream is unusable for the
partitions of this refresh, mv="
+ + mtmv.getName());
+ }
+ ivmFallbackReason = IvmFailureReason.STREAM_UNSUPPORTED.name();
+ LOG.warn("IVM stream is unusable for the partitions this refresh
plans, mv={}, taskId={}. "
+ + "Continuing with COMPLETE refresh.", mtmv.getName(),
getTaskId());
+ return false;
+ }
if (refreshMode == MTMVTaskRefreshMode.NOT_REFRESH) {
return true;
}
+ writeIvmBaselineBarrier(RefreshMode.PARTITIONS);
executePartitionBasedRefresh(partitionPlan.context,
RefreshMode.PARTITIONS, ctx);
return true;
}
@@ -788,14 +872,6 @@ public class MTMVTask extends AbstractTask {
throws JobException, AnalysisException {
boolean useIvmFallbackStreams = mtmv.isIvm();
Map<TableIf, String> tableWithPartKey = getIncrementalTableMap();
- if (useIvmFallbackStreams) {
- // Persist the guard before the first baseline data transaction.
- mtmv.persistIvmBaselineGuard(refreshMode,
Sets.newHashSet(needRefreshPartitions),
- mtmvSchemaChangeVersion);
- if (refreshMode == RefreshMode.COMPLETE) {
- reconcileIvmStreams(ctx);
- }
- }
this.completedPartitions = Lists.newCopyOnWriteArrayList();
try {
// Snapshot persistence happens after refresh partitions are split
into execution groups. Load the
@@ -862,6 +938,121 @@ public class MTMVTask extends AbstractTask {
mtmv.getDatabase().getFullName(), mtmv.getName(), getTaskId());
}
+ /**
+ * Whether a base table the refresh reads has no stream that can be read,
which no attempt other
+ * than COMPLETE can work around.
+ *
+ * <p>A base table that cannot be resolved is skipped rather than judged:
it says nothing about the
+ * streams, and the refresh fails on it for its own reasons -- the attempt
that runs reports that,
+ * this one only decides which attempt that should be.
+ */
+ private boolean hasUnusableIvmStream() {
+ Database mvDb = (Database) mtmv.getDatabase();
+ if (mvDb == null) {
+ // Nothing to look the streams up in, so there is nothing to
decide here.
+ return false;
+ }
+ Set<TableNameInfo> excluded = mtmv.getExcludedTriggerTables();
+ // The tables in the plan, not the relation's closure: a chained MV is
created with a stream for
+ // every base table behind the MVs it reads, but no rewrite ever looks
those up -- the incremental
+ // rewriter and the full refresh take the streams of the plan's scans
-- so judging them would
+ // rebuild an MV whose refresh had nothing wrong with it.
+ for (BaseTableInfo baseTableInfo :
relation.getBaseTablesOneLevelAndFromView()) {
+ OlapTable baseTable = resolveIvmBaseTable(baseTableInfo);
+ if (baseTable == null) {
+ continue;
+ }
+ if (MTMVPartitionUtil.isTableExcluded(excluded,
+ new TableNameInfo(baseTable.getFullQualifiers()))) {
+ continue;
+ }
+ if (usableIvmStream(mvDb, baseTable) == null) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Whether a stream this partition refresh will read is missing or
unusable.
+ *
+ * <p>The set is narrower than the plan, and which tables are in it
depends on the partitions being
+ * refreshed: a PCT table that no refreshed partition's mapping names
keeps its place in the plan as
+ * an ordinary scan, so its stream is never read, and a table outside the
plan's own tables is not
+ * scanned at all. Judging the whole plan instead sends a partition
refresh that would have worked to
+ * a full rebuild whenever such an unread stream is missing.
+ */
+ private boolean hasUnusableIvmStreamForPartitions(MTMVRefreshContext
context,
+ List<String> mvPartitionNames) {
+ Database mvDb = (Database) mtmv.getDatabase();
+ if (mvDb == null) {
+ // Nothing to look the streams up in, so there is nothing to
decide here.
+ return false;
+ }
+ Set<TableNameInfo> excluded = mtmv.getExcludedTriggerTables();
+ Set<OlapTable> streamedTables = Sets.newLinkedHashSet();
+ Set<BaseTableInfo> pctTableInfos = Sets.newHashSet();
+ for (BaseColInfo pctInfo : mtmv.getMvPartitionInfo().getPctInfos()) {
+ pctTableInfos.add(pctInfo.getTableInfo());
+ }
+ // Every table of the plan that is not a PCT table is read through its
stream: the partition
+ // refresh gives those tables a read mode (IvmRewriteContext.full).
+ for (BaseTableInfo baseTableInfo :
relation.getBaseTablesOneLevelAndFromView()) {
+ if (pctTableInfos.contains(baseTableInfo)) {
+ continue;
+ }
+ OlapTable baseTable = resolveIvmBaseTable(baseTableInfo);
+ if (baseTable != null) {
+ streamedTables.add(baseTable);
+ }
+ }
+ // A PCT table is read through its stream only while the mapping of a
refreshed partition names
+ // it, which is exactly the mapping the reset read mode is built from.
+ for (String mvPartitionName : mvPartitionNames) {
+ for (MTMVRelatedTableIf relatedTable :
context.getByPartitionName(mvPartitionName).keySet()) {
+ if (relatedTable instanceof OlapTable) {
+ streamedTables.add((OlapTable) relatedTable);
+ }
+ }
+ }
+ for (OlapTable baseTable : streamedTables) {
+ if (MTMVPartitionUtil.isTableExcluded(excluded,
+ new TableNameInfo(baseTable.getFullQualifiers()))) {
+ continue;
+ }
+ if (usableIvmStream(mvDb, baseTable) == null) {
+ LOG.warn("IVM stream is unusable for the partitions this
refresh plans, mv={}, baseTable={}",
+ mtmv.getName(), baseTable.getName());
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * A base table of the plan, or null when it cannot be resolved -- which
says nothing about its stream,
+ * so a caller that is only judging streams skips it.
+ */
+ private OlapTable resolveIvmBaseTable(BaseTableInfo baseTableInfo) {
+ try {
+ return (OlapTable) MTMVUtil.getTable(baseTableInfo);
+ } catch (Exception e) {
+ LOG.warn("Cannot resolve base table {} of mv={}", baseTableInfo,
mtmv.getName(), e);
+ return null;
+ }
+ }
+
+ /** The MV's stream for the base table, or null when it is missing or
cannot be used. */
+ private BaseTableStream usableIvmStream(Database mvDb, OlapTable
baseTable) {
+ TableIf stream = mvDb.getTableNullable(
+ IvmUtil.streamName(mtmv.getId(),
baseTable.getFullQualifiers()));
+ if (stream instanceof BaseTableStream
+ && IvmUtil.isIvmStreamUsable((BaseTableStream) stream,
baseTable)) {
+ return (BaseTableStream) stream;
+ }
+ return null;
+ }
+
private void reconcileIvmStreams(ConnectContext ctx) throws JobException {
try {
Database mvDb = (Database) mtmv.getDatabase();
@@ -872,10 +1063,7 @@ public class MTMVTask extends AbstractTask {
new TableNameInfo(baseTable.getFullQualifiers()))) {
continue;
}
- String streamName = IvmUtil.streamName(mtmv.getId(),
baseTable.getFullQualifiers());
- TableIf stream = mvDb.getTableNullable(streamName);
- if (stream instanceof BaseTableStream
- && IvmUtil.isIvmStreamUsable((BaseTableStream) stream,
baseTable)) {
+ if (usableIvmStream(mvDb, baseTable) != null) {
continue;
}
CreateMTMVCommand.createTableStream(ctx, mvDb, mtmv,
baseTable);
@@ -1336,38 +1524,6 @@ public class MTMVTask extends AbstractTask {
}
}
- public List<String> calculateNeedRefreshPartitions(MTMVRefreshContext
context)
- throws AnalysisException, JobException {
- RefreshRequest request = resolveRefreshRequest();
- if (request.refreshMode == RefreshMode.COMPLETE) {
- return Lists.newArrayList(mtmv.getPartitionNames());
- }
- // check whether the user manually triggers it
- if (taskContext.getTriggerMode() == MTMVTaskTriggerMode.MANUAL) {
- if (!CollectionUtils.isEmpty(taskContext.getPartitions())) {
- return taskContext.getPartitions();
- }
- }
- // if refreshMethod is COMPLETE, we must FULL refresh, avoid external
table MTMV always not refresh
- if (mtmv.getRefreshInfo().getRefreshMethod() ==
RefreshMethod.COMPLETE) {
- return Lists.newArrayList(mtmv.getPartitionNames());
- }
- // We need to use a newly generated relationship and cannot retrieve
it using mtmv.getRelation()
- // to avoid rebuilding the baseTable and causing a change in the
tableId
- boolean fresh = MTMVPartitionUtil.isMTMVSync(context,
relation.getBaseTablesOneLevelAndFromView(),
- mtmv.getExcludedTriggerTables());
- if (fresh) {
- return Lists.newArrayList();
- }
- // current, if partitionType is SELF_MANAGE, we can only FULL refresh
- if (mtmv.getMvPartitionInfo().getPartitionType() ==
MTMVPartitionType.SELF_MANAGE) {
- return Lists.newArrayList(mtmv.getPartitionNames());
- }
- // We need to use a newly generated relationship and cannot retrieve
it using mtmv.getRelation()
- // to avoid rebuilding the baseTable and causing a change in the
tableId
- return MTMVPartitionUtil.getMTMVNeedRefreshPartitions(context,
relation.getBaseTablesOneLevelAndFromView());
- }
-
public MTMVTaskContext getTaskContext() {
return taskContext;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmFailureReason.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmFailureReason.java
index 61754e1556a..e57d74b3b90 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmFailureReason.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmFailureReason.java
@@ -38,6 +38,7 @@ public enum IvmFailureReason {
public boolean requiresCompleteRefresh() {
return this == BINLOG_BROKEN
+ || this == STREAM_UNSUPPORTED
|| this == MIN_MAX_BOUNDARY_HIT
|| this == BITMAP_AGG_DELETE
|| this == PLAN_SIGNATURE_MISMATCH;
diff --git a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTaskTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTaskTest.java
index 0f4d44634ca..47a0d2403e4 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTaskTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTaskTest.java
@@ -25,6 +25,7 @@ import org.apache.doris.catalog.MTMV;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.Partition;
import org.apache.doris.catalog.info.TableNameInfo;
+import org.apache.doris.catalog.stream.OlapTableStream;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
@@ -50,6 +51,7 @@ import org.apache.doris.mtmv.ivm.IvmInfo;
import org.apache.doris.mtmv.ivm.IvmPlanSignature;
import org.apache.doris.mtmv.ivm.IvmPlanSignatureGenerator;
import org.apache.doris.mtmv.ivm.IvmRewriteResult;
+import org.apache.doris.mtmv.ivm.IvmUtil;
import org.apache.doris.nereids.CascadesContext;
import org.apache.doris.nereids.NereidsPlanner;
import org.apache.doris.nereids.StatementContext;
@@ -70,6 +72,7 @@ import org.apache.doris.thrift.TUniqueId;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import org.apache.commons.collections4.CollectionUtils;
import org.junit.jupiter.api.AfterEach;
@@ -142,23 +145,6 @@ public class MTMVTaskTest {
mtmvPartitionUtilStatic.close();
}
- @Test
- public void testCalculateNeedRefreshPartitionsManualComplete() throws
AnalysisException, JobException {
- MTMVTaskContext context =
MTMVTaskContext.of(MTMVTaskTriggerMode.MANUAL, null, RefreshMode.COMPLETE);
- MTMVTask task = new MTMVTask(mtmv, relation, context);
- List<String> result = task.calculateNeedRefreshPartitions(null);
- Assertions.assertEquals(allPartitionNames, result);
- }
-
- @Test
- public void testCalculateNeedRefreshPartitionsManualPartitions() throws
AnalysisException, JobException {
- MTMVTaskContext context =
MTMVTaskContext.of(MTMVTaskTriggerMode.MANUAL, Lists.newArrayList(poneName),
- RefreshMode.AUTO);
- MTMVTask task = new MTMVTask(mtmv, relation, context);
- List<String> result = task.calculateNeedRefreshPartitions(null);
- Assertions.assertEquals(Lists.newArrayList(poneName), result);
- }
-
@Test
public void testGenerateRefreshModeDistinguishesFullAndPartialScope() {
MTMVTask task = new MTMVTask(mtmv, relation, new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL));
@@ -244,92 +230,187 @@ public class MTMVTaskTest {
Assertions.assertEquals(Lists.newArrayList("PARTITIONS"),
toNames(attempts));
}
- private static List<String> toNames(List<?> attempts) {
- List<String> names = Lists.newArrayList();
- for (Object attempt : attempts) {
- names.add(String.valueOf(attempt));
- }
- return names;
- }
-
@Test
- public void testCalculateNeedRefreshPartitionsSystem() throws
AnalysisException, JobException {
-
Mockito.when(mtmvRefreshInfo.getRefreshMethod()).thenReturn(RefreshMethod.AUTO);
- MTMVTaskContext context = new
MTMVTaskContext(MTMVTaskTriggerMode.SYSTEM);
- MTMVTask task = new MTMVTask(mtmv, relation, context);
- List<String> result = task.calculateNeedRefreshPartitions(null);
- Assertions.assertTrue(CollectionUtils.isEmpty(result));
+ public void
testBuildAttemptsGoesStraightToCompleteWhenTheStreamIsUnusable() throws
Exception {
+ Mockito.when(mtmv.isIvm()).thenReturn(true);
+ Mockito.when(mtmv.getName()).thenReturn("test_mv");
+
Mockito.when(mtmvRefreshInfo.getRefreshMethod()).thenReturn(RefreshMethod.INCREMENTAL);
+
Mockito.when(mtmv.getExcludedTriggerTables()).thenReturn(Collections.emptySet());
+
Mockito.when(mtmv.getFullQualifiers()).thenReturn(Lists.newArrayList("internal",
"db", "t1"));
+ // The MV's database holds no stream for the base table.
+
Mockito.when(mtmv.getDatabase()).thenReturn(Mockito.mock(Database.class));
+ mtmvUtilStatic.when(() ->
MTMVUtil.getTable(Mockito.any(BaseTableInfo.class))).thenReturn(mtmv);
+
+ MTMVTask task = new MTMVTask(mtmv, relationWithOneBaseTable(),
MTMVTaskContext.of(
+ MTMVTaskTriggerMode.MANUAL, null, RefreshMode.INCREMENTAL,
true, null));
+ Object request = Deencapsulation.invoke(task, "resolveRefreshRequest");
+ List<?> attempts = (List<?>) Deencapsulation.invoke(task,
"buildAttempts", request, false);
+
+ // Neither the incremental rewrite nor a partition refresh can read a
stream that is not there,
+ // and the IVM attempt would be rejected while a baseline barrier is
pending, so the refresh goes
+ // to the only attempt that reconciles the streams.
+ Assertions.assertEquals(Lists.newArrayList("COMPLETE"),
toNames(attempts));
+ Assertions.assertEquals(IvmFailureReason.STREAM_UNSUPPORTED.name(),
+ Deencapsulation.getField(task, "ivmFallbackReason"));
}
@Test
- public void testPlanPartitionRefreshSelfManageWhenSync() throws Exception {
-
Mockito.when(mtmvPartitionInfo.getPartitionType()).thenReturn(MTMVPartitionType.SELF_MANAGE);
- MTMVTask task = new MTMVTask(mtmv, relation,
- MTMVTaskContext.of(MTMVTaskTriggerMode.MANUAL, null,
RefreshMode.AUTO));
+ public void testBuildAttemptsKeepsTheChainWhenTheStreamsAreUsable() throws
Exception {
+ Mockito.when(mtmv.isIvm()).thenReturn(true);
+ Mockito.when(mtmv.getName()).thenReturn("test_mv");
+ Mockito.when(mtmv.getId()).thenReturn(7L);
+
Mockito.when(mtmv.getFullQualifiers()).thenReturn(Lists.newArrayList("internal",
"db", "t1"));
+
Mockito.when(mtmvRefreshInfo.getRefreshMethod()).thenReturn(RefreshMethod.INCREMENTAL);
+
Mockito.when(mtmv.getExcludedTriggerTables()).thenReturn(Collections.emptySet());
+ OlapTableStream stream = Mockito.mock(OlapTableStream.class);
+ Mockito.when(stream.getBaseTableFullQualifiers())
+ .thenReturn(Lists.newArrayList("internal", "db", "t1"));
+ Mockito.when(stream.isDisabled()).thenReturn(false);
+ Mockito.when(stream.isStale()).thenReturn(false);
+ Mockito.when(stream.getBaseTableNullable()).thenReturn(mtmv);
+ Database mvDb = Mockito.mock(Database.class);
+
Mockito.when(mvDb.getTableNullable(Mockito.anyString())).thenReturn(stream);
+ Mockito.when(mtmv.getDatabase()).thenReturn(mvDb);
+ mtmvUtilStatic.when(() ->
MTMVUtil.getTable(Mockito.any(BaseTableInfo.class))).thenReturn(mtmv);
+
+ MTMVTask task = new MTMVTask(mtmv, relationWithOneBaseTable(),
MTMVTaskContext.of(
+ MTMVTaskTriggerMode.MANUAL, null, RefreshMode.INCREMENTAL,
true, null));
Object request = Deencapsulation.invoke(task, "resolveRefreshRequest");
+ List<?> attempts = (List<?>) Deencapsulation.invoke(task,
"buildAttempts", request, false);
- Object plan = Deencapsulation.invoke(task, "planPartitionRefresh",
- Mockito.mock(MTMVRefreshContext.class), request);
-
- Assertions.assertTrue((Boolean) Deencapsulation.getField(plan,
"canRefreshByPartitions"));
-
Assertions.assertTrue(CollectionUtils.isEmpty(Deencapsulation.getField(plan,
"partitions")));
+ // A usable stream is not a reason to refresh more than the request
asked for.
+ Assertions.assertEquals(Lists.newArrayList("IVM", "PARTITIONS",
"COMPLETE"), toNames(attempts));
}
@Test
- public void testCalculateNeedRefreshPartitionsSystemComplete() throws
AnalysisException, JobException {
- MTMVTaskContext context = new
MTMVTaskContext(MTMVTaskTriggerMode.SYSTEM);
- MTMVTask task = new MTMVTask(mtmv, relation, context);
- List<String> result = task.calculateNeedRefreshPartitions(null);
- Assertions.assertEquals(allPartitionNames, result);
+ public void testBuildAttemptsIgnoresAStreamOnlyTheClosureCarries() throws
Exception {
+ Mockito.when(mtmv.isIvm()).thenReturn(true);
+ Mockito.when(mtmv.getName()).thenReturn("test_mv");
+ Mockito.when(mtmv.getId()).thenReturn(7L);
+
Mockito.when(mtmvRefreshInfo.getRefreshMethod()).thenReturn(RefreshMethod.INCREMENTAL);
+
Mockito.when(mtmv.getExcludedTriggerTables()).thenReturn(Collections.emptySet());
+ // The MV of a chain reads the upstream MV, and the upstream's own
base table is in the relation
+ // only because the closure carries it: (t1) => upstream => mv.
+ OlapTable upstream = Mockito.mock(OlapTable.class);
+
Mockito.when(upstream.getFullQualifiers()).thenReturn(Lists.newArrayList("internal",
"db", "upstream"));
+ OlapTable grandParent = Mockito.mock(OlapTable.class);
+
Mockito.when(grandParent.getFullQualifiers()).thenReturn(Lists.newArrayList("internal",
"db", "t1"));
+ BaseTableInfo upstreamInfo = Mockito.mock(BaseTableInfo.class);
+ BaseTableInfo grandParentInfo = Mockito.mock(BaseTableInfo.class);
+ mtmvUtilStatic.when(() ->
MTMVUtil.getTable(upstreamInfo)).thenReturn(upstream);
+ mtmvUtilStatic.when(() ->
MTMVUtil.getTable(grandParentInfo)).thenReturn(grandParent);
+ // The upstream's stream is there, the grandparent's is not.
+ OlapTableStream stream = Mockito.mock(OlapTableStream.class);
+ Mockito.when(stream.getBaseTableFullQualifiers())
+ .thenReturn(Lists.newArrayList("internal", "db", "upstream"));
+ Mockito.when(stream.isDisabled()).thenReturn(false);
+ Mockito.when(stream.isStale()).thenReturn(false);
+ Mockito.when(stream.getBaseTableNullable()).thenReturn(upstream);
+ Database mvDb = Mockito.mock(Database.class);
+ Mockito.when(mvDb.getTableNullable(IvmUtil.streamName(7L,
upstream.getFullQualifiers())))
+ .thenReturn(stream);
+ Mockito.when(mtmv.getDatabase()).thenReturn(mvDb);
+ MTMVRelation chainedRelation = new
MTMVRelation(Sets.newHashSet(upstreamInfo, grandParentInfo),
+ Sets.newHashSet(upstreamInfo), Sets.newHashSet(upstreamInfo),
Sets.newHashSet(),
+ Sets.newHashSet());
+
+ MTMVTask task = new MTMVTask(mtmv, chainedRelation, MTMVTaskContext.of(
+ MTMVTaskTriggerMode.MANUAL, null, RefreshMode.INCREMENTAL,
true, null));
+ Object request = Deencapsulation.invoke(task, "resolveRefreshRequest");
+ List<?> attempts = (List<?>) Deencapsulation.invoke(task,
"buildAttempts", request, false);
+
+ // No rewrite reads the grandparent's stream, so its absence is no
reason to rebuild the MV.
+ Assertions.assertEquals(Lists.newArrayList("IVM", "PARTITIONS",
"COMPLETE"), toNames(attempts));
+ Assertions.assertNull(Deencapsulation.getField(task,
"ivmFallbackReason"));
}
@Test
- public void
testCalculateNeedRefreshPartitionsSystemIncompleteRefreshSnapshot() throws
AnalysisException, JobException {
-
Mockito.when(mtmvRefreshInfo.getRefreshMethod()).thenReturn(RefreshMethod.AUTO);
- Mockito.when(mtmv.hasRefreshSnapshot()).thenReturn(false);
+ public void testPartitionRefreshChecksOnlyTheStreamsItsPartitionsRead()
throws Exception {
+ // t1 UNION ALL t2, both PCT tables of the MV, each backing one of its
partitions. Refreshing the
+ // t1-backed partition reads t1's stream and leaves t2 to an ordinary
scan, so t2's stream is not
+ // part of this refresh and its absence must not send it to COMPLETE.
+ Mockito.when(mtmv.isIvm()).thenReturn(true);
+ Mockito.when(mtmv.getName()).thenReturn("test_mv");
+ Mockito.when(mtmv.getId()).thenReturn(7L);
+
Mockito.when(mtmv.getExcludedTriggerTables()).thenReturn(Collections.emptySet());
+ OlapTable t1 = mockBaseTable("t1");
+ OlapTable t2 = mockBaseTable("t2");
+ BaseTableInfo t1Info = Mockito.mock(BaseTableInfo.class);
+ BaseTableInfo t2Info = Mockito.mock(BaseTableInfo.class);
+ mtmvUtilStatic.when(() -> MTMVUtil.getTable(t1Info)).thenReturn(t1);
+ mtmvUtilStatic.when(() -> MTMVUtil.getTable(t2Info)).thenReturn(t2);
+ OlapTableStream t1Stream = usableStreamFor(t1);
+ Database mvDb = Mockito.mock(Database.class);
+ Mockito.when(mvDb.getTableNullable(IvmUtil.streamName(7L,
t1.getFullQualifiers())))
+ .thenReturn(t1Stream);
+ Mockito.when(mtmv.getDatabase()).thenReturn(mvDb);
+
Mockito.when(mtmvPartitionInfo.getPctInfos()).thenReturn(Lists.newArrayList(
+ new BaseColInfo("dt", t1Info), new BaseColInfo("dt", t2Info)));
+
+ // The partition this refresh plans reads t1, and t2 keeps its place
in the plan as a plain scan.
+ Map<MTMVRelatedTableIf, Set<String>> mapping = Maps.newHashMap();
+ mapping.put(t1, Sets.newHashSet("p1"));
+ MTMVRefreshContext context = Mockito.mock(MTMVRefreshContext.class);
+
Mockito.when(context.getByPartitionName(Mockito.anyString())).thenReturn(mapping);
+
+ MTMVRelation relation = new MTMVRelation(Sets.newHashSet(t1Info,
t2Info), Sets.newHashSet(t1Info, t2Info),
+ Sets.newHashSet(t1Info, t2Info), Sets.newHashSet(),
Sets.newHashSet());
+ MTMVTask task = new MTMVTask(mtmv, relation, new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL));
- MTMVTaskContext context = new
MTMVTaskContext(MTMVTaskTriggerMode.SYSTEM);
- MTMVTask task = new MTMVTask(mtmv, relation, context);
- List<String> result = task.calculateNeedRefreshPartitions(null);
+ Assertions.assertFalse((Boolean) Deencapsulation.invoke(task,
"hasUnusableIvmStreamForPartitions",
+ context, Lists.newArrayList("p_t1")));
- Assertions.assertTrue(CollectionUtils.isEmpty(result));
- mtmvPartitionUtilStatic.verify(() -> MTMVPartitionUtil.isMTMVSync(
- Mockito.nullable(MTMVRefreshContext.class),
Mockito.nullable(Set.class), Mockito.nullable(Set.class)));
+ // Once a refreshed partition's mapping names t2, its stream is read,
and its absence decides.
+ mapping.put(t2, Sets.newHashSet("p2"));
+ Assertions.assertTrue((Boolean) Deencapsulation.invoke(task,
"hasUnusableIvmStreamForPartitions",
+ context, Lists.newArrayList("p_t1")));
}
- @Test
- public void
testCalculateNeedRefreshPartitionsManualPartitionsIncompleteRefreshSnapshot()
- throws AnalysisException, JobException {
- Mockito.when(mtmv.hasRefreshSnapshot()).thenReturn(false);
+ private OlapTable mockBaseTable(String name) {
+ OlapTable baseTable = Mockito.mock(OlapTable.class);
+ Mockito.when(baseTable.getName()).thenReturn(name);
+
Mockito.when(baseTable.getFullQualifiers()).thenReturn(Lists.newArrayList("internal",
"db", name));
+ return baseTable;
+ }
- MTMVTaskContext context =
MTMVTaskContext.of(MTMVTaskTriggerMode.MANUAL, Lists.newArrayList(poneName),
- RefreshMode.PARTITIONS, false, null);
- MTMVTask task = new MTMVTask(mtmv, relation, context);
- List<String> result = task.calculateNeedRefreshPartitions(null);
+ /** A stream that {@code IvmUtil.isIvmStreamUsable} accepts for the given
base table. */
+ private OlapTableStream usableStreamFor(OlapTable baseTable) {
+ List<String> qualifiers = baseTable.getFullQualifiers();
+ OlapTableStream stream = Mockito.mock(OlapTableStream.class);
+
Mockito.when(stream.getBaseTableFullQualifiers()).thenReturn(qualifiers);
+ Mockito.when(stream.isDisabled()).thenReturn(false);
+ Mockito.when(stream.isStale()).thenReturn(false);
+ Mockito.when(stream.getBaseTableNullable()).thenReturn(baseTable);
+ return stream;
+ }
- Assertions.assertEquals(Lists.newArrayList(poneName), result);
+ private MTMVRelation relationWithOneBaseTable() {
+ BaseTableInfo baseTable = Mockito.mock(BaseTableInfo.class);
+ // A table of the query is in the plan, in the first level of the
query, and in the closure.
+ return new MTMVRelation(Sets.newHashSet(baseTable),
Sets.newHashSet(baseTable),
+ Sets.newHashSet(baseTable), Sets.newHashSet(),
Sets.newHashSet());
}
- @Test
- public void testCalculateNeedRefreshPartitionsSystemNotSyncComplete()
throws AnalysisException, JobException {
- mtmvPartitionUtilStatic.when(() ->
MTMVPartitionUtil.isMTMVSync(Mockito.nullable(MTMVRefreshContext.class),
Mockito.nullable(Set.class), Mockito.nullable(Set.class))).thenReturn(false);
- MTMVTaskContext context = new
MTMVTaskContext(MTMVTaskTriggerMode.SYSTEM);
- MTMVTask task = new MTMVTask(mtmv, relation, context);
- List<String> result = task.calculateNeedRefreshPartitions(null);
- Assertions.assertEquals(allPartitionNames, result);
+ private static List<String> toNames(List<?> attempts) {
+ List<String> names = Lists.newArrayList();
+ for (Object attempt : attempts) {
+ names.add(String.valueOf(attempt));
+ }
+ return names;
}
@Test
- public void testCalculateNeedRefreshPartitionsSystemNotSyncAuto() throws
AnalysisException, JobException {
- mtmvPartitionUtilStatic.when(() ->
MTMVPartitionUtil.isMTMVSync(Mockito.nullable(MTMVRefreshContext.class),
Mockito.nullable(Set.class), Mockito.nullable(Set.class))).thenReturn(false);
+ public void testPlanPartitionRefreshSelfManageWhenSync() throws Exception {
+
Mockito.when(mtmvPartitionInfo.getPartitionType()).thenReturn(MTMVPartitionType.SELF_MANAGE);
+ MTMVTask task = new MTMVTask(mtmv, relation,
+ MTMVTaskContext.of(MTMVTaskTriggerMode.MANUAL, null,
RefreshMode.AUTO));
+ Object request = Deencapsulation.invoke(task, "resolveRefreshRequest");
-
Mockito.when(mtmvRefreshInfo.getRefreshMethod()).thenReturn(RefreshMethod.AUTO);
+ Object plan = Deencapsulation.invoke(task, "planPartitionRefresh",
+ Mockito.mock(MTMVRefreshContext.class), request);
- mtmvPartitionUtilStatic.when(() ->
MTMVPartitionUtil.getMTMVNeedRefreshPartitions(Mockito.nullable(MTMVRefreshContext.class),
Mockito.nullable(Set.class))).thenReturn(Lists.newArrayList(ptwoName));
- MTMVTaskContext context = new
MTMVTaskContext(MTMVTaskTriggerMode.SYSTEM);
- MTMVTask task = new MTMVTask(mtmv, relation, context);
- List<String> result = task.calculateNeedRefreshPartitions(null);
- Assertions.assertEquals(Lists.newArrayList(ptwoName), result);
+ Assertions.assertTrue((Boolean) Deencapsulation.getField(plan,
"canRefreshByPartitions"));
+
Assertions.assertTrue(CollectionUtils.isEmpty(Deencapsulation.getField(plan,
"partitions")));
}
@Test
@@ -377,14 +458,14 @@ public class MTMVTaskTest {
}
@Test
- public void testMvDefaultUnknownRefreshMethodRejected() throws
AnalysisException {
+ public void testMvDefaultUnknownRefreshMethodRejected() {
Mockito.when(mtmv.getName()).thenReturn("test_mv");
Mockito.when(mtmvRefreshInfo.getRefreshMethod()).thenReturn(null);
MTMVTaskContext context =
MTMVTaskContext.forMvDefault(MTMVTaskTriggerMode.SYSTEM);
MTMVTask task = new MTMVTask(mtmv, relation, context);
JobException exception = Assertions.assertThrows(JobException.class,
- () -> task.calculateNeedRefreshPartitions(null));
+ () -> Deencapsulation.invoke(task, "resolveRefreshRequest"));
Assertions.assertTrue(exception.getMessage().contains("unknown refresh
method"));
}
@@ -832,6 +913,95 @@ public class MTMVTaskTest {
Mockito.verify(mtmv,
Mockito.never()).releaseIvmBaselineRebuild(Mockito.anyLong());
}
+ @Test
+ public void testPendingBaselineRebuildChecksTheStreamsItsPartitionsRead()
throws Exception {
+ // A partial barrier left by an earlier failed refresh. The pre-step
rebuilds those partitions
+ // before the attempts run, and that rebuild reads their streams, so a
stream missing for them
+ // decides the request just as it does for the partition attempt --
and PARTITIONS FALLBACK
+ // reaches this pre-step without an IVM attempt for buildAttempts to
have judged.
+ Mockito.when(mtmv.isIvm()).thenReturn(true);
+ Mockito.when(mtmv.getName()).thenReturn("test_mv");
+ Mockito.when(mtmv.getId()).thenReturn(7L);
+
Mockito.when(mtmv.getExcludedTriggerTables()).thenReturn(Collections.emptySet());
+ IvmInfo ivmInfo = new IvmInfo();
+ ivmInfo.addPendingBaselineRebuildPartitions(Sets.newHashSet(poneName));
+ Mockito.when(mtmv.getIvmInfo()).thenReturn(ivmInfo);
+ OlapTable t1 = mockBaseTable("t1");
+ BaseTableInfo t1Info = Mockito.mock(BaseTableInfo.class);
+ mtmvUtilStatic.when(() -> MTMVUtil.getTable(t1Info)).thenReturn(t1);
+ // t1 is not a PCT table, so every partition this rebuild refreshes
reads it through its stream,
+ // and the MV's database holds no stream for it.
+
Mockito.when(mtmv.getDatabase()).thenReturn(Mockito.mock(Database.class));
+ MTMVRefreshContext context = Mockito.mock(MTMVRefreshContext.class);
+
Mockito.when(context.getByPartitionName(Mockito.anyString())).thenReturn(Maps.newHashMap());
+
+ MTMVRelation relation = new MTMVRelation(Sets.newHashSet(t1Info),
Sets.newHashSet(t1Info),
+ Sets.newHashSet(t1Info), Sets.newHashSet(), Sets.newHashSet());
+ MTMVTask task = new MTMVTask(mtmv, relation, MTMVTaskContext.of(
+ MTMVTaskTriggerMode.MANUAL, null, RefreshMode.PARTITIONS,
true, null));
+ Object request = Deencapsulation.invoke(task, "resolveRefreshRequest");
+ List<Object> attempts = Lists.newArrayList();
+ attempts.addAll(Deencapsulation.invoke(task, "buildAttempts", request,
false));
+ Assertions.assertEquals("[PARTITIONS, COMPLETE]", attempts.toString());
+
+ try {
+ Deencapsulation.invoke(task, "handlePendingIvmBaselineRebuild",
context, request,
+ new ConnectContext(), attempts);
+ } catch (Exception expected) {
+ // Without the stream check the pre-step rebuilds inline, and how
far that rebuild gets
+ // against these mocks is not what this test is about; the
attempts it leaves behind are.
+ }
+
+ // The rebuild that cannot read its streams is skipped rather than
attempted: its own barrier
+ // would have guarded nothing but the data it never wrote, and the
COMPLETE attempt left in the
+ // list reconciles the stream and clears the barrier that is already
pending.
+ Assertions.assertEquals("[COMPLETE]", attempts.toString());
+ Assertions.assertEquals(IvmFailureReason.STREAM_UNSUPPORTED.name(),
+ Deencapsulation.getField(task, "ivmFallbackReason"));
+ Mockito.verify(mtmv,
Mockito.never()).persistIvmBaselineGuard(Mockito.any(), Mockito.anySet(),
+ Mockito.anyLong());
+ Mockito.verify(mtmv,
Mockito.never()).releaseIvmBaselineRebuild(Mockito.anyLong());
+
+ // The same rebuild without fallback is not covered by a COMPLETE
attempt, so it fails here
+ // rather than starting a rebuild that cannot read its streams.
+ MTMVTask strictTask = new MTMVTask(mtmv, relation, MTMVTaskContext.of(
+ MTMVTaskTriggerMode.MANUAL, null, RefreshMode.PARTITIONS,
false, null));
+ Object strictRequest = Deencapsulation.invoke(strictTask,
"resolveRefreshRequest");
+ List<Object> strictAttempts = Lists.newArrayList();
+ strictAttempts.addAll(Deencapsulation.invoke(strictTask,
"buildAttempts", strictRequest, false));
+ Assertions.assertEquals("[PARTITIONS]", strictAttempts.toString());
+
+ JobException exception = Assertions.assertThrows(JobException.class,
+ () -> Deencapsulation.invoke(strictTask,
"handlePendingIvmBaselineRebuild", context,
+ strictRequest, new ConnectContext(), strictAttempts));
+
+ Assertions.assertTrue(exception.getMessage().contains("IVM stream is
unusable"));
+ }
+
+ @Test
+ public void testCompleteAttemptWritesTheBarrierBeforeReconcilingStreams()
throws Exception {
+ Mockito.when(mtmv.isIvm()).thenReturn(true);
+
Mockito.when(mtmv.getPartitionNames()).thenReturn(Sets.newHashSet(poneName));
+ // The reconcile starts from the MV's database, which is what makes it
visible to the order check.
+
Mockito.when(mtmv.getDatabase()).thenReturn(Mockito.mock(Database.class));
+ MTMVTask task = new MTMVTask(mtmv, relation, new
MTMVTaskContext(MTMVTaskTriggerMode.MANUAL));
+ InOrder inOrder = Mockito.inOrder(mtmv);
+
+ try {
+ Deencapsulation.invoke(task, "executeCompleteAttempt",
+ Mockito.mock(MTMVRefreshContext.class), new
ConnectContext());
+ } catch (Exception expected) {
+ // How far the rebuild itself gets is not what this test is about.
+ }
+
+ // A recreated stream starts from the base table's current rows, so
the barrier that makes the
+ // next refresh rebuild the MV has to be durable before the stream is
replaced. The other order
+ // loses those rows with no error anywhere.
+ inOrder.verify(mtmv).persistIvmBaselineGuard(Mockito.any(),
Mockito.anySet(), Mockito.anyLong());
+ inOrder.verify(mtmv).getDatabase();
+ }
+
+
@Test
public void testDroppedBaselinePartitionsReleaseBarrierWithoutRebuild()
throws Exception {
Mockito.when(mtmv.isIvm()).thenReturn(true);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmFailureReasonTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmFailureReasonTest.java
index 3660f830ad3..46df766b99a 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmFailureReasonTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmFailureReasonTest.java
@@ -25,6 +25,9 @@ public class IvmFailureReasonTest {
@Test
public void testRequiresCompleteRefresh() {
Assertions.assertTrue(IvmFailureReason.BINLOG_BROKEN.requiresCompleteRefresh());
+ // An unusable stream can only be recreated by the COMPLETE path's
reconcile step; a
+ // PARTITIONS fallback would read the same stream and fail again.
+
Assertions.assertTrue(IvmFailureReason.STREAM_UNSUPPORTED.requiresCompleteRefresh());
Assertions.assertTrue(IvmFailureReason.MIN_MAX_BOUNDARY_HIT.requiresCompleteRefresh());
Assertions.assertTrue(IvmFailureReason.BITMAP_AGG_DELETE.requiresCompleteRefresh());
Assertions.assertTrue(IvmFailureReason.PLAN_SIGNATURE_MISMATCH.requiresCompleteRefresh());
diff --git a/regression-test/data/mtmv_p0/ivm/test_ivm_chained_stream_scope.out
b/regression-test/data/mtmv_p0/ivm/test_ivm_chained_stream_scope.out
new file mode 100644
index 00000000000..d11be25f06d
--- /dev/null
+++ b/regression-test/data/mtmv_p0/ivm/test_ivm_chained_stream_scope.out
@@ -0,0 +1,6 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !downstream_mv --
+1 10
+2 20
+3 30
+
diff --git
a/regression-test/data/mtmv_p0/ivm/test_ivm_partitions_fallback_stream_unusable.out
b/regression-test/data/mtmv_p0/ivm/test_ivm_partitions_fallback_stream_unusable.out
new file mode 100644
index 00000000000..7b7862c7774
--- /dev/null
+++
b/regression-test/data/mtmv_p0/ivm/test_ivm_partitions_fallback_stream_unusable.out
@@ -0,0 +1,13 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !baseline_mv --
+1 2026-01-10 100 dim-a
+2 2026-02-10 200 dim-b
+
+-- !reconciled_mv --
+1 2026-01-10 100 dim-a
+2 2026-02-10 200 dim-b
+
+-- !after_strict_failure_mv --
+1 2026-01-10 100 dim-a
+2 2026-02-10 200 dim-b
+
diff --git
a/regression-test/suites/mtmv_p0/ivm/test_ivm_chained_stream_scope.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_chained_stream_scope.groovy
new file mode 100644
index 00000000000..895d42646fd
--- /dev/null
+++ b/regression-test/suites/mtmv_p0/ivm/test_ivm_chained_stream_scope.groovy
@@ -0,0 +1,162 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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
+//
+// http://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.
+
+import org.awaitility.Awaitility
+
+import static java.util.concurrent.TimeUnit.SECONDS
+
+/**
+ * Which streams decide that a refresh needs the complete attempt.
+ *
+ * <p>An IVM MV is created with a stream for every base table in its relation,
and for a chained MV the
+ * relation carries more than the plan reads: mv2 reads mv1, while the table
behind mv1 is in mv2's
+ * relation only through the closure. Both rewrites that read streams -- the
incremental delta rewriter
+ * and the full refresh -- take the streams of the plan's scans, so mv2's
stream on that table is never
+ * read, and its absence must not turn mv2's incremental refresh into a
rebuild of the whole MV.
+ *
+ * <p>Pinned here:
+ * <ol>
+ * <li>the unused stream exists, so the setup is the one the case is
about;</li>
+ * <li>an incremental refresh of the upstream MV succeeds with it dropped,
which is what makes the
+ * refresh below able to stay incremental;</li>
+ * <li>the downstream MV still refreshes incrementally rather than falling
back to a complete
+ * refresh for a stream nothing reads.</li>
+ * </ol>
+ */
+suite("test_ivm_chained_stream_scope") {
+ def baseTable = "ivm_chained_scope_base"
+ def upstreamMv = "ivm_chained_scope_up"
+ def downstreamMv = "ivm_chained_scope_down"
+
+ def waitForNewTask = { String mvName, String previousTaskId ->
+ def taskResult
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
+ taskResult = sql_return_maparray("""
+ SELECT TaskId, Status
+ FROM tasks('type'='mv')
+ WHERE MvDatabaseName = '${context.dbName}'
+ AND MvName = '${mvName}'
+ ORDER BY CreateTime DESC, TaskId DESC LIMIT 1
+ """)
+ return !taskResult.isEmpty()
+ && taskResult[0].TaskId.toString() != previousTaskId
+ && taskResult[0].Status.toString() != 'PENDING'
+ && taskResult[0].Status.toString() != 'RUNNING'
+ })
+ return taskResult[0].TaskId.toString()
+ }
+
+ def taskRecord = { String taskId ->
+ // Unset RefreshMode comes back as the literal two-character string
"\N", which does not survive
+ // the .out round trip, so fold the unset value into a printable token.
+ return sql_return_maparray("""
+ SELECT Status,
+ CASE WHEN RefreshMode IN ('COMPLETE', 'PARTIAL',
'NOT_REFRESH')
+ THEN RefreshMode ELSE 'NONE' END AS RefreshMode
+ FROM tasks('type'='mv')
+ WHERE TaskId = '${taskId}'
+ """)[0]
+ }
+
+ def mvId = { String mvName ->
+ def rows = sql_return_maparray("""
+ SELECT Id FROM mv_infos('database'='${context.dbName}') WHERE Name
= '${mvName}'
+ """)
+ return rows[0].Id.toString()
+ }
+
+ // Streams are named after the MV that owns them, so the ones of the
downstream MV on the upstream's
+ // base table can be told apart from the ones the upstream MV has on it.
+ def streamNamesOf = { String mvName, String baseTableName ->
+ def rows = sql_return_maparray("""
+ SELECT STREAM_NAME FROM information_schema.table_streams
+ WHERE DB_NAME = '${context.dbName}' AND BASE_TABLE_NAME =
'${baseTableName}'
+ """)
+ def prefix = "__doris_ivm_stream_${mvId(mvName)}_"
+ return rows.collect { it.STREAM_NAME.toString() }.findAll {
it.startsWith(prefix) }
+ }
+
+ sql """DROP MATERIALIZED VIEW IF EXISTS ${downstreamMv}"""
+ sql """DROP MATERIALIZED VIEW IF EXISTS ${upstreamMv}"""
+ sql """DROP TABLE IF EXISTS ${baseTable}"""
+
+ sql """
+ CREATE TABLE ${baseTable} (
+ k1 INT NOT NULL,
+ v1 INT
+ )
+ UNIQUE KEY(k1)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true"
+ )
+ """
+ sql """INSERT INTO ${baseTable} VALUES (1, 10), (2, 20)"""
+
+ // The upstream MV carries row binlog, which is what the downstream MV
reads its changes from.
+ sql """
+ CREATE MATERIALIZED VIEW ${upstreamMv}
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ DISTRIBUTED BY RANDOM BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true"
+ )
+ AS SELECT * FROM ${baseTable}
+ """
+ sql """REFRESH MATERIALIZED VIEW ${upstreamMv} COMPLETE"""
+ def upstreamTaskId = waitForNewTask(upstreamMv, null)
+
+ sql """
+ CREATE MATERIALIZED VIEW ${downstreamMv}
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ DISTRIBUTED BY RANDOM BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ AS SELECT k1, v1 FROM ${upstreamMv}
+ """
+ sql """REFRESH MATERIALIZED VIEW ${downstreamMv} COMPLETE"""
+ def downstreamTaskId = waitForNewTask(downstreamMv, null)
+
+ // The downstream MV's plan scans the upstream MV only, but its relation
also carries the upstream's
+ // base table, and the MV was created with a stream for it.
+ def unusedStreams = streamNamesOf(downstreamMv, baseTable)
+ assertEquals(1, unusedStreams.size())
+ sql """DROP STREAM ${context.dbName}.${unusedStreams[0]} FORCE"""
+ assertEquals(0, streamNamesOf(downstreamMv, baseTable).size())
+
+ // One change reaches the downstream MV through the upstream MV, so the
refresh below has something
+ // to apply and cannot pass as an empty incremental refresh.
+ sql """INSERT INTO ${baseTable} VALUES (3, 30)"""
+ sql """REFRESH MATERIALIZED VIEW ${upstreamMv} INCREMENTAL"""
+ upstreamTaskId = waitForNewTask(upstreamMv, upstreamTaskId)
+ assertEquals("SUCCESS", taskRecord(upstreamTaskId).Status.toString())
+
+ sql """REFRESH MATERIALIZED VIEW ${downstreamMv} AUTO"""
+ downstreamTaskId = waitForNewTask(downstreamMv, downstreamTaskId)
+ def downstreamTask = taskRecord(downstreamTaskId)
+ // An incremental refresh of an IVM MV records no refresh mode, so an
unset mode is the incremental
+ // attempt succeeding; COMPLETE would say the whole MV was rebuilt.
+ assertEquals("SUCCESS", downstreamTask.Status.toString())
+ assertEquals("NONE", downstreamTask.RefreshMode.toString())
+ order_qt_downstream_mv """SELECT k1, v1 FROM ${downstreamMv}"""
+}
diff --git
a/regression-test/suites/mtmv_p0/ivm/test_ivm_partitions_fallback_stream_unusable.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_partitions_fallback_stream_unusable.groovy
new file mode 100644
index 00000000000..c9ca457fab8
--- /dev/null
+++
b/regression-test/suites/mtmv_p0/ivm/test_ivm_partitions_fallback_stream_unusable.groovy
@@ -0,0 +1,187 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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
+//
+// http://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.
+
+import org.awaitility.Awaitility
+
+import static java.util.concurrent.TimeUnit.SECONDS
+
+/**
+ * What a partition-based refresh does when the MV's base table streams are
gone.
+ *
+ * <p>A partition refresh reads the stream of every base table it realigns
from -- the PCT tables of the
+ * partitions it rebuilds, and the non-PCT tables the MV joins -- so a missing
stream makes it fail,
+ * and only the COMPLETE attempt reconciles streams. A request that allows
falling back must therefore
+ * end up doing COMPLETE rather than failing the task; a request that does not
allow it must keep
+ * failing, because COMPLETE would refresh more than it was asked to.
+ *
+ * <p>Driven by dropping the joined table's stream by hand: that is the state
the fallback exists for and the one a
+ * unit test can only mock.
+ */
+suite("test_ivm_partitions_fallback_stream_unusable") {
+ def factTable = "ivm_stream_f"
+ def dimTable = "ivm_stream_d"
+ def mvName = "ivm_stream_mv"
+
+ def waitForNewTask = { previousTaskId ->
+ def taskResult
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
+ taskResult = sql_return_maparray("""
+ SELECT TaskId, Status
+ FROM tasks('type'='mv')
+ WHERE MvDatabaseName = '${context.dbName}'
+ AND MvName = '${mvName}'
+ ORDER BY CreateTime DESC, TaskId DESC LIMIT 1
+ """)
+ return !taskResult.isEmpty()
+ && taskResult[0].TaskId.toString() != previousTaskId
+ && taskResult[0].Status.toString() != 'PENDING'
+ && taskResult[0].Status.toString() != 'RUNNING'
+ })
+ return taskResult[0].TaskId.toString()
+ }
+
+ // Unset IvmFallbackReason comes back as the literal two-character string
"\N", which does not
+ // survive a .out round trip, so fold it into a printable token.
+ def taskOutcome = { String taskId ->
+ def rows = sql_return_maparray("""
+ SELECT Status,
+ CASE WHEN RefreshMode IN ('COMPLETE', 'PARTIAL',
'NOT_REFRESH')
+ THEN RefreshMode ELSE 'NONE' END AS Mode,
+ CASE WHEN IvmFallbackReason = 'BINLOG_BROKEN' OR
IvmFallbackReason = 'STREAM_UNSUPPORTED'
+ THEN IvmFallbackReason ELSE 'NONE' END AS Fallback
+ FROM tasks('type'='mv')
+ WHERE TaskId = '${taskId}'
+ """)
+ // toString: a GString never equals a String, and these are compared
against literals.
+ return
"${rows[0].Status}\t${rows[0].Mode}\t${rows[0].Fallback}".toString()
+ }
+
+ def taskStatus = { String taskId ->
+ def rows = sql_return_maparray("""
+ SELECT Status FROM tasks('type'='mv') WHERE TaskId = '${taskId}'
+ """)
+ return rows[0].Status.toString()
+ }
+
+ // The streams of every suite in this directory live in the one database
the runner gives them, so
+ // they have to be looked up by base table: taking every stream of the
database would take the
+ // streams of the suites running next to this one, and fail them with "IVM
stream not found".
+ def streamNameOf = { String tableName ->
+ def rows = sql("""
+ SELECT STREAM_NAME FROM information_schema.table_streams
+ WHERE DB_NAME = '${context.dbName}' AND BASE_TABLE_NAME =
'${tableName}'
+ """)
+ return rows.isEmpty() ? null : rows[0][0].toString()
+ }
+
+ def dropStreamOf = { String tableName ->
+ def name = streamNameOf(tableName)
+ assertTrue(name != null, "the MV should have created a stream for " +
tableName)
+ sql """DROP STREAM ${name} FORCE"""
+ }
+
+ sql """DROP MATERIALIZED VIEW IF EXISTS ${mvName}"""
+ sql """DROP TABLE IF EXISTS ${factTable}"""
+ sql """DROP TABLE IF EXISTS ${dimTable}"""
+
+ sql """
+ CREATE TABLE ${factTable} (
+ order_id BIGINT NOT NULL,
+ dt DATE NOT NULL,
+ dimension_id INT,
+ amount INT
+ )
+ UNIQUE KEY(order_id, dt)
+ PARTITION BY RANGE(dt) ()
+ DISTRIBUTED BY HASH(order_id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true"
+ )
+ """
+ sql """ALTER TABLE ${factTable} ADD PARTITION p202601 VALUES
[('2026-01-01'), ('2026-02-01'))"""
+ sql """ALTER TABLE ${factTable} ADD PARTITION p202602 VALUES
[('2026-02-01'), ('2026-03-01'))"""
+
+ // Partitioned as well, but joined on a non-partition column: a base table
of the MV that is not one
+ // of its PCT tables, and therefore read through its stream by a partition
refresh.
+ sql """
+ CREATE TABLE ${dimTable} (
+ dimension_id INT NOT NULL,
+ dt DATE NOT NULL,
+ dimension_name VARCHAR(32)
+ )
+ UNIQUE KEY(dimension_id, dt)
+ PARTITION BY RANGE(dt) ()
+ DISTRIBUTED BY HASH(dimension_id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true"
+ )
+ """
+ sql """ALTER TABLE ${dimTable} ADD PARTITION d202601 VALUES
[('2026-01-01'), ('2026-02-01'))"""
+ sql """ALTER TABLE ${dimTable} ADD PARTITION d202602 VALUES
[('2026-02-01'), ('2026-03-01'))"""
+
+ sql """INSERT INTO ${dimTable} VALUES (10, '2026-01-15', 'dim-a'), (20,
'2026-02-15', 'dim-b')"""
+ sql """INSERT INTO ${factTable} VALUES
+ (1, '2026-01-10', 10, 100),
+ (2, '2026-02-10', 20, 200)"""
+
+ sql """
+ CREATE MATERIALIZED VIEW ${mvName}
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ KEY(order_id, dt)
+ PARTITION BY(dt)
+ DISTRIBUTED BY HASH(order_id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ AS SELECT f.order_id, f.dt, f.amount, d.dimension_name
+ FROM ${factTable} f
+ LEFT JOIN ${dimTable} d ON f.dimension_id = d.dimension_id
+ """
+
+ sql """REFRESH MATERIALIZED VIEW ${mvName} COMPLETE"""
+ def taskId = waitForNewTask(null)
+ assertEquals("SUCCESS\tCOMPLETE\tNONE", taskOutcome(taskId))
+ order_qt_baseline_mv """SELECT order_id, dt, amount, dimension_name FROM
${mvName}"""
+
+ // With the streams gone, a partition refresh cannot read anything.
Falling back is what makes it
+ // reach the COMPLETE attempt, which reconciles the streams and rebuilds
the MV; without it the
+ // refresh would fail and stay failed.
+ dropStreamOf(dimTable)
+ sql """REFRESH MATERIALIZED VIEW ${mvName} PARTITIONS FALLBACK"""
+ taskId = waitForNewTask(taskId)
+ assertEquals("SUCCESS\tCOMPLETE\tSTREAM_UNSUPPORTED", taskOutcome(taskId))
+ assertTrue(streamNameOf(dimTable) != null, "the complete attempt should
have reconciled the stream")
+ order_qt_reconciled_mv """SELECT order_id, dt, amount, dimension_name FROM
${mvName}"""
+
+ // The same request without fallback keeps failing: COMPLETE refreshes
more than the request asked
+ // for, so it may only be reached when the user allowed it. Its failure
leaves the MV as it was.
+ //
+ // The base table has to be out of sync for the partition refresh to do
anything at all: an MV that
+ // is already up to date refreshes no partition, and therefore never reads
a stream.
+ sql """INSERT INTO ${factTable} VALUES (3, '2026-01-20', 10, 300)"""
+ dropStreamOf(dimTable)
+ sql """REFRESH MATERIALIZED VIEW ${mvName} PARTITIONS"""
+ taskId = waitForNewTask(taskId)
+ assertEquals("FAILED", taskStatus(taskId))
+ order_qt_after_strict_failure_mv """SELECT order_id, dt, amount,
dimension_name FROM ${mvName}"""
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
