github-actions[bot] commented on code in PR #68180:
URL: https://github.com/apache/doris/pull/68180#discussion_r4059069137
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java:
##########
@@ -616,33 +622,199 @@ public void invalidateIvmBaseline() {
editLogItem.await();
}
- public void invalidateIvmBaseline(BaseTableInfo baseTableInfo, Map<String,
Long> changedPartitions) {
+ /**
+ * Mark the MV partitions that may hold rows read from the changed base
table partitions as needing a
+ * rebuild. When those partitions cannot be determined, the whole MV is
marked instead.
+ */
+ /**
+ * @return whether a barrier was recorded. The caller reports the two
outcomes differently: a change
+ * that no MV partition reads leaves nothing to rebuild and must
not be logged as one.
+ */
+ public boolean invalidateIvmBaseline(BaseTableInfo baseTableInfo,
Map<String, Long> changedPartitions) {
+ // Computed before the MV lock is taken, not inside it: the mapping
reads the partition items of the
+ // MV and of every PCT table, so it takes those tables' locks, and the
MV lock has to stay a leaf
+ // (nothing may be acquired under it) the way the rest of this class
assumes. The selection does not
+ // need to be atomic with the barrier it produces: the barrier is
recorded under the lock below, and
+ // the names it carries are intersected with the live partition names
when they are consumed
+ // (MTMVTask).
+ Optional<Set<String>> affectedMvPartitions =
selectAffectedMvPartitions(baseTableInfo,
+ changedPartitions);
+ if (affectedMvPartitions.isPresent() &&
affectedMvPartitions.get().isEmpty()) {
+ // No MV partition reads any of the changed base partitions, so
this change cannot leave
+ // anything behind here: there is no barrier to persist, and
skipping the version bump
+ // keeps it from discarding the result of a task that is already
running.
+ LOG.debug("No MV partition is affected by changed base partitions,
mv={}, baseTable={}, "
+ + "changedPartitions={}", name, baseTableInfo,
changedPartitions);
+ return false;
+ }
EditLogItem editLogItem;
writeMvLock();
try {
if (ivmInfo == null) {
ivmInfo = new IvmInfo();
}
- if (mvPartitionInfo.getPartitionType() !=
MTMVPartitionType.SELF_MANAGE
- && mvPartitionInfo.getPctInfos().stream()
- .anyMatch(pctInfo ->
pctInfo.getTableInfo().equals(baseTableInfo))) {
- Optional<Set<String>> mvPartitionNames =
refreshSnapshot.getMvPartitionNames(baseTableInfo,
- changedPartitions);
- if (mvPartitionNames.isPresent()) {
-
ivmInfo.addPendingBaselineRebuildPartitions(mvPartitionNames.get());
- } else {
- // Without a snapshot for every changed base partition, a
PARTITIONS rebuild is unsafe.
- ivmInfo.requireCompleteBaselineRebuild();
- }
- } else {
+ if (!affectedMvPartitions.isPresent()) {
+ // A narrower rebuild could leave a partition holding rows of
the changed base partition
+ // untouched, and those rows cannot be repaired later: the
change emitted no row binlog.
ivmInfo.requireCompleteBaselineRebuild();
+ } else {
+
ivmInfo.addPendingBaselineRebuildPartitions(affectedMvPartitions.get());
}
schemaChangeVersion++;
editLogItem = submitIvmInfoChange();
} finally {
writeMvUnlock();
}
editLogItem.await();
+ return true;
+ }
+
+ /**
+ * Select the MV partitions that may hold rows read from the changed base
table partitions.
+ *
+ * <p>This asks which MV partitions read the changed base partitions at
all, instead of (as the
+ * refresh snapshot based selection did) which of them had already seen
them. The snapshot is a lower
+ * bound that is allowed to lag: a base partition that was added after the
snapshot was captured never
+ * appears in it, so it can report "this partition never read the changed
base partition" about a
+ * partition that does hold its rows. Missing a partition here is not
repaired by a later refresh --
+ * dropping or truncating a base partition emits no row binlog, so the
incremental path never learns
+ * about those orphan rows and they stay in the MV forever.
+ *
+ * <p>Three cases have no answer in the mapping, and each of them must
rebuild the whole MV instead:
+ * a SELF_MANAGE MV (the mapping API answers nothing for it, although its
single partition reads every
+ * base partition); a base table that is not one of the MV's PCT tables
(the mapping is seeded from
+ * {@code getPctTables()} and never gains a table later, so a joined
partition table that the MV's
+ * partition column does not reach is not described at all); and a changed
partition that is not in
+ * the base table's metadata right now, which is how RECOVER PARTITION
arrives here -- it marks before
+ * the partition is added back, so at this point the partition is still in
the recycle bin.
+ *
+ * <p>Locking is the fourth way to end up rebuilding everything, but it is
contention rather than a
+ * property of the MV: the tables whose partition items the mapping reads
are locked with a bounded
+ * tryLock, and the MV is rebuilt only while one of them is being written.
See the comment at that
+ * loop.
+ *
+ * <p>An empty result is meaningful, on the other hand: the mapping lists
every base partition read by
+ * the MV, so a changed base partition that no MV partition maps to is
read by none of them.
+ *
+ * @param changedBasePartitions base partition name to partition id, never
empty
+ * @return {@link Optional#empty()} when the affected MV partitions cannot
be determined, otherwise the
+ * (possibly empty) set of MV partition names that must be rebuilt
+ */
+ private Optional<Set<String>> selectAffectedMvPartitions(BaseTableInfo
baseTableInfo,
+ Map<String, Long> changedBasePartitions) {
+ if (mvPartitionInfo.getPartitionType() ==
MTMVPartitionType.SELF_MANAGE) {
+ return Optional.empty();
+ }
+ MTMVRelatedTableIf pctTable = findPctTable(baseTableInfo);
+ if (pctTable == null) {
+ return Optional.empty();
+ }
+ // Computing the mapping reads the partition items of the MV and of
every PCT table, which means
+ // taking their read locks. The caller already holds the changed
table's write lock (a partition DDL
+ // marks before it releases it), so these other reads must not block:
two partition DDLs on two PCT
+ // tables of this MV would otherwise each hold the write lock the
other one needs, and acquiring in
+ // id order cannot break a cycle whose first lock is already held.
They are taken with a bounded
+ // tryLock instead, the way the stream cleanup treats a busy table: a
busy table means a writer is
+ // involved, and then the whole MV is rebuilt. The list is still
sorted by id so that the acquisition
+ // order matches the rest of the code base.
+ List<TableIf> tablesToRead =
Lists.newArrayListWithCapacity(mvPartitionInfo.getPctInfos().size() + 1);
+ tablesToRead.add(this);
+ for (BaseColInfo pctInfo : mvPartitionInfo.getPctInfos()) {
+ if (pctInfo.getTableInfo().equals(baseTableInfo)) {
+ continue;
+ }
+ try {
+ tablesToRead.add(MTMVUtil.getTable(pctInfo.getTableInfo()));
+ } catch (Exception e) {
+ LOG.warn("Failed to resolve PCT table {}, rebuild the whole
MV. mv={}",
+ pctInfo.getTableInfo(), name, e);
+ return Optional.empty();
+ }
+ }
+ tablesToRead.sort(Comparator.comparing(TableIf::getId));
+ if (!MetaLockUtils.tryReadLockTables(tablesToRead,
Table.TRY_LOCK_TIMEOUT_MS, TimeUnit.MILLISECONDS)) {
+ LOG.warn("A PCT table is busy, rebuild the whole MV {} instead of
selecting part of it", name);
+ return Optional.empty();
+ }
+ try {
+ // A partition that is missing from the metadata here is invisible
to the mapping as well, so
+ // an empty answer below would be indistinguishable from "no MV
partition reads it". The match
+ // is deliberately exact: the mapping is keyed by the metadata's
spelling, so a name that only
+ // differs in case must take the whole-MV path too, or the lookup
below would quietly select
+ // nothing for a partition that some MV partition does read. Base
tables that do not implement
+ // getPartitionNames -- the external ones -- report no partition
at all, so a partition change
+ // on them always rebuilds the whole MV. That matches what the
refresh-snapshot selection
+ // answered for them, and the mapping has never been exercised for
external tables (IVM does
+ // not support them as base tables yet): revisit before taking the
narrow path for them.
+ if
(!pctTable.getPartitionNames().containsAll(changedBasePartitions.keySet())) {
+ return Optional.empty();
+ }
+ // By lineage, not by the partition_sync_limit window: the window
can move after a base
+ // partition was read, and a partition it does not cover right now
can still have its rows in
+ // an MV partition. Partition sync keeps that MV partition once
the window covers the base
+ // partition again, and the change itself left no binlog to repair
those rows with.
+ Map<String, Map<MTMVRelatedTableIf, Set<String>>>
partitionMappings =
Review Comment:
[P2] Keep the full lineage calculation out of the lock set. The 100 ms
timeout applies to each sequential acquisition, so a many-PCT MV can already
spend roughly N x 100 ms acquiring locks; after that, this call
copies/transforms every MV/PCT partition (and sorts combined ranges for
multi-PCT MVs) before the `finally` block releases any lock. Because lineage
deliberately ignores `partition_sync_limit`, historical partition count bounds
the work, while DROP/TRUNCATE/REPLACE still holds the changed table's write
lock and writers on the MV/other PCT tables are blocked. Please snapshot all
immutable mapping inputs under the ordered locks and run the generator/reverse
mapping after release, or otherwise bound the protected work.
##########
regression-test/suites/mtmv_p0/ivm/test_ivm_baseline_marker_scope.groovy:
##########
@@ -0,0 +1,179 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import org.awaitility.Awaitility
+
+import static java.util.concurrent.TimeUnit.SECONDS
+
+/**
+ * Which MV partitions a base-table partition change invalidates.
+ *
+ * <p>A partition drop / truncate / replace / recover changes the table
through metadata and emits no row
+ * binlog, so any MV partition that read the dropped rows keeps them forever
unless it is rebuilt. Which
+ * partitions those are is decided by the MV's partition mapping, and the
three cases the mapping cannot
+ * answer -- a joined base table that is not a PCT table, a partition that is
not in the metadata yet
+ * (RECOVER), a SELF_MANAGE MV -- fall back to rebuilding the whole MV.
+ *
+ * <p>Cases pinned here:
+ * <ol>
+ * <li>a partition dropped from a joined table that the MV's partition
column does not reach: the whole
+ * MV is rebuilt, so the rows that joined through the dropped partition
are recomputed (and lose
+ * their dimension value) instead of keeping the stale one;</li>
+ * <li>a base partition that no MV partition reads: nothing is invalidated,
so a strict INCREMENTAL
+ * refresh still starts (under a marker that cannot tell "not read" from
"not in the snapshot",
+ * this left a complete-rebuild barrier behind and the refresh
failed);</li>
+ * <li>the ordinary case: dropping a base partition that one MV partition
reads removes its rows.</li>
+ * </ol>
+ *
+ * <p>All dates are literals and every partition is created by hand: no
current_date() and no dynamic
+ * partition scheduler, so the expectation does not depend on the run date.
+ */
+suite("test_ivm_baseline_marker_scope") {
+ def factTable = "ivm_marker_f"
+ def dimTable = "ivm_marker_d"
+ def mvName = "ivm_marker_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 ${factTable}"""
+ sql """DROP TABLE IF EXISTS ${dimTable}"""
+
+ sql """
+ CREATE TABLE ${factTable} (
+ order_id BIGINT NOT NULL,
+ dt DATE NOT NULL,
+ dimension_id INT,
+ amount INT
+ )
+ UNIQUE KEY(order_id, dt)
+ PARTITION BY RANGE(dt) ()
+ DISTRIBUTED BY HASH(order_id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true"
+ )
+ """
+ sql """ALTER TABLE ${factTable} ADD PARTITION p202601 VALUES
[('2026-01-01'), ('2026-02-01'))"""
+ sql """ALTER TABLE ${factTable} ADD PARTITION p202602 VALUES
[('2026-02-01'), ('2026-03-01'))"""
+
+ // Partitioned as well, but joined on a non-partition column, so it is a
base table of the MV without
+ // being one of its PCT tables.
+ sql """
+ CREATE TABLE ${dimTable} (
+ dimension_id INT NOT NULL,
+ dt DATE NOT NULL,
+ dimension_name VARCHAR(32)
+ )
+ UNIQUE KEY(dimension_id, dt)
+ PARTITION BY RANGE(dt) ()
+ DISTRIBUTED BY HASH(dimension_id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true"
+ )
+ """
+ sql """ALTER TABLE ${dimTable} ADD PARTITION d202601 VALUES
[('2026-01-01'), ('2026-02-01'))"""
+ sql """ALTER TABLE ${dimTable} ADD PARTITION d202602 VALUES
[('2026-02-01'), ('2026-03-01'))"""
+
+ sql """INSERT INTO ${dimTable} VALUES (10, '2026-01-15', 'dim-a'), (20,
'2026-02-15', 'dim-b')"""
+ sql """INSERT INTO ${factTable} VALUES
+ (1, '2026-01-10', 10, 100),
+ (2, '2026-02-10', 20, 200)"""
+
+ sql """
+ CREATE MATERIALIZED VIEW ${mvName}
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ KEY(order_id, dt)
+ PARTITION BY(dt)
+ DISTRIBUTED BY HASH(order_id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ AS SELECT f.order_id, f.dt, f.amount, d.dimension_name
+ FROM ${factTable} f
+ LEFT JOIN ${dimTable} d ON f.dimension_id = d.dimension_id
+ """
+
+ sql """REFRESH MATERIALIZED VIEW ${mvName} COMPLETE"""
+ def taskId = waitForNewTask(null)
+ qt_baseline_task taskQuery(taskId)
+ order_qt_baseline_mv """SELECT order_id, dt, amount, dimension_name
+ FROM ${mvName}"""
+
+ // A partition dropped from the joined table: the MV's partition column
does not reach it, so which
+ // MV partitions read it cannot be determined and all of them are rebuilt.
The row that joined
+ // through d202601 must be recomputed without its dimension value, not
left as it was.
+ sql """ALTER TABLE ${dimTable} DROP PARTITION d202601"""
+ sql """REFRESH MATERIALIZED VIEW ${mvName} AUTO"""
+ taskId = waitForNewTask(taskId)
+ qt_non_pct_drop_task taskQuery(taskId)
+ order_qt_non_pct_drop_mv """SELECT order_id, dt, amount, dimension_name
+ FROM ${mvName}"""
+
+ // Added after the last refresh and dropped before the next one, so no MV
partition reads it. Nothing
+ // is invalidated, and a strict INCREMENTAL refresh still starts: with a
complete-rebuild barrier left
+ // behind it would be rejected with a baseline-rebuild error instead.
+ sql """ALTER TABLE ${factTable} ADD PARTITION p202603 VALUES
[('2026-03-01'), ('2026-04-01'))"""
+ sql """ALTER TABLE ${factTable} DROP PARTITION p202603"""
+ sql """REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL"""
+ taskId = waitForNewTask(taskId)
+ qt_unread_partition_task """
+ SELECT Status FROM tasks('type'='mv') WHERE TaskId = '${taskId}'
+ """
+
+ // The ordinary narrowing: one MV partition reads the dropped base
partition, and its rows go away.
+ sql """ALTER TABLE ${factTable} DROP PARTITION p202601"""
+ sql """REFRESH MATERIALIZED VIEW ${mvName} AUTO"""
+ taskId = waitForNewTask(taskId)
Review Comment:
[P2] Assert that the narrowed refresh itself succeeds. `waitForNewTask`
returns for any terminal state, including FAILED, and partition sync drops the
MV partition for `p202601` before the pending-barrier rebuild or IVM attempt
runs. The February-only row set can therefore match even if this task fails
after sync, leaving the new partial-barrier consumption path untested. Please
query this final `taskId` and pin at least `Status = SUCCESS` (and the intended
refresh/fallback mode if that is part of the contract).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
