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 f8ed33fa701 [fix](ivm) Refresh the surviving partitions after an IVM
baseline rebuild (#67802)
f8ed33fa701 is described below
commit f8ed33fa70133b7c0dac0c7178dedc8d76e8b535
Author: yujun <[email protected]>
AuthorDate: Fri Sep 11 17:21:38 2026 +0800
[fix](ivm) Refresh the surviving partitions after an IVM baseline rebuild
(#67802)
### What problem does this PR solve?
Dropping a base-table partition marks the IVM baseline as broken,
because those rows disappear through metadata rather than through row
binlog entries. Partition sync then removes the very MV partitions that
barrier named, so `handlePendingIvmBaselineRebuild` intersected the
barrier with the current MV partitions and always got an empty set. The
refresh reported SUCCESS with refresh mode `NOT_REFRESH` and cleared the
barrier without refreshing anything, leaving the delta that had
accumulated on the surviving partitions unapplied: the task claimed the
MV was up to date while the MV was missing rows.
Reproduction: drop a base partition, insert a row into a surviving
partition, then run `REFRESH MATERIALIZED VIEW ... INCREMENTAL
FALLBACK`. The task succeeds, the expired MV partition is dropped, and
the new row never reaches the MV.
### What changed
- `MTMVTask.handlePendingIvmBaselineRebuild` is a pre-step now instead
of a terminal branch. It rebuilds only the barrier partitions that still
exist, releases the barrier, and then lets the normal attempt list run,
so the surviving partitions catch up in the same task. Partitions the
barrier named that partition sync already dropped need no rebuild: the
partition and its IVM offsets are both gone, which is what satisfies
those entries.
- The attempt list is rewritten in place. A lone `COMPLETE` attempt
rebuilds everything anyway and needs no pre-rebuild; a complete baseline
rebuild rewrites the list to `COMPLETE` instead of executing it inline.
- `MTMV.releaseIvmBaselineRebuild` releases the barrier under a
`schemaChangeVersion` compare-and-clear and journals the new state
immediately, like the other `ivmInfo` mutations. Without the release the
IVM attempt that follows would be rejected by `validateIvmRefreshStart`;
without the compare-and-clear a concurrent base-table change would lose
the barrier entry it had just recorded.
- `MTMVRelationManager.markIvmBaselineRebuild` takes an explicit
all-partitions-changed flag instead of inferring it from an empty
partition map.
The COMPLETE paths are unchanged. They end the task, so the existing
`addTaskResult` cleanup releases the barrier once the refresh succeeds.
A strict `REFRESH ... INCREMENTAL`, and a scheduled refresh of an MV
declared without `FALLBACK`, still fail while the barrier is pending,
because that check runs before partition sync. Relaxing it is left to a
follow-up PR.
### Test
-
`regression-test/suites/mtmv_p0/ivm/test_ivm_partition_drop_live_delta.groovy`
(new): partitions added and dropped by hand with literal dates and no
dynamic-partition scheduler, so the case is deterministic. Asserts the
MV matches the base table right after `INCREMENTAL FALLBACK`. The
scenario was reproduced on the pre-fix build, where the fallback task
reported SUCCESS while the MV was missing the row written to the
surviving partition.
-
`regression-test/suites/mtmv_p0/ivm/test_ivm_partition_baseline_rebuild_dup_keys.groovy`
(new): covers the `TRUNCATE` path, where the affected MV partition
survives and the pre-step really has something to rebuild. Runs on a
duplicate-key MV with repeated identical rows, so a delta applied twice
would show up as extra rows instead of hiding behind a unique key.
- `test_ivm_partition_baseline_rebuild`,
`test_ivm_partition_sync_retry`, `test_ivm_partition_unique_key`,
`test_ivm_partition_window_limit`, `test_ivm_partition_window_remove`,
`test_ivm_one_row_relation_partitioned`,
`test_ivm_strict_failure_partition_atomicity`
Trace issue: https://github.com/apache/doris/issues/65418
---
.../main/java/org/apache/doris/catalog/MTMV.java | 35 ++++++
.../apache/doris/job/extensions/mtmv/MTMVTask.java | 56 ++++++---
.../org/apache/doris/mtmv/MTMVRelationManager.java | 8 +-
.../java/org/apache/doris/mtmv/MTMVTaskTest.java | 42 ++++++-
...est_ivm_partition_baseline_rebuild_dup_keys.out | 32 +++++
.../ivm/test_ivm_partition_drop_live_delta.out | 45 +++++++
..._ivm_partition_baseline_rebuild_dup_keys.groovy | 123 +++++++++++++++++++
.../ivm/test_ivm_partition_drop_live_delta.groovy | 134 +++++++++++++++++++++
8 files changed, 452 insertions(+), 23 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java
index 354081ef5b6..b162a2b2d51 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java
@@ -645,6 +645,41 @@ public class MTMV extends OlapTable {
editLogItem.await();
}
+ /**
+ * Release the IVM baseline barrier after the partitions it named have
been rebuilt, or after
+ * partition sync removed them (a dropped partition resolves its own
entry: the partition and its
+ * IVM offsets are both gone).
+ *
+ * <p>Guarded by schemaChangeVersion, like {@link
#persistIvmBaselineGuard}: a base-table change
+ * landing while the rebuild runs carries its own barrier entry, and a
blind clear would swallow
+ * it. Failing instead preserves that entry -- the next refresh rebuilds
it together with the
+ * partitions this task handled.
+ *
+ * <p>Journals the new state right away, like every other ivmInfo mutation
here. A task that dies
+ * before {@link #addTaskResult} would otherwise leave the release in
memory only, and a restart
+ * would resurrect the barrier from disk.
+ */
+ public void releaseIvmBaselineRebuild(long expectedSchemaChangeVersion)
throws JobException {
+ EditLogItem editLogItem;
+ writeMvLock();
+ try {
+ if (ivmInfo == null || !ivmInfo.isBaselineRebuildRequired()) {
+ // Nothing to release: skip both the mutation and the journal
entry. Any base-table
+ // change that raced us in is still caught by
validateIvmRefreshStart() below.
+ return;
+ }
+ if (schemaChangeVersion != expectedSchemaChangeVersion) {
+ throw new JobException("Base table metadata changed before IVM
baseline refresh, mv="
+ + getName());
+ }
+ ivmInfo.clearBaselineRebuild();
+ editLogItem = submitIvmInfoChange();
+ } finally {
+ writeMvUnlock();
+ }
+ editLogItem.await();
+ }
+
public void persistIvmBaselineGuard(RefreshMode refreshMode, Set<String>
baselinePartitions,
long expectedSchemaChangeVersion) throws JobException {
EditLogItem editLogItem;
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 d97e142fee9..1eab83fe576 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
@@ -319,9 +319,7 @@ public class MTMVTask extends AbstractTask {
throw new JobException(e.getMessage(), e);
}
MTMVRefreshContext refreshContext = buildRefreshContext(tableIfs);
- if (handlePendingIvmBaselineRebuild(refreshContext, request, ctx))
{
- return;
- }
+ handlePendingIvmBaselineRebuild(refreshContext, request, ctx,
attempts);
boolean disablePartitionRefresh = false;
for (RefreshAttemptType attemptType : attempts) {
switch (attemptType) {
@@ -558,28 +556,56 @@ public class MTMVTask extends AbstractTask {
executePartitionBasedRefresh(context, RefreshMode.COMPLETE, ctx);
}
- private boolean handlePendingIvmBaselineRebuild(MTMVRefreshContext
context, RefreshRequest request,
- ConnectContext ctx)
+ /**
+ * Rebuild the MV partitions whose IVM baseline is broken, before the
normal refresh runs.
+ *
+ * <p>This is a pre-step, not a terminal branch: the caller keeps running
{@code attempts}
+ * afterwards, so a broken baseline no longer skips the refresh entirely.
The list is rewritten
+ * in place when the baseline demands a different set of attempts.
+ *
+ * <p>Partition sync drops the MV partitions whose base partition
disappeared, which is exactly
+ * what the barrier recorded when that base partition was dropped. Those
partitions are resolved
+ * by the drop itself (the partition and its IVM offsets are both gone),
so only the partitions
+ * that still exist need a rebuild. The barrier is released either way,
otherwise the IVM attempt
+ * that follows would be rejected by {@link MTMV#validateIvmRefreshStart}.
+ */
+ private void handlePendingIvmBaselineRebuild(MTMVRefreshContext context,
+ RefreshRequest request, ConnectContext ctx,
List<RefreshAttemptType> attempts)
throws JobException, AnalysisException {
if (!mtmv.isIvm() || request.refreshMode == RefreshMode.COMPLETE
|| !mtmv.getIvmInfo().isBaselineRebuildRequired()) {
- return false;
+ return;
}
ivmFallbackReason = IvmFailureReason.BINLOG_BROKEN.name();
IvmInfo ivmInfo = mtmv.getIvmInfo();
+ // A lone COMPLETE attempt rebuilds every partition anyway, so a
partial pre-rebuild here
+ // would be redundant; it also releases the barrier by itself once it
succeeds.
+ if (attempts.size() == 1 && attempts.get(0) ==
RefreshAttemptType.COMPLETE) {
+ LOG.info("IVM baseline barrier is covered by the pending COMPLETE
attempt, mv={}, taskId={}",
+ mtmv.getName(), getTaskId());
+ return;
+ }
if (ivmInfo.requiresCompleteBaselineRebuild()) {
- executeCompleteAttempt(context, ctx);
- return true;
+ LOG.warn("IVM baseline requires a complete rebuild, mv={},
taskId={}. "
+ + "Continuing with COMPLETE refresh.", mtmv.getName(),
getTaskId());
+ attempts.clear();
+ attempts.add(RefreshAttemptType.COMPLETE);
+ return;
}
- this.needRefreshPartitions = Lists.newArrayList(Sets.intersection(
+ List<String> baselinePartitions = Lists.newArrayList(Sets.intersection(
ivmInfo.getPendingBaselineRebuildPartitions(),
mtmv.getPartitionNames()));
- this.needRefreshPartitions.sort(String::compareTo);
- this.refreshMode = generateRefreshMode(needRefreshPartitions);
- if (refreshMode == MTMVTaskRefreshMode.NOT_REFRESH) {
- return true;
+ if (baselinePartitions.isEmpty()) {
+ // Partition sync has already dropped every partition the barrier
named, so there is
+ // nothing left to rebuild. The surviving partitions are picked up
by the attempts below.
+ LOG.info("IVM baseline partitions were removed by partition sync,
mv={}, taskId={}",
+ mtmv.getName(), getTaskId());
+ } else {
+ baselinePartitions.sort(String::compareTo);
+ this.needRefreshPartitions = baselinePartitions;
+ this.refreshMode = generateRefreshMode(baselinePartitions);
+ executePartitionBasedRefresh(context, RefreshMode.PARTITIONS, ctx);
}
- executePartitionBasedRefresh(context, RefreshMode.PARTITIONS, ctx);
- return true;
+ mtmv.releaseIvmBaselineRebuild(mtmvSchemaChangeVersion);
}
private void validateIvmBaselineBeforePartitionSync(RefreshRequest
request) throws JobException {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelationManager.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelationManager.java
index dd2806fd88b..19266feb1f9 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelationManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelationManager.java
@@ -86,16 +86,16 @@ public class MTMVRelationManager implements MTMVHookService
{
}
public void markIvmBaselineRebuild(BaseTableInfo baseTableInfo, String
reason) {
- markIvmBaselineRebuild(baseTableInfo, Collections.emptyMap(), reason);
+ markIvmBaselineRebuild(baseTableInfo, true, Collections.emptyMap(),
reason);
}
public void markIvmBaselineRebuildForPartitionChange(BaseTableInfo
baseTableInfo,
Map<String, Long> changedPartitions, String reason) {
Preconditions.checkArgument(!changedPartitions.isEmpty(), "changed
partitions can not be empty");
- markIvmBaselineRebuild(baseTableInfo, changedPartitions, reason);
+ markIvmBaselineRebuild(baseTableInfo, false, changedPartitions,
reason);
}
- private void markIvmBaselineRebuild(BaseTableInfo baseTableInfo,
+ private void markIvmBaselineRebuild(BaseTableInfo baseTableInfo, boolean
allPartitionsChanged,
Map<String, Long> changedPartitions, String reason) {
TableNameInfo baseTableName = new
TableNameInfo(baseTableInfo.getCtlName(),
baseTableInfo.getDbName(), baseTableInfo.getTableName());
@@ -115,7 +115,7 @@ public class MTMVRelationManager implements MTMVHookService
{
if
(MTMVPartitionUtil.isTableExcluded(mtmv.getExcludedTriggerTables(),
baseTableName)) {
continue;
}
- if (changedPartitions.isEmpty()) {
+ if (allPartitionsChanged) {
mtmv.invalidateIvmBaseline();
} else {
mtmv.invalidateIvmBaseline(baseTableInfo, changedPartitions);
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 c01fbe16aa5..0f4d44634ca 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
@@ -816,12 +816,46 @@ public class MTMVTaskTest {
Object request = Deencapsulation.invoke(task, "resolveRefreshRequest");
Deencapsulation.invoke(task, "validateIvmBaselineBeforePartitionSync",
request);
- Assertions.assertTrue((Boolean) Deencapsulation.invoke(task,
"handlePendingIvmBaselineRebuild",
- Mockito.mock(MTMVRefreshContext.class), request, new
ConnectContext()));
- Assertions.assertEquals(MTMVTask.MTMVTaskRefreshMode.NOT_REFRESH,
- Deencapsulation.getField(task, "refreshMode"));
+ List<Object> attempts = Lists.newArrayList();
+ attempts.addAll(Deencapsulation.invoke(task, "buildAttempts", request,
false));
+ Assertions.assertEquals("[PARTITIONS, COMPLETE]", attempts.toString());
+
+ Deencapsulation.invoke(task, "handlePendingIvmBaselineRebuild",
+ Mockito.mock(MTMVRefreshContext.class), request, new
ConnectContext(), attempts);
+
+ // A pending COMPLETE rebuild reshapes the attempt list instead of
rebuilding inline, so
+ // PARTITIONS FALLBACK rebuilds the whole MV through the COMPLETE
attempt it keeps.
+ Assertions.assertEquals("[COMPLETE]", attempts.toString());
+ Assertions.assertEquals(IvmFailureReason.BINLOG_BROKEN.name(),
+ Deencapsulation.getField(task, "ivmFallbackReason"));
+ // The barrier is released by the caller once the reshaped attempts
have run.
+ Mockito.verify(mtmv,
Mockito.never()).releaseIvmBaselineRebuild(Mockito.anyLong());
+ }
+
+ @Test
+ public void testDroppedBaselinePartitionsReleaseBarrierWithoutRebuild()
throws Exception {
+ Mockito.when(mtmv.isIvm()).thenReturn(true);
+ IvmInfo ivmInfo = new IvmInfo();
+ ivmInfo.addPendingBaselineRebuildPartitions(Sets.newHashSet(poneName));
+ Mockito.when(mtmv.getIvmInfo()).thenReturn(ivmInfo);
+ // Partition sync already dropped the partition the barrier named, so
nothing is left to
+ // pre-rebuild and the surviving partitions catch up through the
attempts themselves.
+
Mockito.when(mtmv.getPartitionNames()).thenReturn(Sets.newHashSet(ptwoName));
+ MTMVTask task = new MTMVTask(mtmv, relation, MTMVTaskContext.of(
+ MTMVTaskTriggerMode.MANUAL, null, RefreshMode.PARTITIONS,
true, null));
+ Deencapsulation.setField(task, "mtmvSchemaChangeVersion", 7L);
+ Object request = Deencapsulation.invoke(task, "resolveRefreshRequest");
+
+ List<Object> attempts = Lists.newArrayList();
+ attempts.addAll(Deencapsulation.invoke(task, "buildAttempts", request,
false));
+ Deencapsulation.invoke(task, "handlePendingIvmBaselineRebuild",
+ Mockito.mock(MTMVRefreshContext.class), request, new
ConnectContext(), attempts);
+
+ Assertions.assertEquals("[PARTITIONS, COMPLETE]", attempts.toString());
+ Assertions.assertNull(Deencapsulation.getField(task, "refreshMode"));
Assertions.assertEquals(IvmFailureReason.BINLOG_BROKEN.name(),
Deencapsulation.getField(task, "ivmFallbackReason"));
+ Mockito.verify(mtmv).releaseIvmBaselineRebuild(7L);
}
@Test
diff --git
a/regression-test/data/mtmv_p0/ivm/test_ivm_partition_baseline_rebuild_dup_keys.out
b/regression-test/data/mtmv_p0/ivm/test_ivm_partition_baseline_rebuild_dup_keys.out
new file mode 100644
index 00000000000..271a5f56d8e
--- /dev/null
+++
b/regression-test/data/mtmv_p0/ivm/test_ivm_partition_baseline_rebuild_dup_keys.out
@@ -0,0 +1,32 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !baseline_task --
+SUCCESS NONE NONE
+
+-- !baseline_base --
+2026-01-10 1 10
+2026-01-10 1 10
+2026-02-10 3 30
+2026-02-10 3 30
+
+-- !baseline_mv --
+2026-01-10 1 10
+2026-01-10 1 10
+2026-02-10 3 30
+2026-02-10 3 30
+
+-- !strict_task --
+FAILED NOT_REFRESH BINLOG_BROKEN
+
+-- !fallback_task --
+SUCCESS PARTIAL BINLOG_BROKEN
+
+-- !fallback_base --
+2026-02-10 3 30
+2026-02-10 3 30
+2026-02-15 4 40
+
+-- !fallback_mv --
+2026-02-10 3 30
+2026-02-10 3 30
+2026-02-15 4 40
+
diff --git
a/regression-test/data/mtmv_p0/ivm/test_ivm_partition_drop_live_delta.out
b/regression-test/data/mtmv_p0/ivm/test_ivm_partition_drop_live_delta.out
new file mode 100644
index 00000000000..fab4869556a
--- /dev/null
+++ b/regression-test/data/mtmv_p0/ivm/test_ivm_partition_drop_live_delta.out
@@ -0,0 +1,45 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !baseline_task --
+SUCCESS NONE
+
+-- !baseline_base --
+2026-01-10 1 10
+2026-02-10 2 20
+2026-03-10 3 30
+
+-- !baseline_mv --
+2026-01-10 1 10
+2026-02-10 2 20
+2026-03-10 3 30
+
+-- !strict_task --
+FAILED BINLOG_BROKEN
+
+-- !fallback_task --
+SUCCESS BINLOG_BROKEN
+
+-- !fallback_base --
+2026-02-10 2 20
+2026-02-15 4 40
+2026-03-10 3 30
+
+-- !fallback_mv --
+2026-02-10 2 20
+2026-02-15 4 40
+2026-03-10 3 30
+
+-- !resumed_task --
+SUCCESS NONE
+
+-- !resumed_base --
+2026-02-10 2 20
+2026-02-15 4 40
+2026-03-10 3 30
+2026-03-15 5 50
+
+-- !resumed_mv --
+2026-02-10 2 20
+2026-02-15 4 40
+2026-03-10 3 30
+2026-03-15 5 50
+
diff --git
a/regression-test/suites/mtmv_p0/ivm/test_ivm_partition_baseline_rebuild_dup_keys.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_partition_baseline_rebuild_dup_keys.groovy
new file mode 100644
index 00000000000..8f849196d47
--- /dev/null
+++
b/regression-test/suites/mtmv_p0/ivm/test_ivm_partition_baseline_rebuild_dup_keys.groovy
@@ -0,0 +1,123 @@
+// 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
+
+/**
+ * Same baseline-rebuild pre-step as test_ivm_partition_drop_live_delta, but
on the branch where the
+ * affected MV partition SURVIVES: TRUNCATE keeps the partition range, so
partition sync leaves the
+ * MV partition in place and the pre-step really has something to rebuild.
+ *
+ * <p>Two things are pinned here. The refreshed partition is picked up again
by the IVM attempt that
+ * follows, which may only apply the remaining delta -- on a duplicate-key MV
a double apply shows up
+ * as extra copies of the same row, not as a wrong value. And the row written
to the surviving
+ * partition after the truncate must still be consumed. Both are checked by
comparing whole result
+ * sets, so row multiplicities are part of the expectation.
+ */
+suite("test_ivm_partition_baseline_rebuild_dup_keys") {
+ def tableName = "ivm_part_dup_t"
+ def mvName = "ivm_part_dup_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 RefreshMode / IvmFallbackReason come 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.
+ def taskQuery = { String taskId ->
+ """
+ SELECT Status,
+ CASE WHEN RefreshMode IN ('COMPLETE', 'PARTIAL',
'NOT_REFRESH')
+ THEN RefreshMode ELSE 'NONE' END,
+ CASE WHEN IvmFallbackReason = 'BINLOG_BROKEN'
+ THEN IvmFallbackReason ELSE 'NONE' END
+ FROM tasks('type'='mv')
+ WHERE TaskId = '${taskId}'
+ """
+ }
+
+ sql """DROP MATERIALIZED VIEW IF EXISTS ${mvName}"""
+ sql """DROP TABLE IF EXISTS ${tableName}"""
+ sql """
+ CREATE TABLE ${tableName} (
+ dt DATE NOT NULL,
+ id INT NOT NULL,
+ v INT
+ )
+ DUPLICATE KEY(dt, id)
+ PARTITION BY RANGE(dt) ()
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW"
+ )
+ """
+ sql """ALTER TABLE ${tableName} ADD PARTITION p202601 VALUES
[('2026-01-01'), ('2026-02-01'))"""
+ sql """ALTER TABLE ${tableName} ADD PARTITION p202602 VALUES
[('2026-02-01'), ('2026-03-01'))"""
+ // Repeated identical rows: a double-applied delta grows the multiplicity
instead of hiding in a
+ // unique key.
+ sql """INSERT INTO ${tableName} VALUES
+ ('2026-01-10', 1, 10), ('2026-01-10', 1, 10),
+ ('2026-02-10', 3, 30), ('2026-02-10', 3, 30)"""
+
+ sql """
+ CREATE MATERIALIZED VIEW ${mvName}
+ BUILD DEFERRED REFRESH INCREMENTAL FALLBACK ON MANUAL
+ PARTITION BY(dt)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ AS SELECT dt, id, v FROM ${tableName}
+ """
+
+ sql """REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL"""
+ def taskId = waitForNewTask(null)
+ qt_baseline_task taskQuery(taskId)
+ order_qt_baseline_base """SELECT dt, id, v FROM ${tableName} ORDER BY dt,
id, v"""
+ order_qt_baseline_mv """SELECT dt, id, v FROM ${mvName} ORDER BY dt, id,
v"""
+
+ // TRUNCATE replaces the partition, so the MV partition of that range
stays alive and the
+ // baseline pre-step has a real partition to rebuild.
+ sql """TRUNCATE TABLE ${tableName} PARTITION(p202601)"""
+ sql """INSERT INTO ${tableName} VALUES ('2026-02-15', 4, 40)"""
+
+ sql """REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL"""
+ taskId = waitForNewTask(taskId)
+ qt_strict_task taskQuery(taskId)
+
+ sql """REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL FALLBACK"""
+ taskId = waitForNewTask(taskId)
+ qt_fallback_task taskQuery(taskId)
+ order_qt_fallback_base """SELECT dt, id, v FROM ${tableName} ORDER BY dt,
id, v"""
+ order_qt_fallback_mv """SELECT dt, id, v FROM ${mvName} ORDER BY dt, id,
v"""
+}
diff --git
a/regression-test/suites/mtmv_p0/ivm/test_ivm_partition_drop_live_delta.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_partition_drop_live_delta.groovy
new file mode 100644
index 00000000000..bc9f182e306
--- /dev/null
+++
b/regression-test/suites/mtmv_p0/ivm/test_ivm_partition_drop_live_delta.groovy
@@ -0,0 +1,134 @@
+// 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
+
+/**
+ * Dropping a base-table partition invalidates the IVM baseline, because the
rows disappear through
+ * metadata rather than through row binlog entries. The MV partition built
from that base partition
+ * is then removed by partition sync, which is exactly what the baseline
barrier recorded.
+ *
+ * <p>The refresh must still consume the delta that accumulated on the
*surviving* partitions: it
+ * may not report SUCCESS while leaving those partitions stale. This case
inserts a row into a
+ * surviving partition after the drop, so an EMPTY baseline-rebuild
intersection cannot be mistaken
+ * for "nothing to do".
+ *
+ * <p>Partitions are managed by hand (no dynamic partition scheduler) and
every dt is a literal, so
+ * the case is fully deterministic.
+ */
+suite("test_ivm_partition_drop_live_delta") {
+ def tableName = "ivm_part_drop_t"
+ def mvName = "ivm_part_drop_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()
+ }
+
+ // An unset IvmFallbackReason 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.
+ def taskQuery = { String taskId ->
+ """
+ SELECT Status,
+ CASE WHEN IvmFallbackReason = 'BINLOG_BROKEN' THEN
IvmFallbackReason ELSE 'NONE' END
+ FROM tasks('type'='mv')
+ WHERE TaskId = '${taskId}'
+ """
+ }
+
+ sql """DROP MATERIALIZED VIEW IF EXISTS ${mvName}"""
+ sql """DROP TABLE IF EXISTS ${tableName}"""
+ sql """
+ CREATE TABLE ${tableName} (
+ dt DATE NOT NULL,
+ id INT NOT NULL,
+ v INT
+ )
+ UNIQUE KEY(dt, id)
+ PARTITION BY RANGE(dt) ()
+ DISTRIBUTED BY HASH(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 ${tableName} ADD PARTITION p202601 VALUES
[('2026-01-01'), ('2026-02-01'))"""
+ sql """ALTER TABLE ${tableName} ADD PARTITION p202602 VALUES
[('2026-02-01'), ('2026-03-01'))"""
+ sql """ALTER TABLE ${tableName} ADD PARTITION p202603 VALUES
[('2026-03-01'), ('2026-04-01'))"""
+ sql """ALTER TABLE ${tableName} ADD PARTITION p202604 VALUES
[('2026-04-01'), ('2026-05-01'))"""
+ sql """INSERT INTO ${tableName} VALUES
+ ('2026-01-10', 1, 10), ('2026-02-10', 2, 20), ('2026-03-10', 3,
30)"""
+
+ sql """
+ CREATE MATERIALIZED VIEW ${mvName}
+ BUILD DEFERRED REFRESH INCREMENTAL FALLBACK ON MANUAL
+ KEY(dt, id)
+ PARTITION BY(dt)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ AS SELECT dt, id, v FROM ${tableName}
+ """
+
+ sql """REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL"""
+ def taskId = waitForNewTask(null)
+ qt_baseline_task taskQuery(taskId)
+ order_qt_baseline_base """SELECT dt, id, v FROM ${tableName} ORDER BY dt,
id"""
+ order_qt_baseline_mv """SELECT dt, id, v FROM ${mvName} ORDER BY dt, id"""
+
+ sql """ALTER TABLE ${tableName} DROP PARTITION p202601"""
+ sql """INSERT INTO ${tableName} VALUES ('2026-02-15', 4, 40)"""
+
+ // A strict incremental refresh must refuse to run against a broken
baseline.
+ sql """REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL"""
+ taskId = waitForNewTask(taskId)
+ qt_strict_task taskQuery(taskId)
+
+ // The fallback reports SUCCESS, so the MV has to match the base table
afterwards: the expired
+ // partition is gone AND the row written to the surviving partition has
been consumed. An MV
+ // that is missing that row means the refresh silently skipped the
surviving partitions' delta.
+ sql """REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL FALLBACK"""
+ taskId = waitForNewTask(taskId)
+ qt_fallback_task taskQuery(taskId)
+ order_qt_fallback_base """SELECT dt, id, v FROM ${tableName} ORDER BY dt,
id"""
+ order_qt_fallback_mv """SELECT dt, id, v FROM ${mvName} ORDER BY dt, id"""
+
+ // A following strict incremental refresh must be able to continue from
the repaired baseline.
+ sql """INSERT INTO ${tableName} VALUES ('2026-03-15', 5, 50)"""
+ sql """REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL"""
+ taskId = waitForNewTask(taskId)
+ qt_resumed_task taskQuery(taskId)
+ order_qt_resumed_base """SELECT dt, id, v FROM ${tableName} ORDER BY dt,
id"""
+ order_qt_resumed_mv """SELECT dt, id, v FROM ${mvName} ORDER BY dt, id"""
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
