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 354081ef5b6def..b162a2b2d511b7 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 void invalidateIvmBaseline(BaseTableInfo baseTableInfo, Map 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). + * + *

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. + * + *

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 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 d97e142fee91a0..1eab83fe576d3e 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 void run() throws JobException { 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 @@ private void executeCompleteAttempt(MTMVRefreshContext context, ConnectContext c 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. + * + *

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. + * + *

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 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 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 dd2806fd88b3ce..19266feb1f9a2f 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 Set getMtmvsByBaseTableOneLevelAndFromView(BaseTableInfo t } public void markIvmBaselineRebuild(BaseTableInfo baseTableInfo, String reason) { - markIvmBaselineRebuild(baseTableInfo, Collections.emptyMap(), reason); + markIvmBaselineRebuild(baseTableInfo, true, Collections.emptyMap(), reason); } public void markIvmBaselineRebuildForPartitionChange(BaseTableInfo baseTableInfo, Map 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 changedPartitions, String reason) { TableNameInfo baseTableName = new TableNameInfo(baseTableInfo.getCtlName(), baseTableInfo.getDbName(), baseTableInfo.getTableName()); @@ -115,7 +115,7 @@ private void markIvmBaselineRebuild(BaseTableInfo baseTableInfo, 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 c01fbe16aa5599..0f4d44634caeeb 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 void testPartitionsFallbackRebuildsPendingBaselineWithComplete() throws E 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 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 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 00000000000000..271a5f56d8e39b --- /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 00000000000000..fab4869556a1d7 --- /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 00000000000000..8f849196d4767a --- /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. + * + *

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 00000000000000..bc9f182e306685 --- /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. + * + *

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". + * + *

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""" +}