yujun777 commented on code in PR #68390:
URL: https://github.com/apache/doris/pull/68390#discussion_r4083017314
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java:
##########
@@ -342,7 +347,14 @@ public boolean addTaskResult(AlterMTMV alterMTMV, boolean
isReplay) {
}
this.jobInfo.addHistoryTask(task);
compatiblePctSnapshot(partitionSnapshots);
- this.refreshSnapshot.updateSnapshots(partitionSnapshots,
getPartitionNames());
+ // What this task wrote is described by the epochs just recorded,
so a partition the result
+ // left dirty is left out: its snapshot would otherwise come back
after an invalidation
+ // dropped it, and transparent rewrite reads that map to decide
what it may serve.
+ Map<String, MTMVRefreshPartitionSnapshot> snapshotsToWrite =
partitionSnapshots;
+ if (!isReplay && ivmInfo.isEnableIvm()) {
+ snapshotsToWrite =
snapshotsOfCleanPartitions(partitionSnapshots);
Review Comment:
Implemented. `MTMV#addTaskResult` filters the map once into
`snapshotsToWrite` and uses that one map for both the live
`refreshSnapshot.updateSnapshots(...)` and the journaled payload, so a replay
restores what the leader has rather than the raw task map that still names the
dirty partition.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java:
##########
@@ -715,13 +664,36 @@ private AttemptResultType
executeIvmAttempt(MTMVRefreshContext refreshContext,
+ "Continuing with COMPLETE refresh.", mtmv.getName(),
getTaskId());
return AttemptResultType.FALLBACK_TO_COMPLETE;
}
+ // The partitions the criterion says must be rebuilt rather than
caught up: the delta path can only
+ // append, so a partition it treated as current would record that in
its epoch while its rows still
+ // come from before the change. Rebuilt first, with the partition
executor, because that is the
+ // full recomputation they need -- and only in this task's batches, so
a change that arrives while
+ // it runs leaves them dirty for the next round instead of being
swallowed.
+ Set<String> dirtyPartitions = mtmv.getDirtyPartitions();
+ Map<String, MTMVRefreshPartitionSnapshot> rebuiltSnapshots =
Maps.newHashMap();
+ if (!dirtyPartitions.isEmpty()) {
+ LOG.info("Rebuilding {} invalidated MV partitions before the
incremental refresh, mv={}, taskId={}",
+ dirtyPartitions.size(), mtmv.getName(), getTaskId());
+ List<String> toRebuild = Lists.newArrayList(dirtyPartitions);
+ toRebuild.sort(Comparator.naturalOrder());
+ this.needRefreshPartitions = toRebuild;
+ this.refreshMode = generateRefreshMode(toRebuild);
+ executePartitionBasedRefresh(refreshContext,
RefreshMode.PARTITIONS, ctx);
+ rebuiltSnapshots.putAll(partitionSnapshots);
+ recordRebuiltPartitions(request, dirtyPartitions.size());
+ }
MTMVRefreshContext currentRefreshContext = refreshContext;
int ivmAttemptLimit = Math.max(Config.max_query_retry_time, 0) + 1;
IvmIncrRefreshResult ivmResult = null;
for (int partitionSyncRetryCount = 0;
partitionSyncRetryCount < ivmAttemptLimit;
partitionSyncRetryCount++) {
- ivmResult = executeSingleIvmAttempt(currentRefreshContext);
+ ivmResult = executeSingleIvmAttempt(currentRefreshContext,
dirtyPartitions);
if (ivmResult.isSuccess()) {
+ // The incremental attempt reset the accumulators it owns, so
the rebuild's are merged back
+ // here: its batches committed, and without them the
partitions it rebuilt would look
+ // unsynced and be refreshed again on every following round.
+ this.partitionSnapshots.putAll(rebuiltSnapshots);
+ this.completedPartitions.addAll(dirtyPartitions);
Review Comment:
Implemented. The rebuild phase keeps its scope and its committed partitions
before the incremental attempt resets them, and the successful path publishes
the union of the two scopes as the denominator while keeping what each phase
completed, so a rebuilt-plus-caught-up refresh reports one of each rather than
two of one.
--
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]
