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 d6c1a2b5f26 [fix](ivm) Choose IVM baseline rebuild partitions from the
MV partition mapping (#68180)
d6c1a2b5f26 is described below
commit d6c1a2b5f26334515d14d6751d5ab47b56a6b06e
Author: yujun <[email protected]>
AuthorDate: Tue Sep 22 18:29:19 2026 +0800
[fix](ivm) Choose IVM baseline rebuild partitions from the MV partition
mapping (#68180)
An IVM materialized view has to fully rebuild every MV partition that
may hold rows of a base-table
partition that was dropped / truncated / replaced / recovered. Those
operations change the table
through metadata and emit no row binlog, so the incremental path can
never remove the stale rows: if
such an MV partition is not marked, the orphan rows stay in the MV
forever, with no error anywhere.
Which MV partitions to mark was decided by asking the refresh snapshot
which of them had *seen* the
changed base partition (`MTMVRefreshSnapshot#getMvPartitionNames`). A
snapshot is captured before the
base table is read, so it is only a lower bound: a base partition that
appears in the base table after
the capture is not in it, and each partition's snapshot is written back
by whichever task finishes
last. When at least one other MV partition did match, the lookup
returned that subset and the partition
that had actually read the dropped one was silently left out:
```
partition p1: capture snapshot p1 reads the base table p1
still running
base table: X is added p2 refreshes (its
snapshot has X) X dropped
marker: p1's snapshot has no X, p2's has -> only p2 is marked
result: p1 keeps the rows of X forever
```
This asks the MV partition mapping instead: which MV partitions read the
changed base partition is
exactly the question, it is metadata rather than a lagging record, and
it is the same mapping the
refresh already uses for partition sync.
| case | before | after |
| --- | --- | --- |
| MV partitions that read the changed base partition | marked | marked
(unchanged) |
| changed base partition read by no MV partition | whole MV rebuilt |
nothing marked |
| SELF_MANAGE MV, base table is not a PCT table, `RECOVER PARTITION`, a
PCT table of a multi-PCT MV, mapping computation fails | whole MV
rebuilt | whole MV rebuilt (unchanged) |
The second row is intentional: the old lookup could not tell "no MV
partition reads this partition"
apart from "the snapshot does not know it", so it rebuilt the whole MV
-- a table whose dynamic
partitions are dropped ahead of the MV's partition sync forced a full
rebuild every time. The mapping
tells them apart, and the cases that genuinely cannot be answered from
it are explicit checks
rather than a guess from an empty result:
- `RECOVER PARTITION` marks before the partition is added back to the
table, so the partition is still
in the recycle bin at that moment and no metadata-derived mapping can
describe it.
- The mapping is seeded from the MV's PCT tables and never gains a table
later, so a joined partition
table the MV's partition column does not reach is not described by it at
all.
- A mapping computation failure must not fail the base table DDL: it
warns and rebuilds the whole MV.
- The mapping reads the partition items of the MV and of every PCT
table, so it takes their read locks,
while the caller already holds the changed table's write lock: two
concurrent partition DDLs on two PCT
tables of the same MV would each hold the write lock the other one
needs. Those reads are therefore
taken with a bounded `tryReadLock` (`Table.TRY_LOCK_TIMEOUT_MS`, the
same way the stream cleanup treats
a busy table) and never block: while the other tables are free -- the
normal case -- the selection still
narrows to the MV partitions that read the changed base partition, and
while one of them is being
written the whole MV is rebuilt. Acquiring in id order cannot help here,
because the first lock of the
pair is already held before this point.
The mapping is computed *before* the MV lock is taken, not inside it:
computing it takes the partition
items of the MV and of its PCT tables, and the MV lock is a leaf lock --
nothing is acquired under it
today, and the lock analysis of the baseline barrier depends on that.
The selection does not need to be
atomic with the barrier it produces: the barrier is recorded under the
lock, and the names it carries
are intersected with the live partition names when they are consumed
(`MTMVTask`).
`MTMVRefreshSnapshot#getMvPartitionNames`, the old selection, is removed
with the unit test that
covered it; it has no other caller.
---
.../main/java/org/apache/doris/catalog/MTMV.java | 253 ++++++++++-
.../apache/doris/common/util/MetaLockUtils.java | 20 +
.../org/apache/doris/mtmv/MTMVPartitionUtil.java | 9 +
.../org/apache/doris/mtmv/MTMVPropertyUtil.java | 56 +++
.../org/apache/doris/mtmv/MTMVRefreshSnapshot.java | 24 --
...MTMVRelatedPartitionDescSyncLimitGenerator.java | 2 +-
.../org/apache/doris/mtmv/MTMVRelationManager.java | 15 +-
.../doris/common/util/MetaLockUtilsTest.java | 42 ++
.../test/java/org/apache/doris/mtmv/MTMVTest.java | 21 -
.../doris/mtmv/ivm/IvmBaselineRebuildTest.java | 469 +++++++++++++++++++--
.../mtmv_p0/ivm/test_ivm_baseline_marker_scope.out | 20 +
.../ivm/test_ivm_baseline_marker_scope.groovy | 189 +++++++++
12 files changed, 1025 insertions(+), 95 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 4050426bab6..ec97efb324f 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
@@ -22,6 +22,7 @@ import org.apache.doris.catalog.OlapTableFactory.MTMVParams;
import org.apache.doris.catalog.info.TableNameInfo;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
+import org.apache.doris.common.util.MetaLockUtils;
import org.apache.doris.common.util.PropertyAnalyzer;
import org.apache.doris.datasource.CatalogMgr;
import org.apache.doris.datasource.mvcc.MvccSnapshot;
@@ -29,6 +30,7 @@ import org.apache.doris.datasource.mvcc.MvccTableInfo;
import org.apache.doris.job.common.TaskStatus;
import org.apache.doris.job.exception.JobException;
import org.apache.doris.job.extensions.mtmv.MTMVTask;
+import org.apache.doris.mtmv.BaseColInfo;
import org.apache.doris.mtmv.BaseTableInfo;
import org.apache.doris.mtmv.EnvInfo;
import org.apache.doris.mtmv.MTMVAlterOpType;
@@ -51,6 +53,7 @@ import org.apache.doris.mtmv.MTMVRelatedTableIf;
import org.apache.doris.mtmv.MTMVRelation;
import org.apache.doris.mtmv.MTMVSnapshotIf;
import org.apache.doris.mtmv.MTMVStatus;
+import org.apache.doris.mtmv.MTMVUtil;
import org.apache.doris.mtmv.ivm.IvmInfo;
import org.apache.doris.mtmv.ivm.IvmUtil;
import org.apache.doris.nereids.rules.analysis.SessionVarGuardRewriter;
@@ -71,12 +74,14 @@ import org.apache.logging.log4j.Logger;
import java.io.IOException;
import java.util.Collections;
+import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -379,6 +384,18 @@ public class MTMV extends OlapTable {
Map<TableNameInfo, Integer> oldWindowLimits =
containsPartitionWindowLimit
?
MTMVPropertyUtil.getIvmPartitionWindowLimit(this.mvProperties)
: Maps.newHashMap();
+ // A partition_sync_limit window decides which base partitions the
MV maintains. Only a change
+ // that can bring a partition back into that set needs a complete
baseline rebuild -- a removed
+ // or wider limit -- because its deltas were skipped while it was
outside and nothing
+ // incremental can repair them. That is the same trade as the two
properties around it. A
+ // window that starts applying, a narrower one, and one that
describes the same set as before
+ // leave the applied deltas intact; the partitions they take out
are dropped by partition sync
+ // before the refresh plans, and taking one back in is the
widening this answers. Doing it here,
+ // in the critical section that applies the ALTER, is also what
keeps a window set and cleared
+ // while an invalidation reads the mapping from making that
mapping look unwindowed.
+ boolean containsSyncWindow =
MTMVPropertyUtil.containsPartitionSyncWindow(mvProperties);
+ Map<String, String> oldSyncWindow = containsSyncWindow
+ ?
MTMVPropertyUtil.partitionSyncWindowOf(this.mvProperties) : null;
this.mvProperties.putAll(mvProperties);
// Both excluded_trigger_tables changes and window limit
enlargement/removal
// change the refresh baseline semantics: partitions previously
skipped become
@@ -423,6 +440,11 @@ public class MTMV extends OlapTable {
}
}
}
+ if (containsSyncWindow && ivmInfo != null && ivmInfo.isEnableIvm()
+ &&
MTMVPropertyUtil.partitionSyncWindowWidens(oldSyncWindow,
+
MTMVPropertyUtil.partitionSyncWindowOf(this.mvProperties))) {
+ requireCompleteBaselineRebuild = true;
+ }
if (invalidateRefreshSnapshot || requireCompleteBaselineRebuild) {
this.schemaChangeVersion++;
this.refreshSnapshot = new MTMVRefreshSnapshot();
@@ -682,26 +704,43 @@ public class MTMV extends OlapTable {
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();
@@ -709,6 +748,194 @@ public class MTMV extends OlapTable {
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". It has to
+ // be the partition the caller described, not merely one carrying
the same name: RECOVER
+ // PARTITION reports the recycled partition under its old name,
and a partition added after
+ // the drop may be live under that name again, with a different
range. Matching on the name
+ // alone would accept that replacement, select the MV partitions
of its range, and leave the
+ // recovered range -- whose rows no row binlog can repair --
without a barrier. The lookup is
+ // by exact name, so a name that only differs in case takes the
whole-MV path too. Base tables
+ // that do not implement getPartition -- the external ones --
answer null for every name, 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.
+ for (Entry<String, Long> changedBasePartition :
changedBasePartitions.entrySet()) {
+ Partition livePartition =
pctTable.getPartition(changedBasePartition.getKey());
+ if (livePartition == null || livePartition.getId() !=
changedBasePartition.getValue()) {
+ return Optional.empty();
+ }
+ }
+ // Whether a partition_sync_limit is in effect decides whether the
mapping built below may be
+ // trusted, and it is read on both sides of that construction. It
has to be: the property is
+ // mutable (ALTER MATERIALIZED VIEW ... SET is not generation
guarded) and the mapping is built
+ // from it, so a read taken on one side only can be the stale one.
Reading it after the mapping
+ // alone misses a limit cleared while the mapping was built -- the
mapping is then the windowed
+ // one and would be trusted; reading it before alone misses a
limit set in that same window, for
+ // the opposite reason. The two reads bracket exactly the
construction, and a limit in effect on
+ // either of them means the mapping that came out of it may carry
a window.
+ boolean partitionSyncLimitActiveBeforeMapping =
+ MTMVPartitionUtil.isPartitionSyncLimitActive(mvProperties);
+ Map<String, Map<MTMVRelatedTableIf, Set<String>>>
partitionMappings =
+ calculatePartitionMappings(Maps.newHashMap());
+ boolean partitionSyncLimitActiveAfterMapping =
+ MTMVPartitionUtil.isPartitionSyncLimitActive(mvProperties);
+ Set<String> res = Sets.newHashSet();
+ boolean pctTableMapped = false;
+ // Every base partition this table's part of the mapping
describes, which is what the selection
+ // below is only allowed to trust when it covers the whole change.
+ Set<String> mappedBasePartitions = Sets.newHashSet();
+ for (Entry<String, Map<MTMVRelatedTableIf, Set<String>>> mapping :
partitionMappings.entrySet()) {
+ for (Entry<MTMVRelatedTableIf, Set<String>> tableMapping :
mapping.getValue().entrySet()) {
+ if (!tableMapping.getKey().equals(pctTable)) {
+ continue;
+ }
+ pctTableMapped = true;
+ mappedBasePartitions.addAll(tableMapping.getValue());
+ if (!Collections.disjoint(tableMapping.getValue(),
changedBasePartitions.keySet())) {
+ res.add(mapping.getKey());
+ }
+ }
+ }
+ // The mapping does not describe this base table at all. That
contradicts the PCT check above,
+ // so it is safer to rebuild everything than to trust a selection
that never saw the table --
+ // unless the MV has no partition of its own yet, which is the one
shape where the missing
+ // entries are not a surprise: an MV without partitions holds no
rows.
+ if (!pctTableMapped) {
+ if (getPartitionNames().isEmpty()) {
+ LOG.info("MV has no partition yet, nothing can hold the
changed base partitions. "
+ + "baseTable={}, mv={}", baseTableInfo, name);
+ return Optional.of(Sets.newHashSet());
+ }
+ LOG.warn("Base table is not described by the partition
mapping, rebuild the whole MV. "
+ + "baseTable={}, mv={}", baseTableInfo, name);
+ return Optional.empty();
+ }
+ // A selection is only trustworthy while the mapping describes
every base partition that
+ // changed. With a partition_sync_limit in effect it does not: the
window leaves out the
+ // partitions it dropped, and one of those can still have its rows
in an MV partition --
+ // shrinking the window does not touch the MV's own partitions,
and widening it again makes
+ // partition sync keep them. A name the mapping leaves out cannot
be told apart from a
+ // partition no MV partition reads, so the whole MV is rebuilt
instead. Requiring the whole
+ // change to be described, rather than only a non-empty selection,
is what covers a change
+ // that mixes a partition inside the window with one outside it:
the inside half would
+ // otherwise fill the selection and hide the missing half. Without
a limit the mapping is
+ // complete, and a partition it leaves out really is one no MV
partition reads. Either of the
+ // two reads above counts: a limit that was in effect while the
mapping was built leaves it
+ // incomplete even if the limit is gone by now.
+ if ((partitionSyncLimitActiveBeforeMapping ||
partitionSyncLimitActiveAfterMapping)
+ &&
!mappedBasePartitions.containsAll(changedBasePartitions.keySet())) {
+ LOG.info("Changed base partitions are outside the
partition_sync_limit window and the MV may "
+ + "still hold their rows, rebuild the whole MV.
baseTable={}, changedPartitions={}, "
+ + "undescribed={}, mv={}", baseTableInfo,
changedBasePartitions.keySet(),
+ Sets.difference(changedBasePartitions.keySet(),
mappedBasePartitions), name);
+ return Optional.empty();
+ }
+ return Optional.of(res);
+ } catch (Exception e) {
+ // The base table change is applied either way, so this must not
fail the DDL: warn and take
+ // the safe direction instead.
+ LOG.warn("Failed to map base table partitions to MV partitions,
rebuild the whole MV. "
+ + "baseTable={}, changedPartitions={}, mv={}",
baseTableInfo, changedBasePartitions,
+ name, e);
+ return Optional.empty();
+ } finally {
+ MetaLockUtils.readUnlockTables(tablesToRead);
+ }
+ }
+
+ /**
+ * Resolve the PCT table that {@code baseTableInfo} refers to, or null
when the MV has no PCT entry
+ * for it.
+ */
+ private MTMVRelatedTableIf findPctTable(BaseTableInfo baseTableInfo) {
+ for (BaseColInfo pctInfo : mvPartitionInfo.getPctInfos()) {
+ if (!pctInfo.getTableInfo().equals(baseTableInfo)) {
+ continue;
+ }
+ try {
+ TableIf pctTable = MTMVUtil.getTable(pctInfo.getTableInfo());
+ if (pctTable instanceof MTMVRelatedTableIf) {
+ return (MTMVRelatedTableIf) pctTable;
+ }
+ } catch (Exception e) {
+ LOG.warn("Failed to resolve PCT table {}, mv={}",
pctInfo.getTableInfo(), name, e);
+ }
+ return null;
+ }
+ return null;
}
/**
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/common/util/MetaLockUtils.java
b/fe/fe-core/src/main/java/org/apache/doris/common/util/MetaLockUtils.java
index ffd411d0cf3..6e978c09c07 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/common/util/MetaLockUtils.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/util/MetaLockUtils.java
@@ -52,6 +52,26 @@ public class MetaLockUtils {
}
}
+ /**
+ * Read-lock every table, giving up instead of waiting when one of them
stays busy for {@code timeout}.
+ * The locks taken so far are released before returning false, so a caller
that cannot proceed holds
+ * nothing. Use this where blocking is not an option: a caller that
already holds another table's write
+ * lock can deadlock against a thread that holds this one and wants that
one.
+ *
+ * @return true when every table is read-locked, false when the locks were
released again
+ */
+ public static boolean tryReadLockTables(List<? extends TableIf> tableList,
long timeout, TimeUnit unit) {
+ for (int i = 0; i < tableList.size(); i++) {
+ if (!tableList.get(i).tryReadLock(timeout, unit)) {
+ for (int j = i - 1; j >= 0; j--) {
+ tableList.get(j).readUnlock();
+ }
+ return false;
+ }
+ }
+ return true;
+ }
+
public static void readUnlockTables(List<? extends TableIf> tableList) {
for (int i = tableList.size() - 1; i >= 0; i--) {
tableList.get(i).readUnlock();
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPartitionUtil.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPartitionUtil.java
index d0bbc341aa5..b30df0636a9 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPartitionUtil.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPartitionUtil.java
@@ -91,6 +91,15 @@ public class MTMVPartitionUtil {
new MTMVRelatedPartitionDescTransferGenerator()
);
+ /**
+ * Whether a partition_sync_limit is in effect, i.e. whether the windowed
mapping leaves base partitions
+ * out. Only that mapping is bounded by the window, so only there does its
completeness depend on it.
+ */
+ public static boolean isPartitionSyncLimitActive(Map<String, String>
mvProperties) {
+ return MTMVRelatedPartitionDescSyncLimitGenerator
+
.generateMTMVPartitionSyncConfigByProperties(mvProperties).getSyncLimit() > 0;
+ }
+
/**
* Determine whether the partition is sync with retated partition and
other baseTables
*
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPropertyUtil.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPropertyUtil.java
index ebd4a6f166d..f14c82f1f7d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPropertyUtil.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPropertyUtil.java
@@ -45,6 +45,12 @@ import java.util.Optional;
import java.util.Set;
public class MTMVPropertyUtil {
+ /** The properties a partition_sync_limit window is built from; see
MTMV#alterMvProperties. */
+ private static final List<String> PARTITION_SYNC_WINDOW_KEYS =
Lists.newArrayList(
+ PropertyAnalyzer.PROPERTIES_PARTITION_SYNC_LIMIT,
+ PropertyAnalyzer.PROPERTIES_PARTITION_TIME_UNIT,
+ PropertyAnalyzer.PROPERTIES_PARTITION_DATE_FORMAT);
+
public static final Set<String> MV_PROPERTY_KEYS = Sets.newHashSet(
PropertyAnalyzer.PROPERTIES_GRACE_PERIOD,
PropertyAnalyzer.PROPERTIES_EXCLUDED_TRIGGER_TABLES,
@@ -231,6 +237,56 @@ public class MTMVPropertyUtil {
return !StringUtils.isEmpty(value) && Integer.parseInt(value) > 0;
}
+ /** Whether the given (altered) properties touch the partition_sync_limit
window. */
+ public static boolean containsPartitionSyncWindow(Map<String, String>
properties) {
+ for (String property : PARTITION_SYNC_WINDOW_KEYS) {
+ if (properties.containsKey(property)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Whether the change from {@code oldWindowProperties} to {@code
newWindowProperties} can bring base
+ * partitions back into the window the MV maintains: a limit that is
removed, or a wider one. Those
+ * partitions' deltas were skipped while they were outside, so nothing
incremental can repair them.
+ *
+ * <p>A window that starts applying is not such a change -- it only takes
partitions out of what the MV
+ * maintains, and partition sync drops those before the refresh plans --
and neither is a narrower one.
+ * Windows in different units or date formats are not comparable without a
clock, so a change to either
+ * is treated as one that may widen.
+ */
+ public static boolean partitionSyncWindowWidens(Map<String, String>
oldWindowProperties,
+ Map<String, String> newWindowProperties) {
+ MTMVPartitionSyncConfig oldWindow =
MTMVRelatedPartitionDescSyncLimitGenerator
+
.generateMTMVPartitionSyncConfigByProperties(oldWindowProperties);
+ if (oldWindow.getSyncLimit() <= 0) {
+ return false;
+ }
+ MTMVPartitionSyncConfig newWindow =
MTMVRelatedPartitionDescSyncLimitGenerator
+
.generateMTMVPartitionSyncConfigByProperties(newWindowProperties);
+ if (newWindow.getSyncLimit() <= 0) {
+ return true;
+ }
+ return !oldWindow.getTimeUnit().equals(newWindow.getTimeUnit())
+ || !oldWindow.getDateFormat().equals(newWindow.getDateFormat())
+ || newWindow.getSyncLimit() > oldWindow.getSyncLimit();
+ }
+
+ /**
+ * The window the given properties describe, for comparing it across an
ALTER. Only the properties the
+ * window is built from are read, and the values are compared as they are
stored: setting the same
+ * window again changes nothing about which rows the MV owes and must not
force a rebuild.
+ */
+ public static Map<String, String> partitionSyncWindowOf(Map<String,
String> properties) {
+ Map<String, String> res = Maps.newHashMap();
+ for (String property : PARTITION_SYNC_WINDOW_KEYS) {
+ res.put(property, properties.get(property));
+ }
+ return res;
+ }
+
/**
* Look up the window limit configured for a base table, mirroring the
* excluded_trigger_tables name-matching semantics (empty db/ctl wildcard).
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRefreshSnapshot.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRefreshSnapshot.java
index 608b089bf83..bc6e1827c9e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRefreshSnapshot.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRefreshSnapshot.java
@@ -24,10 +24,8 @@ import com.google.common.collect.Sets;
import com.google.gson.annotations.SerializedName;
import org.apache.commons.collections4.MapUtils;
-import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
-import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentMap;
@@ -61,28 +59,6 @@ public class MTMVRefreshSnapshot {
return partitionSnapshot.getPctSnapshot(pctTableInfo).keySet();
}
- public Optional<Set<String>> getMvPartitionNames(BaseTableInfo
pctTableInfo,
- Map<String, Long> pctPartitions) {
- Set<String> matchedPctPartitions = new HashSet<>();
- Set<String> mvPartitionNames = new HashSet<>();
- for (Map.Entry<String, MTMVRefreshPartitionSnapshot> entry :
partitionSnapshots.entrySet()) {
- Map<String, MTMVSnapshotIf> pctSnapshots =
entry.getValue().getPcts().get(pctTableInfo);
- if (pctSnapshots == null) {
- continue;
- }
- for (Map.Entry<String, Long> pctPartition :
pctPartitions.entrySet()) {
- MTMVSnapshotIf snapshot =
pctSnapshots.get(pctPartition.getKey());
- if (snapshot instanceof MTMVVersionSnapshot
- && ((MTMVVersionSnapshot) snapshot).getId() ==
pctPartition.getValue()) {
- matchedPctPartitions.add(pctPartition.getKey());
- mvPartitionNames.add(entry.getKey());
- }
- }
- }
- return matchedPctPartitions.size() == pctPartitions.size()
- ? Optional.of(mvPartitionNames) : Optional.empty();
- }
-
public boolean equalsWithBaseTable(String mtmvPartitionName, BaseTableInfo
tableInfo,
MTMVSnapshotIf baseTableCurrentSnapshot) {
MTMVRefreshPartitionSnapshot partitionSnapshot =
partitionSnapshots.get(mtmvPartitionName);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelatedPartitionDescSyncLimitGenerator.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelatedPartitionDescSyncLimitGenerator.java
index 48476a6f062..d48c9cbe9b7 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelatedPartitionDescSyncLimitGenerator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelatedPartitionDescSyncLimitGenerator.java
@@ -77,7 +77,7 @@ public class MTMVRelatedPartitionDescSyncLimitGenerator
implements MTMVRelatedPa
* @param mvProperties
* @return
*/
- public MTMVPartitionSyncConfig generateMTMVPartitionSyncConfigByProperties(
+ public static MTMVPartitionSyncConfig
generateMTMVPartitionSyncConfigByProperties(
Map<String, String> mvProperties) {
int syncLimit =
StringUtils.isEmpty(mvProperties.get(PropertyAnalyzer.PROPERTIES_PARTITION_SYNC_LIMIT))
? -1
:
Integer.parseInt(mvProperties.get(PropertyAnalyzer.PROPERTIES_PARTITION_SYNC_LIMIT));
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 3bfa209fd6f..41a53adf77b 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
@@ -115,13 +115,22 @@ public class MTMVRelationManager implements
MTMVHookService {
if
(MTMVPartitionUtil.isTableExcluded(mtmv.getExcludedTriggerTables(),
baseTableName)) {
continue;
}
+ boolean invalidated;
if (allPartitionsChanged) {
mtmv.invalidateIvmBaseline();
+ invalidated = true;
} else {
- mtmv.invalidateIvmBaseline(baseTableInfo, changedPartitions);
+ invalidated = mtmv.invalidateIvmBaseline(baseTableInfo,
changedPartitions);
+ }
+ // A partition change that no MV partition reads leaves nothing to
rebuild, and saying that it
+ // invalidated the baseline would claim a persisted barrier that
does not exist.
+ if (invalidated) {
+ LOG.info("Invalidated IVM baseline, baseTable={}, mtmv={},
reason={}",
+ baseTableInfo, mtmvInfo, reason);
+ } else {
+ LOG.info("No MV partition reads the changed base partitions,
nothing to invalidate. "
+ + "baseTable={}, mtmv={}, reason={}", baseTableInfo,
mtmvInfo, reason);
}
- LOG.info("Invalidated IVM baseline, baseTable={}, mtmv={},
reason={}",
- baseTableInfo, mtmvInfo, reason);
}
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/common/util/MetaLockUtilsTest.java
b/fe/fe-core/src/test/java/org/apache/doris/common/util/MetaLockUtilsTest.java
index 8faa5916ea8..054bd9db802 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/common/util/MetaLockUtilsTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/common/util/MetaLockUtilsTest.java
@@ -28,6 +28,7 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.util.List;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
public class MetaLockUtilsTest {
@@ -60,6 +61,47 @@ public class MetaLockUtilsTest {
tableList.get(1).writeUnlock();
}
+ /**
+ * The read batch gives back what it took when a later table stays busy,
so the caller holds nothing
+ * while it decides what to do. Probed the same way as the tests above: a
read lock still held would
+ * block the write lock of the table that was acquired first.
+ */
+ @Test
+ public void testTryReadLockTablesReleasesWhatItTookWhenALaterTableIsBusy()
throws Exception {
+ List<Table> tables = ImmutableList.of(TableTest.newOlapTable(0,
"readable", 0),
+ TableTest.newOlapTable(1, "busy", 0));
+ Table busy = tables.get(1);
+ // Another thread has to hold the write lock: the same thread could
take a read lock underneath
+ // its own write lock (that direction is allowed), which is not the
case under test. Both
+ // acquisitions are bounded and the holder is asserted to end, so a
failure cannot strand it.
+ CountDownLatch locked = new CountDownLatch(1);
+ CountDownLatch released = new CountDownLatch(1);
+ Thread holder = new Thread(() -> {
+ Assertions.assertTrue(busy.tryWriteLock(30, TimeUnit.SECONDS));
+ locked.countDown();
+ try {
+ released.await(30, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ busy.writeUnlock();
+ }
+ });
+ holder.start();
+ try {
+ Assertions.assertTrue(locked.await(30, TimeUnit.SECONDS));
+ Assertions.assertFalse(MetaLockUtils.tryReadLockTables(tables,
100, TimeUnit.MILLISECONDS),
+ "a busy table must fail the batch");
+ Assertions.assertTrue(tables.get(0).tryWriteLock(100,
TimeUnit.MILLISECONDS),
+ "the read lock taken for the first table must have been
released");
+ tables.get(0).writeUnlock();
+ } finally {
+ released.countDown();
+ holder.join(TimeUnit.SECONDS.toMillis(30));
+ }
+ Assertions.assertFalse(holder.isAlive(), "the holder should have
released the table");
+ }
+
@Test
public void testWriteLockTables() throws MetaNotFoundException {
MetaLockUtils.writeLockTables(tableList);
diff --git a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTest.java
index 6877004f2c2..98aeb274eb9 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTest.java
@@ -65,7 +65,6 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
-import java.util.Optional;
import java.util.Set;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -142,26 +141,6 @@ public class MTMVTest {
Assertions.assertEquals(baseToMv.get("baseP1_2"), "mvp1");
}
- @Test
- public void testChangedBasePartitionsRequireCompleteSnapshotMapping() {
- BaseTableInfo baseTableInfo = Mockito.mock(BaseTableInfo.class);
- MTMVRefreshPartitionSnapshot firstSnapshot = new
MTMVRefreshPartitionSnapshot();
- firstSnapshot.getPctSnapshot(baseTableInfo).put("base_p1", new
MTMVVersionSnapshot(1L, 11L));
- MTMVRefreshPartitionSnapshot secondSnapshot = new
MTMVRefreshPartitionSnapshot();
- secondSnapshot.getPctSnapshot(baseTableInfo).put("base_p2", new
MTMVVersionSnapshot(1L, 12L));
- MTMVRefreshSnapshot refreshSnapshot = new MTMVRefreshSnapshot();
- refreshSnapshot.updateSnapshots(
- Map.of("mv_p1", firstSnapshot, "mv_p2", secondSnapshot),
Set.of("mv_p1", "mv_p2"));
-
- Optional<Set<String>> mappedPartitions =
refreshSnapshot.getMvPartitionNames(
- baseTableInfo, Map.of("base_p1", 11L, "base_p2", 12L));
-
- Assertions.assertTrue(mappedPartitions.isPresent());
- Assertions.assertEquals(Set.of("mv_p1", "mv_p2"),
mappedPartitions.get());
- Assertions.assertFalse(refreshSnapshot.getMvPartitionNames(
- baseTableInfo, Map.of("base_p1", 11L, "base_p3",
13L)).isPresent());
- }
-
private Map<PartitionKeyDesc, Set<String>> mockRelatedPartitionDescs()
throws AnalysisException {
Map<PartitionKeyDesc, Set<String>> res = Maps.newHashMap();
Column k1 = new Column("k1",
ScalarType.createType(PrimitiveType.TINYINT), true, null, "", "key1");
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 dfd3d6dc7ef..af53ab71b6b 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
@@ -23,6 +23,7 @@ import org.apache.doris.catalog.MTMV;
import org.apache.doris.catalog.MaterializedIndex;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.Partition;
+import org.apache.doris.catalog.PartitionItem;
import org.apache.doris.catalog.info.TableNameInfo;
import org.apache.doris.catalog.stream.OlapTableStream;
import org.apache.doris.common.Config;
@@ -32,14 +33,11 @@ import org.apache.doris.job.exception.JobException;
import org.apache.doris.job.extensions.mtmv.MTMVTask;
import org.apache.doris.job.extensions.mtmv.MTMVTask.MTMVTaskTriggerMode;
import org.apache.doris.job.extensions.mtmv.MTMVTaskContext;
-import org.apache.doris.mtmv.BaseColInfo;
import org.apache.doris.mtmv.BaseTableInfo;
import org.apache.doris.mtmv.MTMVAlterOpType;
-import org.apache.doris.mtmv.MTMVPartitionInfo.MTMVPartitionType;
+import org.apache.doris.mtmv.MTMVPartitionUtil;
import org.apache.doris.mtmv.MTMVPlanUtil;
-import org.apache.doris.mtmv.MTMVRefreshPartitionSnapshot;
import org.apache.doris.mtmv.MTMVRelation;
-import org.apache.doris.mtmv.MTMVVersionSnapshot;
import org.apache.doris.persist.AlterMTMV;
import org.apache.doris.persist.DropPartitionInfo;
import org.apache.doris.persist.RecoverInfo;
@@ -48,10 +46,19 @@ import org.apache.doris.persist.TruncateTableInfo;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.utframe.TestWithFeService;
+import com.google.common.collect.Sets;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+import java.time.LocalDate;
import java.util.Collections;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
public class IvmBaselineRebuildTest extends TestWithFeService {
@@ -101,7 +108,9 @@ public class IvmBaselineRebuildTest extends
TestWithFeService {
createPartitionedIvmTableAndMv(db);
executeSql("ALTER TABLE ivm_base DROP PARTITION p202001");
-
Assertions.assertTrue(getMtmv(db).getIvmInfo().isBaselineRebuildRequired());
+ // SELF_MANAGE: the single MV partition reads every base partition,
and the partition mapping API
+ // answers nothing for it, so the whole MV has to be rebuilt.
+
Assertions.assertTrue(getMtmv(db).getIvmInfo().requiresCompleteBaselineRebuild());
}
@Test
@@ -122,35 +131,356 @@ public class IvmBaselineRebuildTest extends
TestWithFeService {
Assertions.assertTrue(mtmv.getIvmInfo().isBaselineRebuildRequired());
}
+ /**
+ * Which MV partitions must be rebuilt is decided by the MV's partition
mapping, not by what the
+ * refresh snapshot happens to record. This test publishes no snapshot at
all: an MV whose partitions
+ * follow the base table's still narrows the rebuild down to the
partitions that read the dropped one.
+ */
@Test
- public void testPublishedPctPartitionUsesPartitionsBaselineRebuild()
throws Exception {
+ public void testDropPartitionMarksOnlyMvPartitionsThatReadIt() throws
Exception {
String db = "ivm_partitions_baseline_rebuild";
- createPartitionedIvmTableAndMv(db);
+ createPartitionedIvmTableAndPartitionedMv(db);
MTMV mtmv = getMtmv(db);
- OlapTable baseTable = getBaseTable(db);
- publishPctPartitionSnapshot(mtmv, baseTable, "p202001");
+ Assertions.assertEquals(2, mtmv.getPartitionNames().size());
+ Set<String> expected = mvPartitionsWithSameRange(mtmv,
getBaseTable(db), "p202001");
+ Assertions.assertEquals(1, expected.size());
executeSql("ALTER TABLE ivm_base DROP PARTITION p202001");
Assertions.assertFalse(mtmv.getIvmInfo().requiresCompleteBaselineRebuild());
- Assertions.assertEquals(Collections.singleton("mv_partition"),
- mtmv.getIvmInfo().getPendingBaselineRebuildPartitions());
+ Assertions.assertEquals(expected,
mtmv.getIvmInfo().getPendingBaselineRebuildPartitions());
}
+ /**
+ * A base partition that no MV partition reads: dropping it cannot leave
any of its rows in the MV, so
+ * there is nothing to rebuild. The previous selection could not tell this
apart from "the snapshot does
+ * not know this partition" and rebuilt the whole MV instead.
+ */
@Test
- public void testMissingPctSnapshotRequiresCompleteBaselineRebuild() throws
Exception {
- String db = "ivm_complete_baseline_rebuild";
- createPartitionedIvmTableAndMv(db);
+ public void testDropPartitionOutsideMvPartitionsMarksNothing() throws
Exception {
+ String db = "ivm_partition_outside_mv";
+ createPartitionedIvmTableAndPartitionedMv(db);
+ MTMV mtmv = getMtmv(db);
+ // Added after the MV was created and never synced into it, so no MV
partition reads it.
+ executeSql("ALTER TABLE ivm_base ADD PARTITION p202003 "
+ + "VALUES [('2020-03-01'), ('2020-04-01'))");
+ Assertions.assertTrue(mvPartitionsWithSameRange(mtmv,
getBaseTable(db), "p202003").isEmpty());
+
+ executeSql("ALTER TABLE ivm_base DROP PARTITION p202003");
+
+ Assertions.assertFalse(mtmv.getIvmInfo().isBaselineRebuildRequired());
+ }
+
+ /**
+ * A base partition the partition_sync_limit window no longer covers can
still have its rows in an MV
+ * partition: the MV was built while that partition was inside the window,
shrinking the window does not
+ * touch the MV's own partitions, and widening it again makes partition
sync keep the partition holding
+ * those rows. TRUNCATE emits no binlog, so nothing incremental can repair
them -- the whole MV has to be
+ * rebuilt rather than a partition being guessed at.
+ */
+ @Test
+ public void testChangedPartitionOutsideTheSyncWindowRebuildsTheWholeMv()
throws Exception {
+ String db = "ivm_baseline_sync_window";
+ String thisYear = LocalDate.now().withDayOfYear(1).toString();
+ // The cutoff is now() truncated to the year, read when the marker
runs, and a partition is kept
+ // while its upper bound is after it. The recent partition therefore
ends more than one year out:
+ // a year boundary falling between building this DDL and marking the
change would otherwise put
+ // its upper bound exactly on the cutoff, drop it from the mapping,
and let this test pass through
+ // the "nothing was selected" answer it exists to rule out.
+ String recentEnd =
LocalDate.now().withDayOfYear(1).plusYears(2).toString();
+ createDatabaseAndUse(db);
+ createTable("CREATE TABLE " + db + ".ivm_base (\n"
+ + " dt date NOT NULL,\n"
+ + " k1 int,\n"
+ + " v1 int\n"
+ + ")\n"
+ + "DUPLICATE KEY(dt, k1)\n"
+ + "PARTITION BY RANGE(dt) (\n"
+ + " PARTITION p202001 VALUES [('2020-01-01'),
('2020-02-01')),\n"
+ + " PARTITION p202002 VALUES [('2020-02-01'),
('2020-03-01')),\n"
+ + " PARTITION pThisYear VALUES [('" + thisYear + "'), ('" +
recentEnd + "'))\n"
+ + ")\n"
+ + "DISTRIBUTED BY HASH(k1) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1', 'binlog.enable' =
'true', 'binlog.format' = 'ROW')");
+ createMvByNereids("CREATE MATERIALIZED VIEW ivm_mv\n"
+ + "BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL\n"
+ + "PARTITION BY(dt)\n"
+ + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')\n"
+ + "AS SELECT dt, k1, v1 FROM ivm_base");
MTMV mtmv = getMtmv(db);
-
mtmv.getMvPartitionInfo().setPartitionType(MTMVPartitionType.FOLLOW_BASE_TABLE);
- mtmv.getMvPartitionInfo().setPctInfos(Collections.singletonList(
- new BaseColInfo("dt", new BaseTableInfo(getBaseTable(db)))));
+ Assertions.assertEquals(3, mtmv.getPartitionNames().size());
+
+ // The window now keeps only this year's partition, so p202001 leaves
the mapping while the MV's own
+ // partition for it stays. TRUNCATE leaves the base partition in
place, so partition sync would keep
+ // that MV partition too -- the rows it still holds are exactly what
the rebuild has to remove.
+ executeSql("ALTER MATERIALIZED VIEW ivm_mv SET ('partition_sync_limit'
= '1',"
+ + " 'partition_sync_time_unit' = 'YEAR')");
+ executeSql("TRUNCATE TABLE ivm_base PARTITION(p202001)");
+
+
Assertions.assertTrue(mtmv.getIvmInfo().requiresCompleteBaselineRebuild());
+ }
+
+ /**
+ * The same window, with a change that touches a partition inside it and
one outside it at once: the
+ * partition inside fills the selection, and the one outside contributes
nothing because the window
+ * left it out of the mapping. Judging the change by "was anything
selected" would mark only the MV
+ * partition backed by the inside half, and the rows of the outside half
-- which the MV partition for
+ * it still holds -- would never be rebuilt.
+ */
+ @Test
+ public void
testChangeThatMixesInWindowAndOutOfWindowPartitionsRebuildsTheWholeMv() throws
Exception {
+ String db = "ivm_baseline_sync_window_mixed";
+ String thisYear = LocalDate.now().withDayOfYear(1).toString();
+ // The cutoff is now() truncated to the year, read when the marker
runs, and a partition is kept
+ // while its upper bound is after it. The recent partition therefore
ends more than one year out:
+ // a year boundary falling between building this DDL and marking the
change would otherwise put
+ // its upper bound exactly on the cutoff, drop it from the mapping,
and let this test pass through
+ // the "nothing was selected" answer it exists to rule out.
+ String recentEnd =
LocalDate.now().withDayOfYear(1).plusYears(2).toString();
+ createDatabaseAndUse(db);
+ createTable("CREATE TABLE " + db + ".ivm_base (\n"
+ + " dt date NOT NULL,\n"
+ + " k1 int,\n"
+ + " v1 int\n"
+ + ")\n"
+ + "DUPLICATE KEY(dt, k1)\n"
+ + "PARTITION BY RANGE(dt) (\n"
+ + " PARTITION p202001 VALUES [('2020-01-01'),
('2020-02-01')),\n"
+ + " PARTITION p202002 VALUES [('2020-02-01'),
('2020-03-01')),\n"
+ + " PARTITION pThisYear VALUES [('" + thisYear + "'), ('" +
recentEnd + "'))\n"
+ + ")\n"
+ + "DISTRIBUTED BY HASH(k1) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1', 'binlog.enable' =
'true', 'binlog.format' = 'ROW')");
+ createMvByNereids("CREATE MATERIALIZED VIEW ivm_mv\n"
+ + "BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL\n"
+ + "PARTITION BY(dt)\n"
+ + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')\n"
+ + "AS SELECT dt, k1, v1 FROM ivm_base");
+ MTMV mtmv = getMtmv(db);
+ Assertions.assertEquals(3, mtmv.getPartitionNames().size());
+
+ executeSql("ALTER MATERIALIZED VIEW ivm_mv SET ('partition_sync_limit'
= '1',"
+ + " 'partition_sync_time_unit' = 'YEAR')");
+ // One statement, so the marker sees both partitions together:
pThisYear is inside the window while
+ // p202001 is not, which is exactly the mix a non-empty selection must
not be allowed to hide.
+ executeSql("TRUNCATE TABLE ivm_base PARTITION(p202001, pThisYear)");
+
+
Assertions.assertTrue(mtmv.getIvmInfo().requiresCompleteBaselineRebuild());
+ }
+
+ /**
+ * A window property decides which base partitions the MV maintains. Only
a change that can bring a
+ * partition back into that set needs a complete baseline rebuild: its
deltas were skipped while it was
+ * outside, so nothing incremental can repair them. A window that starts
applying, a narrower one and
+ * one that describes the same set as before all leave the deltas that
were applied intact, and the
+ * partitions they take out are dropped by partition sync before the
refresh plans.
+ */
+ @Test
+ public void testOnlyAWiderSyncWindowRequiresCompleteBaselineRebuild()
throws Exception {
+ String db = "ivm_sync_window_property_change";
+ createPartitionedIvmTableAndPartitionedMv(db);
+ MTMV mtmv = getMtmv(db);
+ Assertions.assertFalse(mtmv.getIvmInfo().isBaselineRebuildRequired());
+
+ // No limit is in effect, so the unit it is paired with decides
nothing.
+ executeSql("ALTER MATERIALIZED VIEW ivm_mv SET
('partition_sync_time_unit' = 'YEAR')");
+ Assertions.assertFalse(mtmv.getIvmInfo().isBaselineRebuildRequired());
+
+ // The window starts applying: it takes partitions out of what the MV
maintains, it brings none back.
+ executeSql("ALTER MATERIALIZED VIEW ivm_mv SET ('partition_sync_limit'
= '10')");
+ Assertions.assertFalse(mtmv.getIvmInfo().isBaselineRebuildRequired());
+
+ // The same window, restated.
+ executeSql("ALTER MATERIALIZED VIEW ivm_mv SET ('partition_sync_limit'
= '10')");
+ Assertions.assertFalse(mtmv.getIvmInfo().isBaselineRebuildRequired());
+
+ // Narrower: it only removes partitions from the maintained set.
+ executeSql("ALTER MATERIALIZED VIEW ivm_mv SET ('partition_sync_limit'
= '1')");
+ Assertions.assertFalse(mtmv.getIvmInfo().isBaselineRebuildRequired());
+
+ // Wider: the partitions it takes back in skipped their deltas while
they were outside.
+ executeSql("ALTER MATERIALIZED VIEW ivm_mv SET ('partition_sync_limit'
= '10')");
+
Assertions.assertTrue(mtmv.getIvmInfo().requiresCompleteBaselineRebuild());
+ clearBaselineRebuild(mtmv);
+
+ // The limit is gone: every partition comes back.
+ executeSql("ALTER MATERIALIZED VIEW ivm_mv SET ('partition_sync_limit'
= '0')");
+
Assertions.assertTrue(mtmv.getIvmInfo().requiresCompleteBaselineRebuild());
+ clearBaselineRebuild(mtmv);
+
+ // Still no limit in effect, so the unit decides nothing again.
+ executeSql("ALTER MATERIALIZED VIEW ivm_mv SET
('partition_sync_time_unit' = 'DAY')");
+ Assertions.assertFalse(mtmv.getIvmInfo().isBaselineRebuildRequired());
+ }
+
+ /**
+ * The limit is read on both sides of the mapping the selection judges:
the mapping is built under
+ * whatever window the properties hold at that moment, and MV properties
are mutable in between --
+ * <code>ALTER MATERIALIZED VIEW ... SET</code> is not generation guarded,
so a limit can be cleared
+ * while the mapping is built. A read that happens only afterwards then
sees no limit and trusts a
+ * windowed mapping, and a change outside that window is answered with "no
MV partition reads it",
+ * which records no barrier at all. The read taken before the mapping is
the one that cannot be
+ * reconstructed afterwards, so this pins that the selection takes both.
+ *
+ * <p>The interleaving itself is not staged: the mapping is built with no
injection point between the
+ * two reads, so the test pins that both reads happen rather than a racy
outcome.
+ */
+ @Test
+ public void testTheSyncLimitIsReadOnBothSidesOfTheMapping() throws
Exception {
+ String db = "ivm_baseline_sync_limit_both_reads";
+ createPartitionedIvmTableAndPartitionedMv(db);
+ MTMV mtmv = getMtmv(db);
+ OlapTable baseTable = getBaseTable(db);
+ executeSql("ALTER MATERIALIZED VIEW ivm_mv SET ('partition_sync_limit'
= '1',"
+ + " 'partition_sync_time_unit' = 'YEAR')");
+
+ try (MockedStatic<MTMVPartitionUtil> partitionUtil =
Mockito.mockStatic(MTMVPartitionUtil.class,
+ Mockito.CALLS_REAL_METHODS)) {
+ Assertions.assertTrue(mtmv.invalidateIvmBaseline(new
BaseTableInfo(baseTable),
+ Collections.singletonMap("p202001",
baseTable.getPartition("p202001").getId())));
+ partitionUtil.verify(() ->
MTMVPartitionUtil.isPartitionSyncLimitActive(Mockito.any()),
+ Mockito.times(2));
+ }
+
+ // p202001 is outside the window while it is in effect, so its rows
are described by no mapping
+ // entry and only the limit can tell that apart from "no MV partition
reads it".
+
Assertions.assertTrue(mtmv.getIvmInfo().requiresCompleteBaselineRebuild());
+ }
+
+ /**
+ * The partition mapping is built from the MV's PCT tables only. A changed
partition of a joined table
+ * the MV's partition column does not reach is invisible to it, and
missing such a change leaves rows
+ * of the dropped partition in the MV forever, so the whole MV has to be
rebuilt.
+ */
+ @Test
+ public void
testNonPctBaseTablePartitionChangeRequiresCompleteBaselineRebuild() throws
Exception {
+ String db = "ivm_non_pct_partition_change";
+ createPartitionedIvmTable(db);
+ createTable("CREATE TABLE " + db + ".ivm_dim (\n"
+ + " dt date NOT NULL,\n"
+ + " id int NOT NULL,\n"
+ + " v int\n"
+ + ")\n"
+ + "DUPLICATE KEY(dt, id)\n"
+ + "PARTITION BY RANGE(dt) (\n"
+ + " PARTITION d202001 VALUES [('2020-01-01'),
('2020-02-01')),\n"
+ + " PARTITION d202002 VALUES [('2020-02-01'),
('2020-03-01'))\n"
+ + ")\n"
+ + "DISTRIBUTED BY HASH(id) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1', 'binlog.enable' =
'true', "
+ + "'binlog.format' = 'ROW')");
+ // The join is on a non-partition column, so ivm_dim is a base table
of the MV but not a PCT table.
+ createMvByNereids("CREATE MATERIALIZED VIEW ivm_mv\n"
+ + "BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL\n"
+ + "PARTITION BY(dt)\n"
+ + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')\n"
+ + "AS SELECT b.dt, b.k1, b.v1 FROM ivm_base b JOIN ivm_dim d
ON b.k1 = d.id");
+ MTMV mtmv = getMtmv(db);
+ Assertions.assertTrue(mtmv.isIvm());
+ Assertions.assertEquals(Sets.newHashSet("ivm_base"),
+ mtmv.getMvPartitionInfo().getPctInfos().stream()
+ .map(pctInfo -> pctInfo.getTableInfo().getTableName())
+ .collect(Collectors.toSet()));
+
+ executeSql("ALTER TABLE ivm_dim DROP PARTITION d202001");
+
+
Assertions.assertTrue(mtmv.getIvmInfo().requiresCompleteBaselineRebuild());
+ }
+
+ /**
+ * A join whose condition carries the MV's partition column makes both
tables PCT tables, so the mapping
+ * reads both of them. The marker already holds this table's write lock,
so it takes the other one with a
+ * bounded tryLock: while that table is free, the rebuild is still
narrowed to the partitions that read
+ * the dropped one.
+ */
+ @Test
+ public void testMultiPctTablePartitionChangeStillNarrows() throws
Exception {
+ String db = "ivm_multi_pct_partition_change";
+ createTwoPctTableIvm(db);
+ MTMV mtmv = getMtmv(db);
+ Set<String> expected = mvPartitionsWithSameRange(mtmv,
getBaseTable(db), "p202001");
+ Assertions.assertEquals(1, expected.size());
executeSql("ALTER TABLE ivm_base DROP PARTITION p202001");
+
Assertions.assertFalse(mtmv.getIvmInfo().requiresCompleteBaselineRebuild());
+ Assertions.assertEquals(expected,
mtmv.getIvmInfo().getPendingBaselineRebuildPartitions());
+ }
+
+ /**
+ * The same MV, but the other PCT table is being written while the
partition DDL marks. Waiting for it
+ * would close a cycle with the DDL that holds it -- each would hold the
write lock the other one needs --
+ * so the marker gives up on the mapping and the whole MV is rebuilt.
+ */
+ @Test
+ public void testMultiPctTableBusyOtherTableRebuildsWholeMv() throws
Exception {
+ String db = "ivm_multi_pct_busy";
+ createTwoPctTableIvm(db);
+ MTMV mtmv = getMtmv(db);
+ OlapTable otherPctTable = (OlapTable)
getDb(db).getTableOrMetaException("ivm_dim");
+ CountDownLatch locked = new CountDownLatch(1);
+ CountDownLatch released = new CountDownLatch(1);
+ // The acquisitions are bounded on both sides so that a failure cannot
leave this worker holding
+ // the table forever: it is not a daemon, and the test asserts that it
ends.
+ Thread writer = new Thread(() -> {
+ Assertions.assertTrue(otherPctTable.tryWriteLock(30,
TimeUnit.SECONDS),
+ "the test worker should be able to take the other PCT
table");
+ locked.countDown();
+ try {
+ released.await(30, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ otherPctTable.writeUnlock();
+ }
+ });
+ writer.start();
+ Assertions.assertTrue(locked.await(30, TimeUnit.SECONDS));
+ try {
+ executeSql("ALTER TABLE ivm_base DROP PARTITION p202001");
+ } finally {
+ released.countDown();
+ }
+ writer.join(TimeUnit.SECONDS.toMillis(30));
+ Assertions.assertFalse(writer.isAlive(), "the test worker should have
released the table");
+
+ // The batch fails on its first table here, so what this covers is the
whole-MV fallback; the
+ // release of the locks taken before the busy one is covered in
MetaLockUtilsTest.
Assertions.assertTrue(mtmv.getIvmInfo().requiresCompleteBaselineRebuild());
}
+ /**
+ * A join whose condition carries the MV's partition column: both tables
become PCT tables.
+ */
+ private void createTwoPctTableIvm(String db) throws Exception {
+ createPartitionedIvmTable(db);
+ createTable("CREATE TABLE " + db + ".ivm_dim (\n"
+ + " dt date NOT NULL,\n"
+ + " k1 int,\n"
+ + " v int\n"
+ + ")\n"
+ + "DUPLICATE KEY(dt, k1)\n"
+ + "PARTITION BY RANGE(dt) (\n"
+ + " PARTITION p202001 VALUES [('2020-01-01'),
('2020-02-01')),\n"
+ + " PARTITION p202002 VALUES [('2020-02-01'),
('2020-03-01'))\n"
+ + ")\n"
+ + "DISTRIBUTED BY HASH(k1) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1', 'binlog.enable' =
'true', "
+ + "'binlog.format' = 'ROW')");
+ createMvByNereids("CREATE MATERIALIZED VIEW ivm_mv\n"
+ + "BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL\n"
+ + "PARTITION BY(dt)\n"
+ + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')\n"
+ + "AS SELECT b.dt, b.k1, b.v1 FROM ivm_base b JOIN ivm_dim d
ON b.dt = d.dt");
+ MTMV mtmv = getMtmv(db);
+ Assertions.assertTrue(mtmv.isIvm());
+ Assertions.assertEquals(2,
mtmv.getMvPartitionInfo().getPctInfos().size());
+ }
+
@Test
public void testReplacePartitionMarksBaselineRebuild() throws Exception {
String db = "ivm_broken_replace_partition";
@@ -176,6 +506,24 @@ public class IvmBaselineRebuildTest extends
TestWithFeService {
Assertions.assertTrue(getMtmv(db).getIvmInfo().isBaselineRebuildRequired());
}
+ /**
+ * RECOVER PARTITION marks before the partition is added back to the
table, so at that moment the
+ * partition is still in the recycle bin and the mapping cannot describe
it. The rebuild must not lean
+ * on the DROP that came before either: its barrier is released here
before the recovery.
+ */
+ @Test
+ public void
testRecoverPartitionOnPartitionedMvRequiresCompleteBaselineRebuild() throws
Exception {
+ String db = "ivm_recover_partition_complete";
+ createPartitionedIvmTableAndPartitionedMv(db);
+ MTMV mtmv = getMtmv(db);
+ executeSql("ALTER TABLE ivm_base DROP PARTITION p202001");
+ clearBaselineRebuild(mtmv);
+
+ executeSql("RECOVER PARTITION p202001 FROM ivm_base");
+
+
Assertions.assertTrue(mtmv.getIvmInfo().requiresCompleteBaselineRebuild());
+ }
+
@Test
public void testRecoverAndDropKeepGlobalBrokenState() throws Exception {
String db = "ivm_broken_recover_and_drop";
@@ -192,6 +540,32 @@ public class IvmBaselineRebuildTest extends
TestWithFeService {
Assertions.assertTrue(mtmv.getIvmInfo().isBaselineRebuildRequired());
}
+ /**
+ * RECOVER reports the recycled partition under the name it had, and a
partition added after the drop
+ * can be live under that name again by then, with a different range. The
change is then not the one
+ * the mapping describes: the recovered range is the one whose rows have
to come back, and its MV
+ * partition -- which partition sync adds when the recovered partition
returns -- is read from a base
+ * partition that the replacement does not describe at all. Narrowing to
the replacement's MV
+ * partitions would leave that one out, and recovery emits no row binlog
to fill it later, so the
+ * whole MV has to be rebuilt.
+ */
+ @Test
+ public void testRecoveredPartitionWhoseNameWasReusedRebuildsTheWholeMv()
throws Exception {
+ String db = "ivm_recover_partition_name_reused";
+ createPartitionedIvmTableAndPartitionedMv(db);
+ MTMV mtmv = getMtmv(db);
+
+ executeSql("ALTER TABLE ivm_base DROP PARTITION p202001");
+ clearBaselineRebuild(mtmv);
+ // Live again under the dropped name, with a range no MV partition
covers: the RECOVER below is
+ // still about the recycled partition, not about this one.
+ executeSql("ALTER TABLE ivm_base ADD PARTITION p202001 VALUES
[('2020-04-01'), ('2020-05-01'))");
+
+ executeSql("RECOVER PARTITION p202001 AS p202003 FROM ivm_base");
+
+
Assertions.assertTrue(mtmv.getIvmInfo().requiresCompleteBaselineRebuild());
+ }
+
@Test
public void testAddPartitionDoesNotMarkBaselineRebuild() throws Exception {
String db = "ivm_broken_add_partition";
@@ -551,6 +925,37 @@ public class IvmBaselineRebuildTest extends
TestWithFeService {
}
private void createPartitionedIvmTableAndMv(String db) throws Exception {
+ createPartitionedIvmTable(db);
+ createMvByNereids("CREATE MATERIALIZED VIEW ivm_mv\n"
+ + "BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL\n"
+ + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')\n"
+ + "AS SELECT dt, k1, v1 FROM ivm_base");
+ assertFreshMv(db);
+ }
+
+ /**
+ * The same base table, but the MV follows the base table's partitions: it
is created with one MV
+ * partition per base partition, so a partition change can be narrowed to
the MV partitions that read
+ * the changed one.
+ */
+ private void createPartitionedIvmTableAndPartitionedMv(String db) throws
Exception {
+ createPartitionedIvmTable(db);
+ createMvByNereids("CREATE MATERIALIZED VIEW ivm_mv\n"
+ + "BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL\n"
+ + "PARTITION BY(dt)\n"
+ + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')\n"
+ + "AS SELECT dt, k1, v1 FROM ivm_base");
+ assertFreshMv(db);
+ }
+
+ private void assertFreshMv(String db) throws Exception {
+ Assertions.assertTrue(getMtmv(db).isIvm());
+
Assertions.assertFalse(getMtmv(db).getIvmInfo().isBaselineRebuildRequired());
+ }
+
+ private void createPartitionedIvmTable(String db) throws Exception {
createDatabaseAndUse(db);
createTable("CREATE TABLE " + db + ".ivm_base (\n"
+ " dt date NOT NULL,\n"
@@ -564,13 +969,6 @@ public class IvmBaselineRebuildTest extends
TestWithFeService {
+ ")\n"
+ "DISTRIBUTED BY HASH(k1) BUCKETS 1\n"
+ "PROPERTIES ('replication_num' = '1', 'binlog.enable' =
'true', 'binlog.format' = 'ROW')");
- createMvByNereids("CREATE MATERIALIZED VIEW ivm_mv\n"
- + "BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL\n"
- + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
- + "PROPERTIES ('replication_num' = '1')\n"
- + "AS SELECT dt, k1, v1 FROM ivm_base");
- Assertions.assertTrue(getMtmv(db).isIvm());
-
Assertions.assertFalse(getMtmv(db).getIvmInfo().isBaselineRebuildRequired());
}
private MTMV getMtmv(String db) throws Exception {
@@ -587,15 +985,20 @@ public class IvmBaselineRebuildTest extends
TestWithFeService {
return (OlapTable) getDb(db).getTableOrMetaException("ivm_base");
}
- private void publishPctPartitionSnapshot(MTMV mtmv, OlapTable baseTable,
String partitionName) {
- BaseTableInfo baseTableInfo = new BaseTableInfo(baseTable);
-
mtmv.getMvPartitionInfo().setPartitionType(MTMVPartitionType.FOLLOW_BASE_TABLE);
- mtmv.getMvPartitionInfo().setPctInfos(Collections.singletonList(new
BaseColInfo("dt", baseTableInfo)));
- MTMVRefreshPartitionSnapshot snapshot = new
MTMVRefreshPartitionSnapshot();
- snapshot.getPctSnapshot(baseTableInfo).put(partitionName,
- new MTMVVersionSnapshot(1L,
baseTable.getPartition(partitionName).getId()));
- mtmv.getRefreshSnapshot().updateSnapshots(
- Collections.singletonMap("mv_partition", snapshot),
Collections.singleton("mv_partition"));
+ /**
+ * The MV partitions whose range is exactly the range of the given base
partition, derived from the two
+ * tables' partition items rather than from the mapping the implementation
under test computes.
+ */
+ private Set<String> mvPartitionsWithSameRange(MTMV mtmv, OlapTable
baseTable, String basePartitionName) {
+ PartitionItem basePartitionItem = baseTable.getPartitionInfo()
+ .getItem(baseTable.getPartition(basePartitionName).getId());
+ Set<String> res = Sets.newHashSet();
+ for (Entry<String, PartitionItem> entry :
mtmv.getAndCopyPartitionItems().entrySet()) {
+ if
(entry.getValue().toPartitionKeyDesc().equals(basePartitionItem.toPartitionKeyDesc()))
{
+ res.add(entry.getKey());
+ }
+ }
+ return res;
}
private void clearBaselineRebuild(MTMV mtmv) {
diff --git
a/regression-test/data/mtmv_p0/ivm/test_ivm_baseline_marker_scope.out
b/regression-test/data/mtmv_p0/ivm/test_ivm_baseline_marker_scope.out
new file mode 100644
index 00000000000..ed1179718df
--- /dev/null
+++ b/regression-test/data/mtmv_p0/ivm/test_ivm_baseline_marker_scope.out
@@ -0,0 +1,20 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !baseline_task --
+SUCCESS COMPLETE NONE
+
+-- !baseline_mv --
+1 2026-01-10 100 dim-a
+2 2026-02-10 200 dim-b
+
+-- !non_pct_drop_task --
+SUCCESS COMPLETE BINLOG_BROKEN
+
+-- !non_pct_drop_mv --
+1 2026-01-10 100 \N
+2 2026-02-10 200 dim-b
+
+-- !unread_partition_task --
+SUCCESS
+
+-- !narrowed_mv --
+2 2026-02-10 200 dim-b
diff --git
a/regression-test/suites/mtmv_p0/ivm/test_ivm_baseline_marker_scope.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_baseline_marker_scope.groovy
new file mode 100644
index 00000000000..b4da953d2d0
--- /dev/null
+++ b/regression-test/suites/mtmv_p0/ivm/test_ivm_baseline_marker_scope.groovy
@@ -0,0 +1,189 @@
+// 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()
+ }
+
+ def taskStatus = { String id ->
+ def rows = sql_return_maparray("""
+ SELECT Status FROM tasks('type'='mv') WHERE TaskId = '${id}'
+ """)
+ return rows[0].Status.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)
+ // waitForNewTask returns on any terminal state, and partition sync
already dropped the MV partition,
+ // so the row set below would match even if this refresh failed. The task
itself has to be asserted.
+ assertEquals("SUCCESS", taskStatus(taskId))
+ order_qt_narrowed_mv """SELECT order_id, dt, amount, dimension_name
+ FROM ${mvName}"""
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
