yujun777 commented on code in PR #68390:
URL: https://github.com/apache/doris/pull/68390#discussion_r4089232420
##########
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:
Fixed in the branch: `invalidateWholeMv` now holds the MV write lock across
both the apply and the enqueue (`alterStatus` re-enters the same lock), so a
task result cannot be enqueued between them; the callers that hold no outer
lock are the ones this is for, and all of them still await after the unlock.
##########
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:
Fixed, and the guard moved rather than being copied three times: the three
widening checks are now judged together in `rebuildsWholeMv`, under one
`maintainsIvmBaseline()`, since "did this property move in the direction that
owes a rebuild" is only a question an MV maintaining an IVM baseline has.
`IvmBaselineRebuildTest#testWideningTheSyncWindowDoesNotInvalidateANonIvmMv`
covers a plain MV widening `partition_sync_limit` -- and asserts the property
itself still applies, since it is the rebuild it owes that is IVM's, not the
property.
--
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]
