yujun777 commented on code in PR #68390:
URL: https://github.com/apache/doris/pull/68390#discussion_r4091533911
##########
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:
Fixed in 8934d283cc4, and it was this PR's hole rather than the shortcut's:
the refusal this PR added was judged after
`shouldUseCompleteForInitialIvmRefresh`, so in exactly the state the refusal
names -- an IVM MV in SCHEMA_CHANGE that has no snapshot yet and reads an
excluded trigger table or a one-row relation -- the shortcut answered COMPLETE
and the request never heard that its scope was not the one it asked for. The
refusal is judged first now. Both branches answer COMPLETE, so the only request
whose outcome changes is the one that was supposed to be told no. The unit test
asserts both halves: that the shortcut is active in that state, and that the
request is refused rather than widened.
##########
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 8934d283cc4, with the caveat that the path itself predates this PR:
`git blame` puts `MTMVRelationManager`'s `Env.alterMTMVStatus` call at
2023-11-24, and the "apply inside `alterStatus`, enqueue in
`Alter.processAlterMTMV`" split with it. What this PR introduced is the
`invalidateWholeMv` shape, so the fix is to route the generic path through it
-- which is what the sibling invalidation in the same loop already did -- and
to delete `Env.alterMTMVStatus`, whose only caller this was.
Checked with it: same state, same detail string, same journal op and a
byte-identical payload (an `AlterMTMV(name, ALTER_STATUS)` carrying the status,
`needRebuildJob` false in both, so neither called `alterJob`), and the replay
path is untouched. The one thing it does move is that the DDL now awaits the
record instead of enqueueing and returning, which is what the sibling branches
in that loop already do. No replay path reaches this hook
(`replayDropTable`/`replayAlterTable` do not call `MTMVService`), so the await
cannot land on the journal thread.
--
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]
