github-actions[bot] commented on code in PR #68390:
URL: https://github.com/apache/doris/pull/68390#discussion_r4083719393
##########
fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPlanUtil.java:
##########
@@ -1075,13 +1075,32 @@ private static void checkColumnIfChange(MTMV mtmv,
List<ColumnDefinition> analyz
+ "original length is: %s, current length is: %s",
originalColumns.size(), analyzedColumns.size()));
}
- for (int i = 0; i < originalColumns.size(); i++) {
- if (!isTypeLike(originalColumns.get(i).getType(),
analyzedColumns.get(i).getType())) {
+ // Matched by name, not by position. The order of the two lists is
decided by different passes:
+ // the physical schema is laid out when the MV is created, where
MTMVPlanUtil#applyIvmPhysicalKeyLayout
+ // puts the final key columns first, and the analysed list comes from
running that same layout again
+ // with the stored key columns as its input. The two agree except for
a chained IVM MV whose base
+ // tables carry row-id columns of their own: the create pass derives
the visible key prefix from the
+ // identity key slots, the analysed one takes it from the stored keys,
and the base tables' row-id
+ // columns end up in a different block. What this check is for is a
base-table change that makes a
+ // column disappear or change type, and where a column sits is not
part of that.
+ Map<String, Column> originalByName = Maps.newHashMap();
+ for (Column column : originalColumns) {
+ originalByName.put(column.getName().toLowerCase(), column);
+ }
+ for (Column analyzedColumn : analyzedColumns) {
+ Column originalColumn =
originalByName.get(analyzedColumn.getName().toLowerCase());
Review Comment:
[P1] Do not apply name-only matching to non-IVM MVs without restoring their
explicit output aliases. For `CREATE MATERIALIZED VIEW mv (x, y) AS SELECT a, b
...`, creation persists `x/y`, but `rewriteQuerySql` aliases the stored query
only for IVM and `analyzeQueryWithSql` later passes no
`simpleColumnDefinitions`, so this lookup searches for `a/b` and reports
`column not found`. Because the new relation hook runs this for ordinary MVs
and every SCHEMA_CHANGE refresh repeats it, an otherwise compatible base-table
alter leaves such an MV permanently unrefreshable. Preserve positional matching
for non-IVM or persist/reapply the explicit aliases, and add that regression
case.
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java:
##########
@@ -684,33 +759,211 @@ public Map<String, MTMVPartitionState>
getPartitionStates() {
// A payload without the member carries no state at all, which is not the
same as an empty map that
// says the states are now empty: leaving them alone is the only answer
that cannot lose state.
public void alterPartitionStates(Map<String, MTMVPartitionState>
partitionStates) {
- if (partitionStates == null) {
- return;
- }
+ replayAlterPartitionStates(partitionStates, null);
+ }
+
+ /**
+ * ALTER_PARTITION_STATES replay: applies the states the payload carries,
and drops the snapshots it
+ * names. Both in one lock acquisition, because a reader that saw the new
requirement while the
+ * snapshot was still there could let a transparent rewrite serve rows the
rebuild has to replace.
+ *
+ * <p>A payload without the states carries none, which is not the same as
an empty map that says the
+ * states are now empty: leaving them alone is the only answer that cannot
lose state.
+ */
+ public void replayAlterPartitionStates(Map<String, MTMVPartitionState>
partitionStates,
+ Set<String> removedSnapshotPartitions) {
writeMvLock();
try {
- this.partitionStates = MTMVPartitionState.copyOf(partitionStates);
+ if (partitionStates != null) {
+ this.partitionStates =
MTMVPartitionState.copyOf(partitionStates);
+ }
+ refreshSnapshot.removeSnapshots(removedSnapshotPartitions);
} finally {
writeMvUnlock();
}
}
- public void invalidateIvmBaseline() {
- EditLogItem editLogItem;
+ /**
+ * The {@code latestEpoch} of the given MV partitions, taken under the MV
read lock.
+ *
+ * <p>This is the value a refresh has to remember: what it read from the
base tables is described by
+ * the requirement in force when it started reading, so writing that value
back as the new
+ * {@code refreshEpoch} is what keeps an invalidation arriving mid-refresh
from being swallowed. A
+ * partition without an entry is left out -- a caller writes an epoch only
for what it captured.
+ */
+ public Map<String, Long> getLatestEpochs(Set<String> partitionNames) {
+ if (CollectionUtils.isEmpty(partitionNames)) {
+ return Collections.emptyMap();
+ }
+ // Sized before the lock: the state map is what needs it, and building
the map is not part of that.
+ Map<String, Long> res =
Maps.newHashMapWithExpectedSize(partitionNames.size());
+ readMvLock();
+ try {
+ for (String partitionName : partitionNames) {
+ MTMVPartitionState state = partitionStates.get(partitionName);
+ if (state != null) {
+ res.put(partitionName, state.getLatestEpoch());
+ }
+ }
+ return res;
+ } finally {
+ readMvUnlock();
+ }
+ }
+
+ /**
+ * Brings the partition states in line with the MV's partitions: every
partition gets an entry, and
+ * every entry whose partition is gone is dropped.
+ *
+ * <p>Alignment is what makes "the partition exists" and "the entry
exists" the same thing, and it is
+ * why an invalidation cannot miss: rows are only written by a refresh,
and every refresh aligns
+ * before it reads a base table, so a partition that holds rows always has
an entry for the mark to
+ * land on. The other direction is what makes the criterion safe -- an
entry created here describes a
+ * partition with no rows yet, so requiring one generation of it discards
no requirement that was
+ * made earlier.
+ *
+ * <p>What it changes is journaled, because the entry has to be on disk
before the rows it describes
+ * can be: a crash between this and the task result would otherwise leave
a partition that holds rows
+ * with no entry at all, and every later invalidation of it would find
nothing to land on. That is the
+ * one shape in which the criterion cannot be read -- "no entry" is
supposed to mean "no rows" -- so
+ * the entry is made durable before any base table is read rather than
derived again on the next run.
+ *
+ * <p>It is deliberately not a hook on every path that creates or drops a
partition. An entry is
+ * derived state, and rebuilding it from the live partition set also
repairs whatever a crash left
+ * behind: the drop of a partition and the removal of its entry are two
journal records, and only
+ * their order -- partition first -- is safe, which leaves at most a stale
entry that the next
+ * alignment drops.
+ *
+ * <p>Only an IVM MV is aligned. For a non-IVM MV the map stays as it is,
and every reader treats
+ * "empty" and "no state" the same.
+ */
+ public void alignPartitionStates(Set<String> livePartitionNames) {
+ if (!isIvm()) {
+ return;
+ }
+ // Copied up front: callers pass what OlapTable holds, and that is
mutated under the table's own
+ // write lock, not this one. Iterating the live collection could see
it change.
+ Set<String> livePartitions = Sets.newHashSet(livePartitionNames);
+ EditLogItem editLogItem = null;
writeMvLock();
try {
- if (ivmInfo == null) {
- ivmInfo = new IvmInfo();
+ boolean changed =
partitionStates.keySet().retainAll(livePartitions);
+ for (String partitionName : livePartitions) {
+ if (!partitionStates.containsKey(partitionName)) {
+ partitionStates.put(partitionName,
MTMVPartitionState.initial());
+ changed = true;
+ }
+ }
+ if (changed) {
+ editLogItem =
submitPartitionStatesChange(Collections.emptySet());
}
- ivmInfo.requireCompleteBaselineRebuild();
- // Bump the version even when a rebuild is already pending, so a
task that started before
- // this visible base-table change cannot clear the barrier with an
old result.
- schemaChangeVersion++;
- editLogItem = submitIvmInfoChange();
} finally {
writeMvUnlock();
}
- editLogItem.await();
+ if (editLogItem != null) {
+ editLogItem.await();
+ }
+ }
+
+ /**
+ * The snapshots of the partitions that are clean after this result's
epochs were applied.
+ *
+ * <p>An invalidation that reached a partition while the task ran leaves
it dirty, and its snapshot
+ * must stay gone: dropping the entry is what keeps transparent rewrite
away from rows the rebuild has
+ * to replace, and a result written back afterwards would undo exactly
that. Removing only the entry
+ * keeps the rest of the map, which the removal on the invalidation side
cannot express.
+ *
+ * <p>The caller holds the MV write lock and has already applied the
epochs, so {@code isDirty} here
+ * reads the state the data is actually described by.
+ *
+ * <p>Only an IVM MV has partition states, so only its write-back is
narrowed here: every entry of a
+ * non-IVM MV has no state to be dirty in and is written back as it always
was.
+ */
+ private Map<String, MTMVRefreshPartitionSnapshot>
snapshotsOfCleanPartitions(
+ Map<String, MTMVRefreshPartitionSnapshot> snapshots) {
+ if (MapUtils.isEmpty(snapshots)) {
+ return snapshots;
+ }
+ Map<String, MTMVRefreshPartitionSnapshot> res =
Maps.newHashMapWithExpectedSize(snapshots.size());
+ for (Entry<String, MTMVRefreshPartitionSnapshot> entry :
snapshots.entrySet()) {
+ MTMVPartitionState state = partitionStates.get(entry.getKey());
+ // No entry means the partition was created after the alignment,
so it can only hold rows this
+ // task wrote; a dirty one needs its rebuild before anything may
read it through the MV.
+ if (state == null || !state.isDirty()) {
+ res.put(entry.getKey(), entry.getValue());
+ }
+ }
+ return res;
+ }
+
+ /**
+ * Records the epochs the given partitions were read at, which is how a
refresh turns a requirement
+ * into the state of the data.
+ *
+ * <p>Only {@code refreshEpoch} is written: a refresh writes back the
requirement it captured, and the
+ * requirement may have been raised again since that capture. A payload
built from the captured map
+ * would overwrite the newer value and lose the rebuild it asks for, so
{@code latestEpoch} is left
+ * alone here.
+ *
+ * <p>The caller holds the MV write lock (it is applied together with the
rest of a task result).
+ */
+ private void applyRefreshedEpochs(Map<String, Long> capturedEpochs) {
+ if (MapUtils.isEmpty(capturedEpochs)) {
+ return;
+ }
+ for (Entry<String, Long> entry : capturedEpochs.entrySet()) {
+ MTMVPartitionState state = partitionStates.get(entry.getKey());
+ if (state == null) {
+ // The partition was dropped while the task ran, so its state
went with it.
+ continue;
+ }
+ state.setRefreshEpoch(entry.getValue());
+ }
+ }
+
+ /**
+ * The states a task result publishes: the partitions whose epochs this
result just wrote.
+ *
+ * <p>Read under the MV write lock, after {@link #applyRefreshedEpochs},
so what it captures is the state
+ * as published. A requirement raised during the task is carried along
rather than recomputed: the
+ * write-back only moves {@code refreshEpoch}, and a payload that omitted
the newer {@code latestEpoch}
+ * would let a replay restore the older one and lose the rebuild it asks
for.
+ */
+ private Map<String, MTMVPartitionState>
publishedPartitionStates(Map<String, Long> capturedEpochs) {
+ if (MapUtils.isEmpty(capturedEpochs)) {
+ return Collections.emptyMap();
+ }
+ Map<String, MTMVPartitionState> published =
Maps.newLinkedHashMapWithExpectedSize(capturedEpochs.size());
+ for (String partitionName : capturedEpochs.keySet()) {
+ MTMVPartitionState state = partitionStates.get(partitionName);
+ if (state != null) {
+ published.put(partitionName, state);
+ }
+ }
+ return published;
+ }
+
+ /**
+ * Invalidates the whole MV: the state the refresh reads, the version bump
that discards a task result
+ * computed against the state being replaced, and the snapshot drop that
stops the transparent rewrite
+ * serving rows from it.
+ *
+ * <p>Applies the change and submits its journal record, and hands back
the write for the caller to
+ * await. The apply happens here, under whatever lock the caller holds,
and before the record is
+ * enqueued, never after: the state is what a concurrent refresh reads,
and it must not become visible
+ * behind the record that stands for it.
+ *
+ * <p>The caller awaits outside the MV lock. It does not have to hold the
lock across the flush to keep
+ * the order -- the record is enqueued in call order, so submitting this
one before the next one is what
+ * puts it first -- and holding the lock across a journal wait is what the
rest of this class avoids.
+ */
+ public EditLogItem invalidateWholeMv(String detail) {
+ MTMVStatus status = new MTMVStatus(MTMVState.SCHEMA_CHANGE, detail);
+ alterStatus(status);
Review Comment:
[P1] Keep the live invalidation and its enqueue in one MV-lock critical
section. `alterStatus` releases `mvRwLock` before this method reaches
`submitAlterLog`, and the callers in `MTMVRelationManager` hold no outer MV
lock. A refresh can start against the bumped schema version, publish/enqueue
`ADD_TASK` in that gap, and leave the leader NORMAL with its snapshot; this
method then enqueues `ALTER_STATUS`, so replay applies the task first and ends
SCHEMA_CHANGE with an empty snapshot. The earlier wait-under-lock thread is
fixed only if apply plus enqueue stay ordered under the lock and callers await
after unlock. Please latch-test this interleaving against replay.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java:
##########
@@ -448,6 +477,16 @@ private List<RefreshAttemptType>
buildAttempts(RefreshRequest request, boolean c
if (shouldUseCompleteForInitialIvmRefresh(containsOneRowRelation)) {
return Lists.newArrayList(RefreshAttemptType.COMPLETE);
}
+ // A schema-level invalidation is not a set of dirty partitions: it
means every partition, including
+ // the ones partition sync has not created yet, and no per-partition
requirement can express that.
+ // IVM only -- a non-IVM MV reaches the same effect through its
cleared snapshot, which its own
+ // refresh already depends on.
+ if (mtmv.isIvm() && !request.explicitPartitions
+ && mtmv.getStatus().getState() == MTMVState.SCHEMA_CHANGE) {
+ LOG.info("IVM MV is in SCHEMA_CHANGE, rebuilding the whole MV,
mv={}, taskId={}",
+ mtmv.getName(), getTaskId());
+ return Lists.newArrayList(RefreshAttemptType.COMPLETE);
Review Comment:
[P1] Preserve strict `PARTITIONS` scope in this shortcut.
`resolveRefreshRequest` carries bare `PARTITIONS` with `allowFallback=false`,
and the normal switch adds `COMPLETE` only for `PARTITIONS FALLBACK`; this
pre-switch return ignores both and rebuilds every partition, including stream
reconciliation/reset, whenever the IVM is in `SCHEMA_CHANGE`. That regresses
the removed complete-baseline check, which told callers to use `PARTITIONS
FALLBACK`, AUTO, or COMPLETE, and the existing stream-unusable test enforces
the same no-scope-expansion contract. Reject bare PARTITIONS here or let it
reach COMPLETE only with fallback, and cover SCHEMA_CHANGE plus both PARTITIONS
forms.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java:
##########
@@ -259,7 +260,25 @@ private PartitionPlanningException(String message,
Throwable cause) {
// Written by the executing (Disruptor worker) thread via the
executeCommand consumer
// callback and read by the cancel (command) thread, so it must be
volatile.
private volatile StmtExecutor executor;
- private Map<String, MTMVRefreshPartitionSnapshot> partitionSnapshots;
+ // What this task has committed, per MV partition: the snapshot each
partition's rows were read at.
+ // One accumulator for the whole task rather than one per phase, because
that is what the MV publishes
+ // at the end of it -- a phase that started from empty would publish its
own work and drop the work of
+ // the phases before it, leaving partitions a preceding rebuild replaced
looking unsynced.
+ private Map<String, MTMVRefreshPartitionSnapshot> partitionSnapshots =
Maps.newConcurrentMap();
+ // The requirement each refreshed partition was read under, captured
before the base tables were read
+ // and recorded only once that batch's data committed (see
commitCapturedEpochs). In memory only: the
+ // journal carries the resulting states, and a replay applies those
instead of recomputing anything.
+ private transient Map<String, Long> ivmCapturedEpochs = Maps.newHashMap();
Review Comment:
[P1] Do not publish this accumulator from cancellation while the worker can
still mutate it. STOP uses `cancel(false)`, immediately calls `after()`, and
neither the task monitor nor the worker-only job lock covers
`commitCapturedEpochs`; after `Command.run` the executor may already be cleared
while the worker is adding keys here. `MTMV.addTaskResult` then iterates this
live `HashMap`, so a multi-partition commit can add a key during cancellation's
iteration, throwing `ConcurrentModificationException` or partially changing
live epochs without an ADD_TASK journal; cleanup also nulls resources while the
worker continues. Make the worker own terminal publication/cleanup after
quiescence, or synchronize and publish a detached map snapshot, and latch-test
STOP during a multi-key merge.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java:
##########
@@ -343,6 +367,11 @@ public void run() throws JobException {
break;
case COMPLETE:
executeCompleteAttempt(refreshContext, ctx);
+ // Recorded here rather than where the escalation was
decided: the count is what the
+ // rebuild actually replaced, and a refresh that
failed before its first commit must
+ // not report the whole MV as rebuilt. The rebuild
records its own count for the
+ // partitions it replaced; this one is only reached
when it succeeded.
+ recordRebuiltPartitions(request,
mtmv.getPartitionNames().size());
Review Comment:
[P2] Count committed COMPLETE batches even when a later batch fails.
`executePartitionBasedRefresh` publishes each successful group's
partitions/snapshots/epochs, and `addTaskResult` intentionally preserves them
on a FAILED task, but this call is skipped unless the entire COMPLETE returns.
An escalated multi-group refresh can therefore rebuild and publish its early
groups while `IvmRebuiltPartitions` remains 0, hiding the partial side effect
this new diagnostic is meant to expose. Mirror the dirty-rebuild path's
`finally` and count `completedPartitions` while retaining the explicit-COMPLETE
exemption, with a later-group failure test.
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java:
##########
@@ -370,102 +410,142 @@ public boolean addTaskResult(AlterMTMV alterMTMV,
boolean isReplay) {
public void alterMvProperties(AlterMTMV alterMTMV, boolean isReplay) {
EditLogItem editLogItem;
+ EditLogItem invalidation = null;
writeMvLock();
try {
Map<String, String> mvProperties = alterMTMV.getMvProperties();
- boolean containsExcludedTriggerTables = mvProperties.containsKey(
- PropertyAnalyzer.PROPERTIES_EXCLUDED_TRIGGER_TABLES);
- Set<TableNameInfo> oldExcludedTriggerTables =
containsExcludedTriggerTables
- ? parseExcludedTriggerTables()
- : Sets.newHashSet();
- // Enlarging or removing ivm_partition_window_limit brings
previously lossy
- // partitions back into the refresh range. Their stream backlog
was skipped by
- // the windowed refreshes, so a strict incremental refresh would
wrongly judge
- // "all partitions are synced" and return SUCCESS with stale data.
Force the
- // next refresh to rebuild a complete baseline instead.
- boolean containsPartitionWindowLimit = mvProperties.containsKey(
- PropertyAnalyzer.PROPERTIES_IVM_PARTITION_WINDOW_LIMIT);
- 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;
+ // Read the old values before the properties are applied, and
unconditionally: a property that is
+ // not part of this ALTER has to be compared against the value the
MV actually holds. Reading it
+ // only when its key is present would compare an empty default
against the real value, report a
+ // change that is not there, and drop the snapshot of an unrelated
ALTER.
+ Set<TableNameInfo> oldExcludedTriggerTables =
parseExcludedTriggerTables();
+ Map<TableNameInfo, Integer> oldWindowLimits =
+
MTMVPropertyUtil.getIvmPartitionWindowLimit(this.mvProperties);
+ Map<String, String> oldSyncWindow =
MTMVPropertyUtil.partitionSyncWindowOf(this.mvProperties);
this.mvProperties.putAll(mvProperties);
- // Both excluded_trigger_tables changes and window limit
enlargement/removal
- // change the refresh baseline semantics: partitions previously
skipped become
- // refreshable again, and their stream backlog was not applied.
Invalidate the
- // snapshots (once) and require a complete baseline rebuild so the
next refresh
- // covers the new range instead of wrongly judging "all partitions
are synced".
- boolean invalidateRefreshSnapshot = false;
- boolean requireCompleteBaselineRebuild = false;
- if (containsExcludedTriggerTables) {
- Set<TableNameInfo> newExcludedTriggerTables =
parseExcludedTriggerTables();
- if
(!oldExcludedTriggerTables.equals(newExcludedTriggerTables)) {
- invalidateRefreshSnapshot = true;
- if (ivmInfo != null && ivmInfo.isEnableIvm()
- && relation != null && relation.getBaseTables() !=
null) {
- for (BaseTableInfo baseTableInfo :
relation.getBaseTables()) {
- TableNameInfo baseTableName = new
TableNameInfo(baseTableInfo.getCtlName(),
- baseTableInfo.getDbName(),
baseTableInfo.getTableName());
- if
(MTMVPartitionUtil.isTableExcluded(oldExcludedTriggerTables, baseTableName)
- &&
!MTMVPartitionUtil.isTableExcluded(newExcludedTriggerTables, baseTableName)) {
- requireCompleteBaselineRebuild = true;
- break;
- }
- }
- }
- }
- }
- if (containsPartitionWindowLimit && ivmInfo != null &&
ivmInfo.isEnableIvm()
- && relation != null && relation.getBaseTables() != null) {
- Map<TableNameInfo, Integer> newWindowLimits =
-
MTMVPropertyUtil.getIvmPartitionWindowLimit(this.mvProperties);
- for (BaseTableInfo baseTableInfo : relation.getBaseTables()) {
- TableNameInfo baseTableName = new
TableNameInfo(baseTableInfo.getCtlName(),
- baseTableInfo.getDbName(),
baseTableInfo.getTableName());
- int oldLimit =
MTMVPropertyUtil.getPartitionWindowLimit(oldWindowLimits, baseTableName);
- if (oldLimit == -1) {
- continue;
- }
- int newLimit =
MTMVPropertyUtil.getPartitionWindowLimit(newWindowLimits, baseTableName);
- if (newLimit == -1 || newLimit > oldLimit) {
- requireCompleteBaselineRebuild = true;
- break;
- }
- }
- }
- if (containsSyncWindow && ivmInfo != null && ivmInfo.isEnableIvm()
- &&
MTMVPropertyUtil.partitionSyncWindowWidens(oldSyncWindow,
-
MTMVPropertyUtil.partitionSyncWindowOf(this.mvProperties))) {
- requireCompleteBaselineRebuild = true;
- }
- if (invalidateRefreshSnapshot || requireCompleteBaselineRebuild) {
- this.schemaChangeVersion++;
- this.refreshSnapshot = new MTMVRefreshSnapshot();
- }
- if (requireCompleteBaselineRebuild) {
- ivmInfo.requireCompleteBaselineRebuild();
- }
+ // The one thing a property change can owe the refresh baseline: a
whole-MV rebuild, when it
+ // brings base table partitions back into the set the MV
maintains. Their stream backlog was
+ // skipped while they were outside that set, so no delta can
repair them -- and the rebuild is
+ // whole-MV rather than per-partition because it covers the
partitions partition sync has not
+ // created yet. invalidateWholeMv owns all of it: the state the
refresh reads, the version bump
+ // that discards a task result computed before the change, and the
snapshot drop that stops
+ // transparent rewrite serving rows from it.
+ //
+ // Narrowing that set owes nothing. The MV's rows for a table it
no longer maintains are allowed
+ // to be stale by design, and the snapshot entry describing them
is skipped by the next
+ // incremental refresh anyway, so dropping the whole snapshot and
discarding a running task
+ // result for them buys nothing. A change that leaves the
maintained set alone owes nothing
+ // either.
if (isReplay) {
+ // The property change itself is applied above. Nothing else
on this path has to run for a
+ // replay: the state, the version and the snapshot a whole-MV
invalidation moves come back
+ // from the status record that precedes this one, through
MTMV#alterStatus, and this
+ // property record never carried a snapshot.
return;
}
+ if (rebuildsWholeMv(oldExcludedTriggerTables, oldWindowLimits,
oldSyncWindow)) {
+ // Journaled on its own record, ahead of the property change
below; a replay applies both
+ // in that order. Submitted here and awaited below, outside
the lock: the order is the
+ // enqueue order, which the lock already fixes, so there is
nothing to gain by holding the
+ // lock across the flush.
+ invalidation = invalidateWholeMv("The MV's refresh baseline
changed with its properties");
+ }
editLogItem = submitAlterLog(alterMTMV);
} finally {
writeMvUnlock();
}
+ if (invalidation != null) {
+ invalidation.await();
+ }
editLogItem.await();
}
+ /**
+ * Whether a property change brings base table partitions back into the
set the MV maintains, and so
+ * owes a whole-MV rebuild.
+ *
+ * <p>Takes the values the MV held before the change; see the call site
for why they are read
+ * unconditionally. Widening decides on its own: a change that both takes
a partition out of the
+ * maintained set and puts one back is the rebuild, because the partition
coming back is the one whose
+ * backlog was skipped.
+ */
+ private boolean rebuildsWholeMv(Set<TableNameInfo>
oldExcludedTriggerTables,
+ Map<TableNameInfo, Integer> oldWindowLimits, Map<String, String>
oldSyncWindow) {
+ return unexcludesABaseTable(oldExcludedTriggerTables)
+ || widensPartitionWindowLimit(oldWindowLimits)
+ || widensSyncWindow(oldSyncWindow);
+ }
+
+ /**
+ * Whether this MV has an IVM baseline to maintain at all, which every
widening check needs.
+ *
+ * <p>No null check on {@code ivmInfo}: it is initialized where an MV is
built and
+ * {@link #gsonPostProcess()} gives an MV loaded from an image written
before the field existed the
+ * same one, so it is non-null by the time anything reads it.
+ */
+ private boolean maintainsIvmBaseline() {
+ return ivmInfo.isEnableIvm() && relation != null &&
relation.getBaseTables() != null;
+ }
+
+ /**
+ * Whether a base table of this MV stopped being excluded.
+ *
+ * <p>An excluded table has no stream, so the partitions the MV read from
it have no backlog to apply;
+ * while it was excluded the MV did not maintain them.
+ */
+ private boolean unexcludesABaseTable(Set<TableNameInfo>
oldExcludedTriggerTables) {
+ if (!maintainsIvmBaseline()) {
+ return false;
+ }
+ Set<TableNameInfo> newExcludedTriggerTables =
parseExcludedTriggerTables();
+ for (BaseTableInfo baseTableInfo : relation.getBaseTables()) {
+ TableNameInfo baseTableName = new
TableNameInfo(baseTableInfo.getCtlName(),
+ baseTableInfo.getDbName(), baseTableInfo.getTableName());
+ if (MTMVPartitionUtil.isTableExcluded(oldExcludedTriggerTables,
baseTableName)
+ &&
!MTMVPartitionUtil.isTableExcluded(newExcludedTriggerTables, baseTableName)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Whether an ivm_partition_window_limit was removed or enlarged for some
base table, which brings the
+ * partitions the windowed refreshes skipped back into range with their
backlog unapplied.
+ */
+ private boolean widensPartitionWindowLimit(Map<TableNameInfo, Integer>
oldWindowLimits) {
+ if (!maintainsIvmBaseline()) {
+ return false;
+ }
+ Map<TableNameInfo, Integer> newWindowLimits =
+ MTMVPropertyUtil.getIvmPartitionWindowLimit(this.mvProperties);
+ for (BaseTableInfo baseTableInfo : relation.getBaseTables()) {
+ TableNameInfo baseTableName = new
TableNameInfo(baseTableInfo.getCtlName(),
+ baseTableInfo.getDbName(), baseTableInfo.getTableName());
+ int oldLimit =
MTMVPropertyUtil.getPartitionWindowLimit(oldWindowLimits, baseTableName);
+ if (oldLimit == -1) {
+ continue;
+ }
+ int newLimit =
MTMVPropertyUtil.getPartitionWindowLimit(newWindowLimits, baseTableName);
+ if (newLimit == -1 || newLimit > oldLimit) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Whether a partition_sync_limit window was widened.
+ *
+ * <p>A window that starts applying, a narrower one, and one that
describes the same set as before all
+ * 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.
+ */
+ private boolean widensSyncWindow(Map<String, String> oldSyncWindow) {
Review Comment:
[P2] Preserve the IVM guard on sync-window widening. The removed code
required `ivmInfo.isEnableIvm()` for this exact comparison, and both sibling
widening helpers now call `maintainsIvmBaseline()`, but this helper does not.
Consequently widening/removing `partition_sync_limit` on an ordinary MTMV
enters `invalidateWholeMv`, clears every refresh snapshot, and makes the next
refresh recompute already-maintained partitions instead of only admitting the
newly in-window ones. Gate this helper like the siblings and cover a non-IVM
widen/removal.
--
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]
