github-actions[bot] commented on code in PR #68390:
URL: https://github.com/apache/doris/pull/68390#discussion_r4080257032
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java:
##########
@@ -449,8 +475,11 @@ public void alterMvProperties(AlterMTMV alterMTMV, boolean
isReplay) {
this.schemaChangeVersion++;
this.refreshSnapshot = new MTMVRefreshSnapshot();
}
- if (requireCompleteBaselineRebuild) {
- ivmInfo.requireCompleteBaselineRebuild();
+ if (requireCompleteBaselineRebuild && !isReplay) {
+ // Every partition has to be rebuilt, including the ones this
MV does not have yet, so the
+ // MV goes into the state that says exactly that. Journaled on
its own record, ahead of the
+ // property change below; a replay applies both in that order.
+ invalidateWholeMv("The MV's refresh baseline changed with its
properties");
Review Comment:
[P2] Do not wait for this status journal while holding mvRwLock.
invalidateWholeMv re-enters alterStatus and then processAlterMTMV calls
synchronous logAlterMTMV; batch mode waits for the edit-log worker and direct
mode performs the write here. A slow journal therefore blocks every reader,
invalidation, and task-result publication for this MV, contrary to the
submit-under-lock/await-after-unlock pattern used by the surrounding property
record. Enqueue the ordered status and property records under the lock, then
await both after releasing it.
##########
fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelationManager.java:
##########
@@ -359,16 +362,48 @@ public void alterTable(BaseTableInfo oldTableInfo,
Optional<BaseTableInfo> newTa
// when replace, need deal two table
if (isReplace) {
// REPLACE TABLE already invalidates the IVM baseline explicitly,
see Alter#processReplaceTable
- processBaseTableChange(newTableInfo.get(), "The base table has
been updated:", false);
+ processBaseTableChange(newTableInfo.get(), "The base table has
been updated:", false, false);
}
- // A RENAME leaves every column alone, and the failure it does cause
-- the MV query still
- // spells the old name -- is already reported by the refresh itself
(MTMVTask#run resolves
- // the base tables from the query before it ever looks at the
baseline). Invalidating here
- // would only leave a stale flag behind: rename the table back and the
query is analyzable
- // again, yet every strict INCREMENTAL refresh would stay rejected
until a COMPLETE one ran.
boolean renamed = !isReplace && newTableInfo.isPresent()
&& !Objects.equals(oldTableInfo.getTableName(),
newTableInfo.get().getTableName());
- processBaseTableChange(oldTableInfo, "The base table has been
updated:", !renamed);
+ // The invalidation runs first, while the dependencies are still
registered under the name the
+ // rename is leaving: moving them first would make this lookup --
which is by the old name -- find
+ // nothing, and the rename would stop invalidating anything at all.
+ processBaseTableChange(oldTableInfo, "The base table has been
updated:", !renamed, renamed);
+ if (renamed) {
+ renameBaseTable(oldTableInfo, newTableInfo.get());
+ }
+ }
+
+ /**
+ * Move a renamed table's entries in the dependency maps to its new name.
+ *
+ * <p>The maps are keyed by {@link BaseTableInfo}, which compares by name,
and an MV keeps the relation
+ * it was created against -- a rename leaves the MV query spelling the old
name, so it no longer
+ * analyzes and the relation is not recomputed. Without this the maps
would keep the old name, and a
+ * metadata-only change to the table under its new name -- a TRUNCATE,
say, which emits no row binlog --
+ * would find no dependent MV to invalidate. Renaming the table back then
restores an analyzable query
+ * whose MV still holds the rows that change removed, and nothing names
the partition that would have
+ * to be rebuilt. Moving the entries is what a rename needs instead of the
invalidation it used to
+ * carry: a rename changes no rows, so there is nothing to rebuild, only a
lookup that has to keep
+ * working.
+ */
+ private void renameBaseTable(BaseTableInfo oldTableInfo, BaseTableInfo
newTableInfo) {
+ moveRelationKey(tableMTMVs, oldTableInfo, newTableInfo);
+ moveRelationKey(tableMTMVsOneLevelAndFromView, oldTableInfo,
newTableInfo);
+ }
+
+ private void moveRelationKey(Map<BaseTableInfo, Set<BaseTableInfo>> map,
+ BaseTableInfo oldTableInfo, BaseTableInfo newTableInfo) {
+ Set<BaseTableInfo> dependents = map.get(oldTableInfo);
+ if (CollectionUtils.isEmpty(dependents)) {
+ return;
+ }
+ // Registered under the new name before the old one is dropped: a
concurrent base-table change
+ // either still finds the old name or already finds the new one, never
neither. Merged rather than
+ // replaced, because a table dropped and re-created under this name
registers its own dependents.
+ map.computeIfAbsent(newTableInfo, key ->
Sets.newConcurrentHashSet()).addAll(dependents);
Review Comment:
[P1] Fence refresh relations captured before this rename. A task can finish
its DML with relation keyed by t, then t -> tmp moves the live key here without
advancing the IVM schema generation. The later accepted ADD_TASK calls
refreshComplete with that stale relation; refreshMTMVCache re-adds t and prunes
tmp. TRUNCATE tmp is then invisible, and renaming back exposes stale rows.
Reject/translate pre-rename task relations with a dependency generation, and
add a latch test for task return -> rename -> result publication -> TRUNCATE.
##########
fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelationManager.java:
##########
@@ -359,16 +362,48 @@ public void alterTable(BaseTableInfo oldTableInfo,
Optional<BaseTableInfo> newTa
// when replace, need deal two table
if (isReplace) {
// REPLACE TABLE already invalidates the IVM baseline explicitly,
see Alter#processReplaceTable
- processBaseTableChange(newTableInfo.get(), "The base table has
been updated:", false);
+ processBaseTableChange(newTableInfo.get(), "The base table has
been updated:", false, false);
}
- // A RENAME leaves every column alone, and the failure it does cause
-- the MV query still
- // spells the old name -- is already reported by the refresh itself
(MTMVTask#run resolves
- // the base tables from the query before it ever looks at the
baseline). Invalidating here
- // would only leave a stale flag behind: rename the table back and the
query is analyzable
- // again, yet every strict INCREMENTAL refresh would stay rejected
until a COMPLETE one ran.
boolean renamed = !isReplace && newTableInfo.isPresent()
&& !Objects.equals(oldTableInfo.getTableName(),
newTableInfo.get().getTableName());
- processBaseTableChange(oldTableInfo, "The base table has been
updated:", !renamed);
+ // The invalidation runs first, while the dependencies are still
registered under the name the
+ // rename is leaving: moving them first would make this lookup --
which is by the old name -- find
+ // nothing, and the rename would stop invalidating anything at all.
+ processBaseTableChange(oldTableInfo, "The base table has been
updated:", !renamed, renamed);
+ if (renamed) {
+ renameBaseTable(oldTableInfo, newTableInfo.get());
+ }
+ }
+
+ /**
+ * Move a renamed table's entries in the dependency maps to its new name.
+ *
+ * <p>The maps are keyed by {@link BaseTableInfo}, which compares by name,
and an MV keeps the relation
+ * it was created against -- a rename leaves the MV query spelling the old
name, so it no longer
+ * analyzes and the relation is not recomputed. Without this the maps
would keep the old name, and a
+ * metadata-only change to the table under its new name -- a TRUNCATE,
say, which emits no row binlog --
+ * would find no dependent MV to invalidate. Renaming the table back then
restores an analyzable query
+ * whose MV still holds the rows that change removed, and nothing names
the partition that would have
+ * to be rebuilt. Moving the entries is what a rename needs instead of the
invalidation it used to
+ * carry: a rename changes no rows, so there is nothing to rebuild, only a
lookup that has to keep
+ * working.
+ */
+ private void renameBaseTable(BaseTableInfo oldTableInfo, BaseTableInfo
newTableInfo) {
Review Comment:
[P1] Make this dependency re-key survive replay and cache reconstruction.
The move only changes these derived maps; every dependent MTMV still persists
the old name in MTMV.relation, registerMTMV rebuilds the maps from that
relation, and replayRenameTable never calls this hook. After rename t -> tmp
and a restart/replay, TRUNCATE tmp therefore finds no dependent; renaming tmp
back leaves the old MV rows with no dirty epoch. Persist/replay the relation
rename (or use stable table identity), and cover rename -> restart/replay ->
TRUNCATE -> rename-back.
##########
fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelationManager.java:
##########
@@ -359,16 +362,48 @@ public void alterTable(BaseTableInfo oldTableInfo,
Optional<BaseTableInfo> newTa
// when replace, need deal two table
if (isReplace) {
// REPLACE TABLE already invalidates the IVM baseline explicitly,
see Alter#processReplaceTable
- processBaseTableChange(newTableInfo.get(), "The base table has
been updated:", false);
+ processBaseTableChange(newTableInfo.get(), "The base table has
been updated:", false, false);
}
- // A RENAME leaves every column alone, and the failure it does cause
-- the MV query still
- // spells the old name -- is already reported by the refresh itself
(MTMVTask#run resolves
- // the base tables from the query before it ever looks at the
baseline). Invalidating here
- // would only leave a stale flag behind: rename the table back and the
query is analyzable
- // again, yet every strict INCREMENTAL refresh would stay rejected
until a COMPLETE one ran.
boolean renamed = !isReplace && newTableInfo.isPresent()
&& !Objects.equals(oldTableInfo.getTableName(),
newTableInfo.get().getTableName());
- processBaseTableChange(oldTableInfo, "The base table has been
updated:", !renamed);
+ // The invalidation runs first, while the dependencies are still
registered under the name the
+ // rename is leaving: moving them first would make this lookup --
which is by the old name -- find
+ // nothing, and the rename would stop invalidating anything at all.
+ processBaseTableChange(oldTableInfo, "The base table has been
updated:", !renamed, renamed);
+ if (renamed) {
+ renameBaseTable(oldTableInfo, newTableInfo.get());
+ }
+ }
+
+ /**
+ * Move a renamed table's entries in the dependency maps to its new name.
+ *
+ * <p>The maps are keyed by {@link BaseTableInfo}, which compares by name,
and an MV keeps the relation
+ * it was created against -- a rename leaves the MV query spelling the old
name, so it no longer
+ * analyzes and the relation is not recomputed. Without this the maps
would keep the old name, and a
+ * metadata-only change to the table under its new name -- a TRUNCATE,
say, which emits no row binlog --
+ * would find no dependent MV to invalidate. Renaming the table back then
restores an analyzable query
+ * whose MV still holds the rows that change removed, and nothing names
the partition that would have
+ * to be rebuilt. Moving the entries is what a rename needs instead of the
invalidation it used to
+ * carry: a rename changes no rows, so there is nothing to rebuild, only a
lookup that has to keep
+ * working.
+ */
+ private void renameBaseTable(BaseTableInfo oldTableInfo, BaseTableInfo
newTableInfo) {
+ moveRelationKey(tableMTMVs, oldTableInfo, newTableInfo);
+ moveRelationKey(tableMTMVsOneLevelAndFromView, oldTableInfo,
newTableInfo);
+ }
+
+ private void moveRelationKey(Map<BaseTableInfo, Set<BaseTableInfo>> map,
+ BaseTableInfo oldTableInfo, BaseTableInfo newTableInfo) {
+ Set<BaseTableInfo> dependents = map.get(oldTableInfo);
+ if (CollectionUtils.isEmpty(dependents)) {
+ return;
+ }
+ // Registered under the new name before the old one is dropped: a
concurrent base-table change
Review Comment:
[P1] Close the gaps in this rename transition. Env.renameTable exposes and
journals tmp, then releases the database/table locks before Alter later invokes
this hook, so TRUNCATE tmp or DROP PARTITION can mark while the map is still
keyed only by t and commit with no rebuild epoch. Even after this method
starts, computeIfAbsent publishes an empty tmp set before addAll fills it. This
rename then skips IVM invalidation and only moves the key. Make catalog
visibility and the populated dependency entry atomic (or conservatively
invalidate across both gaps), and latch-test rename -> metadata DDL -> hook ->
rename-back.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java:
##########
@@ -715,13 +685,71 @@ 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.
+ // One read of the states decides both what has to be rebuilt and the
requirement each batch may
+ // write back. Reading them separately would leave a window between
the two in which a mark lands,
+ // the routing decision does not see it, and the batch that follows
captures the raised requirement
+ // and records it as met by a delta that cannot remove the rows that
mark made unusable.
+ Map<String, MTMVPartitionState> plannedStates =
mtmv.getPartitionStates();
+ Set<String> livePartitionNames = mtmv.getPartitionNames();
+ Set<String> dirtyPartitions = Sets.newLinkedHashSet();
+ Map<String, Long> plannedEpochs = Maps.newHashMap();
+ for (Entry<String, MTMVPartitionState> plannedState :
plannedStates.entrySet()) {
+ if (!livePartitionNames.contains(plannedState.getKey())) {
+ continue;
+ }
+ plannedEpochs.put(plannedState.getKey(),
plannedState.getValue().getLatestEpoch());
+ if (plannedState.getValue().needsRebuild()) {
+ dirtyPartitions.add(plannedState.getKey());
+ }
+ }
+ this.ivmPlannedEpochs = plannedEpochs;
+ Map<String, MTMVRefreshPartitionSnapshot> rebuiltSnapshots =
Maps.newHashMap();
+ List<String> rebuildScope = Lists.newArrayList();
+ Set<String> rebuildCompleted = Sets.newLinkedHashSet();
+ 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);
+ try {
+ executePartitionBasedRefresh(refreshContext,
RefreshMode.PARTITIONS, ctx);
+ } finally {
+ // Counted from the groups that committed, not from the ones
that were planned: a refresh
+ // that failed part-way through the rebuild must not report
partitions it never replaced.
+ recordRebuiltPartitions(request, partitionSnapshots.size());
+ }
+ rebuiltSnapshots.putAll(partitionSnapshots);
+ // Kept before the incremental attempt resets the accumulators to
its own scope: both phases
+ // belong to this refresh, so the progress it reports is the union
of the two.
+ rebuildScope.addAll(needRefreshPartitions);
+ rebuildCompleted.addAll(completedPartitions);
+ }
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);
+ // The incremental attempt reset the accumulators to its own
scope. Both phases are part of
+ // the refresh that is being reported, so the denominator is
the union of the two and the
+ // completed side keeps what each phase committed: a refresh
that rebuilt one partition and
+ // caught up another would otherwise record two of one.
+ Set<String> mergedScope = Sets.newLinkedHashSet(rebuildScope);
+ mergedScope.addAll(needRefreshPartitions);
+ this.needRefreshPartitions = Lists.newArrayList(mergedScope);
+ this.completedPartitions.addAll(rebuildCompleted);
return AttemptResultType.SUCCESS;
}
if (ivmResult.getFailureReason() !=
IvmFailureReason.MV_PARTITION_NOT_FOUND) {
Review Comment:
[P2] Preserve the dirty rebuild when IVM falls back to PARTITIONS. The
rebuild above commits and saves its snapshots only in local rebuiltSnapshots,
then executeSingleIvmAttempt resets the task accumulators. On any
fallback-allowed result this return happens before those locals are merged, so
the following PARTITIONS attempt replans against mtmv's still-missing
invalidated snapshot and INSERT OVERWRITEs every just-rebuilt partition again.
Carry the committed rebuilt set/snapshots into fallback planning or exclude
that set, and add a forced-fallback test that asserts each dirty partition is
rebuilt once.
--
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]
