From e8984af966b22fbe17a646489b1fa2445c5cb997 Mon Sep 17 00:00:00 2001 From: yujun Date: Thu, 10 Sep 2026 18:22:22 +0800 Subject: [PATCH 1/2] [fix](ivm) Refresh the surviving partitions after an IVM baseline rebuild Dropping a base-table partition marks the IVM baseline as broken, and partition sync then removes the very MV partitions that barrier named. handlePendingIvmBaselineRebuild derived its rebuild set from the intersection of the barrier with the current MV partitions, so that set was always empty: the task reported SUCCESS with refresh mode NOT_REFRESH and cleared the barrier without refreshing anything, leaving the delta accumulated on the surviving partitions unapplied while the MV claimed to be up to date. Treat the pending baseline rebuild as a pre-step instead of a terminal branch, so the attempt list still runs and the surviving partitions catch up in the same task. Key changes: - MTMVTask.handlePendingIvmBaselineRebuild rebuilds only the barrier partitions that still exist, releases the barrier and then lets the normal attempts run instead of returning early - the attempt list is rewritten in place: a lone COMPLETE attempt needs no pre-rebuild, and a complete baseline rebuild rewrites the list to COMPLETE rather than executing it inline - add MTMV.releaseIvmBaselineRebuild, which compare-and-clears on schemaChangeVersion and journals the new state immediately like the other ivmInfo mutations, so the IVM attempt that follows is not rejected by validateIvmRefreshStart - MTMVRelationManager.markIvmBaselineRebuild takes an explicit all-partitions-changed flag instead of inferring it from an empty partition map Unit Test: - test_ivm_partition_drop_live_delta: new, partitions added and dropped by hand with literal dates, asserts the MV matches the base table right after INCREMENTAL FALLBACK - test_ivm_partition_baseline_rebuild_dup_keys: new, covers the TRUNCATE path where the affected MV partition survives, on a duplicate-key MV so a double-applied delta would show up as extra rows - 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 --- .../java/org/apache/doris/catalog/MTMV.java | 35 +++++ .../doris/job/extensions/mtmv/MTMVTask.java | 56 ++++++-- .../doris/mtmv/MTMVRelationManager.java | 8 +- ...vm_partition_baseline_rebuild_dup_keys.out | 32 +++++ .../test_ivm_partition_drop_live_delta.out | 45 ++++++ ...partition_baseline_rebuild_dup_keys.groovy | 123 ++++++++++++++++ .../test_ivm_partition_drop_live_delta.groovy | 134 ++++++++++++++++++ 7 files changed, 414 insertions(+), 19 deletions(-) create mode 100644 regression-test/data/mtmv_p0/ivm/test_ivm_partition_baseline_rebuild_dup_keys.out create mode 100644 regression-test/data/mtmv_p0/ivm/test_ivm_partition_drop_live_delta.out create mode 100644 regression-test/suites/mtmv_p0/ivm/test_ivm_partition_baseline_rebuild_dup_keys.groovy create mode 100644 regression-test/suites/mtmv_p0/ivm/test_ivm_partition_drop_live_delta.groovy 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/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..169784c42bd74c --- /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", "nonConcurrent") { + 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..48d46256a4d626 --- /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", "nonConcurrent") { + 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""" +} From 6ea4990f69b2a4ac98a5e839adc5e58694b56846 Mon Sep 17 00:00:00 2001 From: yujun Date: Thu, 10 Sep 2026 19:49:57 +0800 Subject: [PATCH 2/2] fix --- .../org/apache/doris/mtmv/MTMVTaskTest.java | 42 +++++++++++++++++-- ...partition_baseline_rebuild_dup_keys.groovy | 2 +- .../test_ivm_partition_drop_live_delta.groovy | 2 +- 3 files changed, 40 insertions(+), 6 deletions(-) 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/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 index 169784c42bd74c..8f849196d4767a 100644 --- 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 @@ -30,7 +30,7 @@ import static java.util.concurrent.TimeUnit.SECONDS * 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", "nonConcurrent") { +suite("test_ivm_partition_baseline_rebuild_dup_keys") { def tableName = "ivm_part_dup_t" def mvName = "ivm_part_dup_mv" 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 index 48d46256a4d626..bc9f182e306685 100644 --- 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 @@ -32,7 +32,7 @@ import static java.util.concurrent.TimeUnit.SECONDS *

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", "nonConcurrent") { +suite("test_ivm_partition_drop_live_delta") { def tableName = "ivm_part_drop_t" def mvName = "ivm_part_drop_mv"