github-actions[bot] commented on code in PR #68390:
URL: https://github.com/apache/doris/pull/68390#discussion_r4073338782
##########
fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelationManager.java:
##########
@@ -486,6 +484,15 @@ private void processBaseTableChange(BaseTableInfo
baseTableInfo, String msgPrefi
if (checkIvmQueryUsable) {
invalidateIvmBaselineIfQueryUnusable(baseTableInfo, mtmv);
}
+ if (renamed && mtmv instanceof MTMV && ((MTMV) mtmv).isIvm()) {
+ // A rename leaves every column alone, and the failure it does
cause -- the MV query still
+ // spells the old name -- is reported by the refresh itself:
it resolves the base tables from
+ // the query (MTMVTask#run) before it looks at anything else,
so the state is not what makes
+ // that failure visible. What the state does to an IVM MV is
make the next refresh rebuild the
+ // whole MV (MTMVTask#buildAttempts), which a rename back
would have it repeat for nothing. A
+ // non-IVM MV keeps the state it has always got, which is what
its own refresh reads.
+ continue;
Review Comment:
[P1] Preserve the dependency mapping across this rename. BaseTableInfo
equality is name-based, but this branch neither invalidates the IVM nor re-keys
tableMTMVsOneLevelAndFromView from oldTableInfo to newTableInfo. After t is
renamed to tmp, metadata-only DDL such as TRUNCATE on tmp therefore finds no
dependent MV; renaming tmp back also looks up only tmp and misses. The query is
usable again, yet no dirty epoch exists and the delta stream cannot remove the
truncated rows. Please move/alias the dependency entry on rename or keep a
conservative invalidation, with a rename -> TRUNCATE -> rename-back regression.
##########
fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPartitionState.java:
##########
@@ -61,6 +61,43 @@ public MTMVPartitionState(MTMVPartitionState other) {
this.latestEpoch = other.latestEpoch;
}
+ /**
+ * The state a partition gets when it is first aligned: never refreshed,
one generation required.
+ *
+ * <p>Alignment only ever creates this pair, so "no entry yet" and "this
pair" say the same thing
+ * about the past -- the partition was never marked and holds no rows.
+ */
+ public static MTMVPartitionState initial() {
+ return new MTMVPartitionState(0, 1);
+ }
+
+ /**
+ * Whether this partition holds rows that a metadata-only change of a base
table has made unusable,
+ * so it has to be rebuilt rather than caught up incrementally.
+ *
+ * <p>{@code latestEpoch > refreshEpoch} alone does not say that: {@code
refreshEpoch == 0} means the
+ * partition was never refreshed, so it holds no rows and its first
refresh reads the current base
+ * tables anyway -- it cannot carry in rows that no longer exist. Calling
that dirty would rebuild
+ * every partition of a fresh MV for nothing.
+ *
+ * <p>The rejected alternative asked the partition itself instead ({@code
visibleVersion == 1} means
+ * "no data", and a rebuild would put {@code refreshEpoch} back to 0 for
such a partition). It is
+ * better in one way -- it does not rebuild a partition that is empty
because its base partitions are
+ * empty -- but it leaks silently as soon as an anti-join, {@code NOT IN}
or {@code NOT EXISTS}
+ * incremental refresh exists: removing input rows does not always remove
output rows (a left join
+ * filtered by {@code b.v IS NULL} becomes non-empty when the matching
rows go away), and that change
+ * emits no row binlog. Betting the criterion on today's operator support
is exactly what this state
+ * exists to avoid, so the conservative reading wins.
+ */
+ public boolean isDirty() {
+ return latestEpoch > refreshEpoch && refreshEpoch != 0;
Review Comment:
[P1] Do not equate refreshEpoch == 0 with durable emptiness. Alignment
journals (0,1) before the first batch, but the MV DML commits before the task
publishes its epoch/snapshot and before ADD_TASK is journaled. An FE crash in
that cut (or cancel(true) after the executor callback clears executor but
before batch publication) leaves committed rows behind replayed (0,1). A later
TRUNCATE raises it to (0,2), this returns false, and strict incremental can
accept an empty delta and mark the stale rows clean. Persist an
in-progress/possibly-populated state before writing, or otherwise make batch
commit and durable epoch publication recover as one lifecycle.
##########
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:
[P2] Journal the same filtered snapshot map that is applied live. When an
invalidation lands during a task, the leader deliberately omits that dirty
partition here, but alterMTMV still contains the raw task map. Replay restores
the journaled dirty states, skips this !isReplay filter, and re-adds that
snapshot, so restart does not reconstruct the leader's state or preserve the
documented removal invariant. With all snapshot-bearing partitions dirty this
also changes attempt selection from the no-snapshot COMPLETE branch (rebuilt
count 0) to the all-dirty COMPLETE branch (count N). Put a detached
snapshotsToWrite on the payload before submitAlterLog and cover
invalidation-before-ADD_TASK replay order.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java:
##########
@@ -768,6 +749,9 @@ private IvmIncrRefreshResult
executeSingleIvmAttempt(MTMVRefreshContext refreshC
} catch (Exception e) {
throw new JobException("IVM snapshot generation failed for mv=" +
mtmv.getName(), e);
}
+ // The requirement these partitions are read under, captured before
the read inside doRefresh and
+ // recorded only if the refresh commits; see captureLatestEpochs.
+ Map<String, Long> capturedEpochs =
captureLatestEpochs(Sets.newHashSet(needRefreshPartitions));
Review Comment:
[P1] Recheck invalidation after choosing the rebuild set. A TRUNCATE can
raise this partition's epoch and remove its snapshot after executeIvmAttempt
sampled dirtyPartitions. It then enters incrementalScope, this call captures
the raised epoch, and the row-delta refresh cannot delete the rows removed only
through metadata. Success consequently writes refreshEpoch == latestEpoch and
leaves those old MV rows permanently clean. Dirty selection and epoch capture
need one generation decision (or an epoch advance here must abort/reroute the
partition to rebuild); please add a latch test for the sample -> TRUNCATE ->
capture ordering.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java:
##########
@@ -312,15 +321,20 @@ public void run() throws JobException {
// refresh fallback: incompatible MV definitions must fail
directly.
ensureQueryUsableIfNeeded(ctx, tableIfs);
RefreshRequest request = resolveRefreshRequest();
- validateIvmBaselineBeforePartitionSync(request);
- List<RefreshAttemptType> attempts = buildAttempts(request,
queryAnalysis.containsOneRowRelation());
try {
syncPartitionsIfNeeded(ctx, tableIfs);
} catch (PartitionPlanningException e) {
throw new JobException(e.getMessage(), e);
}
+ // Partition sync has decided which partitions exist, and nothing
has read a base table yet:
+ // this is the point where an entry and the partition it describes
become the same thing.
+ // Doing it any later would let a partition that sync has just
added be refreshed without an
+ // entry, and an invalidation arriving in between would have
nothing to land on.
+ mtmv.alignPartitionStates(mtmv.getPartitionNames());
Review Comment:
[P1] Avoid making every later ADD_TASK carry this full aligned map. The
existing task-result path deep-copies partitionStates under mvRwLock and
serializes the whole map as JSON, but before this change the production map was
never populated. This line now creates one entry per MV partition, so even a
no-op scheduled refresh or a task advancing one partition emits O(total
partitions) state. Doris already exercises 160,000 mapped MV partitions, making
each periodic result a multi-megabyte record and lock-held copy. Persist only
the task's detached epoch delta on ADD_TASK (and omit it when empty), keeping
full maps for alignment/invalidation records.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java:
##########
@@ -448,6 +462,17 @@ 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());
+ recordRebuiltPartitions(request, mtmv.getPartitionNames().size());
+ return Lists.newArrayList(RefreshAttemptType.COMPLETE);
Review Comment:
[P2] Publish the rebuilt plan signature on this direct COMPLETE path.
SCHEMA_CHANGE can come from altering a base view while keeping the MV output
schema valid but changing its normalized join/layout plan. Because this branch
skips the incremental attempt, ivmFallbackReason is never
PLAN_SIGNATURE_MISMATCH; executePartitionBasedRefresh discards the signature
produced by this successful rebuild and ADD_TASK keeps the old one. The next
AUTO refresh then performs a second COMPLETE through mismatch fallback, while a
next strict INCREMENTAL rejects the baseline it just rebuilt. Capture/persist
the consistent full-refresh signature when this branch establishes the new
baseline.
##########
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());
Review Comment:
[P2] Count rebuilt partitions as batches commit. If an early dirty-rebuild
group commits and a later group fails, this line is never reached and
IvmRebuiltPartitions stays 0 even though ADD_TASK preserves the committed
group's epochs/data. The COMPLETE escalation sites have the inverse problem:
they record every planned partition before the first batch, so an immediate
failure reports all rebuilt. The new diagnostic should reflect successful
groups, especially on failed tasks.
##########
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:
[P2] Keep the progress denominator consistent when merging the rebuild
phase. executeSingleIvmAttempt resets needRefreshPartitions to only the
incremental scope and clears completedPartitions; this line then adds the dirty
rebuilds only to the completed side. A task with one rebuilt and one
incremental partition is recorded as 200% (2/1), while a dirty-only success
reports null progress because the denominator is empty. Preserve the union of
both phase scopes (and their completed sets) for task history.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java:
##########
@@ -448,6 +462,17 @@ 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
Review Comment:
[P2] Distinguish whole-IVM invalidation from generic SCHEMA_CHANGE here.
processBaseTableChange sets this state even when re-analysis succeeds and no
partition epoch was raised (for example DROP COLUMN spare, which the changed
test says must not invalidate the baseline). This branch nevertheless converts
the following strict INCREMENTAL request into a whole COMPLETE rebuild; the
test checks only SUCCESS, so it misses that route. Use a dedicated
whole-baseline marker or avoid setting this state for compatible IVM changes,
and assert RefreshMode/IvmRebuiltPartitions.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java:
##########
@@ -567,32 +633,13 @@ private MTMVRefreshContext
buildRefreshContext(List<TableIf> tableIfs) throws An
}
}
- /**
- * Makes the barrier that says "these MV partitions must be rebuilt before
their IVM offsets may be
- * used again" durable. Every caller writes it as soon as it has decided
the partition set and
- * before anything that touches MV data or base table streams, so that a
crash or a rejection can
- * only ever leave a barrier with no rebuild behind it, which merely costs
one extra rebuild, and
- * never a rebuild with no barrier, which silently loses rows.
- */
- private void writeIvmBaselineBarrier(RefreshMode refreshMode) throws
JobException {
- if (mtmv.isIvm()) {
- // Persist the guard before the first baseline data transaction.
- mtmv.persistIvmBaselineGuard(refreshMode,
Sets.newHashSet(needRefreshPartitions),
- mtmvSchemaChangeVersion);
- }
- }
-
private void executeCompleteAttempt(MTMVRefreshContext context,
ConnectContext ctx)
throws JobException, AnalysisException {
this.needRefreshPartitions =
Lists.newArrayList(mtmv.getPartitionNames());
this.refreshMode = generateRefreshMode(needRefreshPartitions);
if (refreshMode == MTMVTaskRefreshMode.NOT_REFRESH) {
return;
}
- // The barrier goes first: a stream this rebuild reconciles carries
the base table's current
- // rows as its initial snapshot, and a later incremental refresh that
consumed it as a delta
- // against data still built from the old baseline would double-count
them.
- writeIvmBaselineBarrier(RefreshMode.COMPLETE);
Review Comment:
[P1] Keep a durable rebuild requirement before recreating an IVM stream. A
dropped/unusable stream sends a fallback refresh to COMPLETE, but
reconcileIvmStreams durably creates its replacement with show_initial_rows=true
before any MV rebuild batch. If FE crashes after that create and before the
rebuild/ADD_TASK, restart has the old populated MV with clean nonzero epochs
plus a usable stream whose historical rows are exposed as APPEND; the next
AUTO/INCREMENTAL refresh can therefore add those rows to the old baseline
again. This removed barrier protected exactly that cut. Persist a
whole-MV/affected-partition requirement before reconciliation and clear it only
after the rebuilt baseline is durably published, with a failover test at this
boundary.
--
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]
