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 3050a9ae8ae [fix](ivm) Invalidate the baseline when a column used by
the MV is dropped (#67837)
3050a9ae8ae is described below
commit 3050a9ae8ae53cf14ecda7a785ffd1a3a28f4d0c
Author: yujun <[email protected]>
AuthorDate: Fri Sep 11 22:32:46 2026 +0800
[fix](ivm) Invalidate the baseline when a column used by the MV is dropped
(#67837)
A light schema change on an IVM base table (`DROP COLUMN`, usually
followed by an `ADD COLUMN` with the same name) only changes metadata
and emits no row binlog. When the dropped column is one the MV uses, the
next `REFRESH ... INCREMENTAL` consumes an empty delta and reports
**SUCCESS**, while the MV silently keeps the rows that were computed
under the old column.
Dropping the column did reach the MTMV hook, but that only moved the MV
status to `SCHEMA_CHANGE`. That state merely re-analyses the MV query on
the next refresh - and by then the column is usually back under the same
name, so the analysis succeeds and the incremental refresh proceeds. The
IVM baseline barrier (`requireCompleteBaselineRebuild`) was never raised
on this path: it is only set for `REPLACE TABLE`, `REPLACE PARTITION`
and partition changes.
### What this PR does
Re-analyse the MV query on the `alterTable` path, right after the alter
was applied, and invalidate the IVM baseline when the query no longer
binds.
This needs no lineage machinery: dropping or renaming a column the MV
uses makes the query unanalysable, so the failure itself is the
dependency signal - and dropping a column the MV does not reference
leaves the incremental path untouched.
Scope: on an IVM base table only `DROP COLUMN` and `RENAME TABLE` can
change a referenced column at all (row binlog tables reject `MODIFY
COLUMN`, `RENAME COLUMN` and `REORDER COLUMNS`), so those two operations
are the whole surface. A dropped base table needs no handling - the IVM
stream records the base table id and stops being usable, which already
fails the refresh.
The analysis runs in a context of its own rather than the session that
issued the alter, because the underlying `analyzeQueryWithSql` reuses
and closes the statement context it is handed.
---
.../org/apache/doris/mtmv/MTMVRelationManager.java | 61 ++++++++-
.../doris/alter/SchemaChangeHandlerTest.java | 8 ++
.../doris/mtmv/ivm/IvmBaselineRebuildTest.java | 36 ++++++
...ivm_drop_referenced_column_baseline_rebuild.out | 8 ++
..._drop_referenced_column_baseline_rebuild.groovy | 144 +++++++++++++++++++++
5 files changed, 253 insertions(+), 4 deletions(-)
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 19266feb1f9..3bfa209fd6f 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
@@ -335,7 +335,9 @@ public class MTMVRelationManager implements MTMVHookService
{
*/
@Override
public void dropTable(Table table) {
- processBaseTableChange(new BaseTableInfo(table), "The base table has
been deleted:");
+ // A dropped base table is already caught by the IVM stream guard (the
stream records the
+ // base table id, so it stops being usable once the table is gone), no
need to re-analyze.
+ processBaseTableChange(new BaseTableInfo(table), "The base table has
been deleted:", false);
}
/**
@@ -347,9 +349,56 @@ public class MTMVRelationManager implements
MTMVHookService {
public void alterTable(BaseTableInfo oldTableInfo, Optional<BaseTableInfo>
newTableInfo, boolean isReplace) {
// when replace, need deal two table
if (isReplace) {
- processBaseTableChange(newTableInfo.get(), "The base table has
been updated:");
+ // REPLACE TABLE already invalidates the IVM baseline explicitly,
see Alter#processReplaceTable
+ processBaseTableChange(newTableInfo.get(), "The base table has
been updated:", false);
+ }
+ // A RENAME leaves every column alone, and the failure it does cause
-- the MV query still
+ // spells the old name -- is already reported by the refresh itself
(MTMVTask#run resolves
+ // the base tables from the query before it ever looks at the
baseline). Invalidating here
+ // would only leave a stale flag behind: rename the table back and the
query is analyzable
+ // again, yet every strict INCREMENTAL refresh would stay rejected
until a COMPLETE one ran.
+ boolean renamed = !isReplace && newTableInfo.isPresent()
+ && !Objects.equals(oldTableInfo.getTableName(),
newTableInfo.get().getTableName());
+ processBaseTableChange(oldTableInfo, "The base table has been
updated:", !renamed);
+ }
+
+ /**
+ * An IVM baseline is only valid while the MV query can still be analyzed
against the current
+ * base table schema. Re-analyzing the MV query here (right after the
alter was applied) is what
+ * detects a changed column identity: dropping or renaming a column the MV
uses makes the query
+ * unanalyzable, and a column re-added with the same name is a different
column, so pre-existing
+ * rows read its default value instead.
+ *
+ * <p>Such a change is metadata-only for light schema changes and emits no
binlog, so an
+ * incremental refresh would consume an empty delta and report SUCCESS
while silently keeping the
+ * rows computed under the old column epoch. Invalidating the baseline
makes a strict INCREMENTAL
+ * refresh fail and tell the user to run a COMPLETE refresh instead.
+ *
+ * <p>Only IVM is covered: a plain MTMV keeps its previous behaviour
(status only).
+ */
+ private void invalidateIvmBaselineIfQueryUnusable(BaseTableInfo
baseTableInfo, Table mtmvTable) {
+ if (!(mtmvTable instanceof MTMV) || !((MTMV) mtmvTable).isIvm()) {
+ return;
+ }
+ MTMV mtmv = (MTMV) mtmvTable;
+ // Analyse in a context owned by this check, never the session that
issued the alter: the check
+ // must not disturb the running statement, and it has to work on
threads that have no session.
+ // Setting a thread local is how a context is made current, so restore
the previous one.
+ ConnectContext previousCtx = ConnectContext.get();
+ try {
+ MTMVPlanUtil.ensureMTMVQueryUsable(mtmv,
+ MTMVPlanUtil.createMTMVContext(mtmv,
MTMVPlanUtil.DISABLE_RULES_WHEN_RUN_MTMV_TASK));
+ } catch (Exception e) {
+ LOG.info("Invalidate IVM baseline, the MV query is no longer
usable. baseTable={}, mtmv={}, "
+ + "reason={}", baseTableInfo, mtmv.getName(),
e.getMessage());
+ mtmv.invalidateIvmBaseline();
+ } finally {
+ if (previousCtx != null) {
+ previousCtx.setThreadLocalInfo();
+ } else {
+ ConnectContext.remove();
+ }
}
- processBaseTableChange(oldTableInfo, "The base table has been
updated:");
}
@Override
@@ -411,7 +460,8 @@ public class MTMVRelationManager implements MTMVHookService
{
}
}
- private void processBaseTableChange(BaseTableInfo baseTableInfo, String
msgPrefix) {
+ private void processBaseTableChange(BaseTableInfo baseTableInfo, String
msgPrefix,
+ boolean checkIvmQueryUsable) {
Set<BaseTableInfo> mtmvsByBaseTable =
getMtmvsByBaseTableOneLevelAndFromView(baseTableInfo);
if (CollectionUtils.isEmpty(mtmvsByBaseTable)) {
return;
@@ -424,6 +474,9 @@ public class MTMVRelationManager implements MTMVHookService
{
LOG.warn(e);
continue;
}
+ if (checkIvmQueryUsable) {
+ invalidateIvmBaselineIfQueryUnusable(baseTableInfo, mtmv);
+ }
TableNameInfo tableNameInfo = new
TableNameInfo(mtmv.getQualifiedDbName(),
mtmv.getName());
MTMVStatus status = new MTMVStatus(MTMVState.SCHEMA_CHANGE,
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeHandlerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeHandlerTest.java
index 4b3962b9142..1be3cca764a 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeHandlerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeHandlerTest.java
@@ -422,6 +422,14 @@ public class SchemaChangeHandlerTest extends
TestWithFeService {
createTable(create);
expectException("ALTER TABLE test." + tableName + " MODIFY COLUMN v1
BIGINT", "Table With binlog<row>");
+ // 1b) RENAME COLUMN / REORDER COLUMNS are not allowed on row binlog
tables either.
+ // This matters for IVM: dropping a column an MV references
invalidates the IVM baseline, so
+ // every other way of changing a referenced column has to be rejected
here. If one of them
+ // were ever allowed, it would become a new way to leave an MV stale
without being noticed.
+ expectException("ALTER TABLE test." + tableName + " RENAME COLUMN v1
TO v1_renamed",
+ "Table With binlog<row>");
+ expectException("ALTER TABLE test." + tableName + " ORDER BY (k1,
v1)", "Table With binlog<row>");
+
// 2) VARIANT not supported
String createVariant = "CREATE TABLE test.binlog_variant (k1 INT NOT
NULL, v1 VARIANT) "
+ "UNIQUE KEY(k1) DISTRIBUTED BY HASH(k1) BUCKETS 1 "
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmBaselineRebuildTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmBaselineRebuildTest.java
index 2588ca5f532..dfd3d6dc7ef 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmBaselineRebuildTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmBaselineRebuildTest.java
@@ -104,6 +104,24 @@ public class IvmBaselineRebuildTest extends
TestWithFeService {
Assertions.assertTrue(getMtmv(db).getIvmInfo().isBaselineRebuildRequired());
}
+ @Test
+ public void testDropColumnMarksBaselineRebuildOnlyWhenReferenced() throws
Exception {
+ String db = "ivm_broken_drop_column";
+ createPartitionedIvmTableAndMv(db);
+ MTMV mtmv = getMtmv(db);
+
+ // ivm_mv selects dt, k1, v1. Dropping a column it does not use must
leave the baseline alone.
+ executeSql("ALTER TABLE ivm_base ADD COLUMN spare int");
+ executeSql("ALTER TABLE ivm_base DROP COLUMN spare");
+ Assertions.assertFalse(mtmv.getIvmInfo().isBaselineRebuildRequired());
+
+ // Dropping a column the MV uses makes the MV query unanalyzable: the
change is metadata-only
+ // and emits no binlog, so an incremental refresh would silently keep
the rows of the old
+ // column. The baseline has to be invalidated instead.
+ executeSql("ALTER TABLE ivm_base DROP COLUMN v1");
+ Assertions.assertTrue(mtmv.getIvmInfo().isBaselineRebuildRequired());
+ }
+
@Test
public void testPublishedPctPartitionUsesPartitionsBaselineRebuild()
throws Exception {
String db = "ivm_partitions_baseline_rebuild";
@@ -217,6 +235,24 @@ public class IvmBaselineRebuildTest extends
TestWithFeService {
Assertions.assertFalse(getMtmv(db).getIvmInfo().isBaselineRebuildRequired());
}
+ @Test
+ public void testRenameTableBackKeepsIncrementalRefreshStartable() throws
Exception {
+ String db = "ivm_broken_rename_table_back";
+ createPartitionedIvmTableAndMv(db);
+
+ executeSql("ALTER TABLE ivm_base RENAME ivm_base_renamed");
+ executeSql("ALTER TABLE ivm_base_renamed RENAME ivm_base");
+
+ // A rename changes no column, so it must not invalidate the baseline
in either direction:
+ // once the table is renamed back, the MV query is analyzable again
and a strict INCREMENTAL
+ // refresh has to be able to start. A "baseline rebuild required" flag
left behind by the
+ // rename would reject every one of them until a COMPLETE refresh had
been run, even though
+ // nothing the MV depends on ever changed.
+ MTMV mtmv = getMtmv(db);
+ Assertions.assertFalse(mtmv.getIvmInfo().isBaselineRebuildRequired());
+ Assertions.assertDoesNotThrow(() ->
mtmv.validateIvmRefreshStart(mtmv.getSchemaChangeVersion()));
+ }
+
@Test
public void testReplaceTableMarksBaselineRebuild() throws Exception {
String db = "ivm_broken_replace_table";
diff --git
a/regression-test/data/mtmv_p0/ivm/test_ivm_drop_referenced_column_baseline_rebuild.out
b/regression-test/data/mtmv_p0/ivm/test_ivm_drop_referenced_column_baseline_rebuild.out
new file mode 100644
index 00000000000..8793c2b56bc
--- /dev/null
+++
b/regression-test/data/mtmv_p0/ivm/test_ivm_drop_referenced_column_baseline_rebuild.out
@@ -0,0 +1,8 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !mv_rows_baseline --
+10 2 300
+20 1 300
+
+-- !mv_rows_after_aba --
+0 3 600
+
diff --git
a/regression-test/suites/mtmv_p0/ivm/test_ivm_drop_referenced_column_baseline_rebuild.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_drop_referenced_column_baseline_rebuild.groovy
new file mode 100644
index 00000000000..26dd37ff2d0
--- /dev/null
+++
b/regression-test/suites/mtmv_p0/ivm/test_ivm_drop_referenced_column_baseline_rebuild.groovy
@@ -0,0 +1,144 @@
+// 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 column that an IVM references, then re-adding a column with the
same name
+// (schema ABA), used to let a strict INCREMENTAL refresh report SUCCESS while
silently
+// keeping the rows computed under the old column epoch.
+//
+// The base table change is metadata-only (light schema change) and emits no
binlog, so the
+// delta is empty and the refresh has nothing to apply -- the MV baseline is
simply stale.
+//
+// Expected: dropping a referenced column invalidates the IVM baseline, so a
strict
+// INCREMENTAL refresh is rejected and the user is told to run a COMPLETE
refresh.
+// Dropping an unreferenced column must still leave the incremental path
untouched.
+suite("test_ivm_drop_referenced_column_baseline_rebuild") {
+ def tableName = "ivm_drop_ref_col_t"
+ def mvName = "ivm_drop_ref_col_mv"
+
+ sql """DROP MATERIALIZED VIEW IF EXISTS ${mvName}"""
+ sql """DROP TABLE IF EXISTS ${tableName}"""
+
+ sql """
+ CREATE TABLE ${tableName} (
+ id BIGINT NOT NULL,
+ grp INT NULL,
+ amount BIGINT NULL,
+ spare INT NULL
+ )
+ UNIQUE KEY(id)
+ 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 """INSERT INTO ${tableName} VALUES (1, 10, 100, 7), (2, 10, 200, 8),
(3, 20, 300, 9)"""
+
+ sql """
+ CREATE MATERIALIZED VIEW ${mvName}
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ KEY(grp)
+ DISTRIBUTED BY HASH(grp) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ AS SELECT grp, COUNT(*) AS cnt, SUM(amount) AS total
+ FROM ${tableName} GROUP BY grp
+ """
+
+ def ddlJobCount = { String table ->
+ return sql("""SHOW ALTER TABLE COLUMN WHERE TableName =
'${table}'""").size()
+ }
+
+ // `SHOW ALTER TABLE COLUMN` keeps finished jobs, so wait for a *new* job
that is FINISHED.
+ def waitDdlFinished = { String table, int previousJobCount ->
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
+ def jobs = sql """SHOW ALTER TABLE COLUMN WHERE TableName =
'${table}'"""
+ return jobs.size() > previousJobCount
+ && jobs.every({ row -> row[9].toString() == 'FINISHED' })
+ })
+ }
+
+ def lastTaskId = null
+ // tasks('type'='mv') can briefly miss the just-finished task, so require
a *new* TaskId.
+ def waitTerminalTask = { String mv ->
+ def taskResult
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
+ taskResult = sql_return_maparray("""
+ SELECT TaskId, Status, RefreshMode, IvmFallbackReason, ErrorMsg
+ FROM tasks('type'='mv')
+ WHERE MvDatabaseName = '${context.dbName}' AND MvName = '${mv}'
+ ORDER BY CreateTime DESC, TaskId DESC LIMIT 1
+ """)
+ return !taskResult.isEmpty()
+ && taskResult[0].TaskId.toString() != lastTaskId
+ && taskResult[0].Status.toString() != 'PENDING'
+ && taskResult[0].Status.toString() != 'RUNNING'
+ })
+ lastTaskId = taskResult[0].TaskId.toString()
+ return taskResult[0]
+ }
+
+ // ---------------------------------------------------------------- 1.
baseline
+ sql """REFRESH MATERIALIZED VIEW ${mvName} COMPLETE"""
+ def task = waitTerminalTask(mvName)
+ assertEquals("SUCCESS", task.Status.toString(), "baseline COMPLETE
refresh: " + task.ErrorMsg)
+ order_qt_mv_rows_baseline "SELECT grp, cnt, total FROM ${mvName}"
+
+ // ------------------------------------- 2. unreferenced column: no
baseline invalidation
+ def before = ddlJobCount(tableName)
+ sql """ALTER TABLE ${tableName} DROP COLUMN spare"""
+ waitDdlFinished(tableName, before)
+
+ sql """REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL"""
+ task = waitTerminalTask(mvName)
+ assertEquals("SUCCESS", task.Status.toString(),
+ "dropping an unreferenced column must not invalidate the IVM
baseline: " + task.ErrorMsg)
+
+ // ---------------------------------------- 3. referenced column: strict
INCREMENTAL rejected
+ before = ddlJobCount(tableName)
+ sql """ALTER TABLE ${tableName} DROP COLUMN grp"""
+ waitDdlFinished(tableName, before)
+
+ sql """REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL"""
+ task = waitTerminalTask(mvName)
+ assertEquals("FAILED", task.Status.toString(),
+ "dropping a referenced column must reject a strict INCREMENTAL
refresh")
+
+ // -------------------------------- 4. same-name re-add (schema ABA) is
still rejected
+ before = ddlJobCount(tableName)
+ sql """ALTER TABLE ${tableName} ADD COLUMN grp INT NULL DEFAULT '0'"""
+ waitDdlFinished(tableName, before)
+
+ sql """REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL"""
+ task = waitTerminalTask(mvName)
+ assertEquals("FAILED", task.Status.toString(),
+ "schema ABA must not be silently accepted by a strict INCREMENTAL
refresh")
+ assertTrue(task.ErrorMsg.toString().contains("baseline rebuild is
pending"),
+ "expected a pending baseline rebuild hint, got: " + task.ErrorMsg)
+
+ // ------------------------------------ 5. COMPLETE rebuild reflects
current base semantics
+ // Every pre-existing row now reads the re-added column's default value.
+ sql """REFRESH MATERIALIZED VIEW ${mvName} COMPLETE"""
+ task = waitTerminalTask(mvName)
+ assertEquals("SUCCESS", task.Status.toString(), "COMPLETE rebuild after
ABA: " + task.ErrorMsg)
+ order_qt_mv_rows_after_aba "SELECT grp, cnt, total FROM ${mvName}"
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
