[feature](ivm) Track the refresh baseline per MV partition, not by a rebuild barrier

fe/fe-core/src/main/java/org/apache/doris/alter/Alter.java

@@ -1346,7 +1346,11 @@ public void processAlterMTMV(AlterMTMV alterMTMV, boolean isReplay) {
                     break;
                 case ALTER_PARTITION_STATES:
                     // Replay only, like ALTER_IVM_INFO: a live change journals itself from inside MTMV.
-                    mtmv.alterPartitionStates(alterMTMV.getPartitionStates());
+                    // The states and the snapshot removal land in one lock acquisition: a reader that saw
+                    // the new requirement but still found the snapshot could let a transparent rewrite
+                    // serve rows the rebuild has not replaced yet.
+                    mtmv.replayAlterPartitionStates(alterMTMV.getPartitionStates(),
+                            alterMTMV.getRemovedSnapshotPartitions());
                     break;
                 default:
                     throw new RuntimeException("Unknown type value: " + alterMTMV.getOpType());

fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPartitionState.java

@@ -48,6 +48,7 @@ public class MTMVPartitionState {
     @SerializedName("le")
     private long latestEpoch;
 
+
     public MTMVPartitionState() {
     }
 
@@ -61,6 +62,35 @@ 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 refreshEpoch == 0} carries no exemption. It reads as "the partition was never refreshed,
+     * so it holds no rows", and that is not durable: a refresh commits the MV data transaction before its
+     * task result publishes the epochs, so a crash in between leaves rows in a partition whose state says
+     * never refreshed ({@code {0, 1}}). Reading that pair as clean would let a later invalidation raising
+     * {@code latestEpoch} -- {@code {0, 2}} -- go unnoticed, and a strict INCREMENTAL refresh would keep
+     * rows nothing can remove. Without the exemption the cut closes on its own: {@code 2 > 0} is dirty.
+     *
+     * <p>The cost is that a fresh MV's first refresh rebuilds every partition instead of skipping the
+     * partitions whose base partitions have no rows. That is the safe direction.
+     */
+    public boolean isDirty() {
+        return latestEpoch > refreshEpoch;
+    }
+
+
     /**
      * Deep-copies a state map, or returns null for null.
      *
@@ -95,4 +125,5 @@ public long getLatestEpoch() {
     public void setLatestEpoch(long latestEpoch) {
         this.latestEpoch = latestEpoch;
     }
+
 }

fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPlanUtil.java

@@ -1075,13 +1075,32 @@ private static void checkColumnIfChange(MTMV mtmv, List<ColumnDefinition> analyz
                             + "original length is: %s, current length is: %s",
                     originalColumns.size(), analyzedColumns.size()));
         }
-        for (int i = 0; i < originalColumns.size(); i++) {
-            if (!isTypeLike(originalColumns.get(i).getType(), analyzedColumns.get(i).getType())) {
+        // Matched by name, not by position. The order of the two lists is decided by different passes:
+        // the physical schema is laid out when the MV is created, where MTMVPlanUtil#applyIvmPhysicalKeyLayout
+        // puts the final key columns first, and the analysed list comes from running that same layout again
+        // with the stored key columns as its input. The two agree except for a chained IVM MV whose base
+        // tables carry row-id columns of their own: the create pass derives the visible key prefix from the
+        // identity key slots, the analysed one takes it from the stored keys, and the base tables' row-id
+        // columns end up in a different block. What this check is for is a base-table change that makes a
+        // column disappear or change type, and where a column sits is not part of that.
+        Map<String, Column> originalByName = Maps.newHashMap();
+        for (Column column : originalColumns) {
+            originalByName.put(column.getName().toLowerCase(), column);
+        }
+        for (Column analyzedColumn : analyzedColumns) {
+            Column originalColumn = originalByName.get(analyzedColumn.getName().toLowerCase());
+            if (originalColumn == null) {
+                throw new JobException(String.format(
+                        "column not found, please check whether columns of base table have changed, "
+                                + "column name is: %s",
+                        analyzedColumn.getName()));
+            }
+            if (!isTypeLike(originalColumn.getType(), analyzedColumn.getType())) {
                 throw new JobException(String.format(
                         "column type not same, please check whether columns of base table have changed, "
                                 + "column name is: %s, original type is: %s, current type is: %s",
-                        originalColumns.get(i).getName(), originalColumns.get(i).getType().toSql(),
-                        analyzedColumns.get(i).getType().toSql()));
+                        analyzedColumn.getName(), originalColumn.getType().toSql(),
+                        analyzedColumn.getType().toSql()));
             }
         }
     }

fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRefreshSnapshot.java

@@ -22,6 +22,7 @@
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
 import com.google.gson.annotations.SerializedName;
+import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.collections4.MapUtils;
 
 import java.util.Iterator;
@@ -86,6 +87,17 @@ public void updateSnapshots(Map<String, MTMVRefreshPartitionSnapshot> addPartiti
         }
     }
 
+    /**
+     * Drops these MV partitions' snapshots. An invalidated partition must not be able to back a
+     * transparent rewrite until it has been rebuilt: what it holds is exactly what the rebuild replaces.
+     */
+    public void removeSnapshots(Set<String> mvPartitionNames) {
+        if (CollectionUtils.isEmpty(mvPartitionNames)) {
+            return;
+        }
+        partitionSnapshots.keySet().removeAll(mvPartitionNames);
+    }
+
     public Map<String, MTMVRefreshPartitionSnapshot> getPartitionSnapshots() {
         return partitionSnapshots;
     }

fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelationManager.java

@@ -117,10 +117,12 @@ private void markIvmBaselineRebuild(BaseTableInfo baseTableInfo, boolean allPart
             }
             boolean invalidated;
             if (allPartitionsChanged) {
-                mtmv.invalidateIvmBaseline();
+                // Awaited here, where no MV lock is held: the DDL does not return until the invalidation is
+                // durable, which is what it was before the record was handed back to the caller.
+                mtmv.invalidateWholeMv(reason).await();
                 invalidated = true;
             } else {
-                invalidated = mtmv.invalidateIvmBaseline(baseTableInfo, changedPartitions);
+                invalidated = mtmv.invalidateIvmBaseline(baseTableInfo, changedPartitions, reason);
             }
             // A partition change that no MV partition reads leaves nothing to rebuild, and saying that it
             // invalidated the baseline would claim a persisted barrier that does not exist.
@@ -344,13 +346,16 @@ public void refreshComplete(MTMV mtmv, MTMVRelation relation, MTMVTask task) {
      */
     @Override
     public void dropTable(Table table) {
-        // A dropped base table is already caught by the IVM stream guard (the stream records the
-        // base table id, so it stops being usable once the table is gone), no need to re-analyze.
+        // The message below names the table and what became of it, which is what an MV that reads it has to
+        // know; the query check would replace that with the weaker "the query is no longer analyzable",
+        // because a dropped table is the one change whose query is gone beyond doubt. What the two record
+        // is the same state either way. Unlike a rename it stays an invalidation: the table is gone for
+        // good, so the state is not something a later alter can make obsolete.
         processBaseTableChange(new BaseTableInfo(table), "The base table has been deleted:", false);
     }
 
     /**
-     * update mtmv status to `SCHEMA_CHANGE`
+     * update mtmv status to `SCHEMA_CHANGE`.
      *
      * @param isReplace
      */
@@ -361,35 +366,43 @@ public void alterTable(BaseTableInfo oldTableInfo, Optional<BaseTableInfo> newTa
             // REPLACE TABLE already invalidates the IVM baseline explicitly, see Alter#processReplaceTable
             processBaseTableChange(newTableInfo.get(), "The base table has been updated:", 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);
+        // A rename is the one change whose query check is skipped: the MV query keeps spelling the old
+        // name, so it is unanalyzable by construction, and the reason it would be invalidated with --
+        // "the query is no longer analyzable" -- says less than the message this call records anyway.
+        boolean checkQueryUsable = !renamed;
+        processBaseTableChange(oldTableInfo, "The base table has been updated:", checkQueryUsable);
     }
 
+
     /**
-     * An IVM baseline is only valid while the MV query can still be analyzed against the current
-     * base table schema. Re-analyzing the MV query here (right after the alter was applied) is what
-     * detects a changed column identity: dropping or renaming a column the MV uses makes the query
-     * unanalyzable, and a column re-added with the same name is a different column, so pre-existing
-     * rows read its default value instead.
+     * An MV's query is only as good as the base table schema it was analyzed against. Re-analyzing the
+     * MV query here (right after the alter was applied) is what detects a changed column identity:
+     * dropping or renaming a column the MV uses makes the query unanalyzable, and a column re-added with
+     * the same name is a different column, so pre-existing rows read its default value instead.
      *
      * <p>Such a change is metadata-only for light schema changes and emits no binlog, so an
      * incremental refresh would consume an empty delta and report SUCCESS while silently keeping the
-     * rows computed under the old column epoch. Invalidating the baseline makes a strict INCREMENTAL
-     * refresh fail and tell the user to run a COMPLETE refresh instead.
+     * rows computed under the old column epoch. Invalidating the MV is what keeps that from being
+     * reported as current.
      *
-     * <p>Only IVM is covered: a plain MTMV keeps its previous behaviour (status only).
+     * <p>Every MV is checked, not only an IVM one: whether the query still analyzes is a property of
+     * the MV and of the base table it reads, not of how the MV refreshes, and the invalidation is the
+     * same one a change to that table records. What an IVM MV has on top of it is a per-partition
+     * requirement, and that is decided elsewhere, from a query that analyzed.
+     *
+     * @return whether the MV was invalidated. That is the whole record for this change: the invalidation
+     *         carries the reason, and the caller has nothing left to write -- a second record would land
+     *         on the same state, and MTMVStatus#updateStateAndDetail would overwrite the detail with the
+     *         blunter "the base table has been updated", which is what knowing the query is unusable is
+     *         for. It would also bump the version and drop the snapshot twice for one change.
      */
-    private void invalidateIvmBaselineIfQueryUnusable(BaseTableInfo baseTableInfo, Table mtmvTable) {
-        if (!(mtmvTable instanceof MTMV) || !((MTMV) mtmvTable).isIvm()) {
-            return;
+    private boolean invalidateMvIfQueryUnusable(BaseTableInfo baseTableInfo, Table mvTable) {
+        if (!(mvTable instanceof MTMV)) {
+            return false;
         }
-        MTMV mtmv = (MTMV) mtmvTable;
+        MTMV mtmv = (MTMV) mvTable;
         // Analyse in a context owned by this check, never the session that issued the alter: the check
         // must not disturb the running statement, and it has to work on threads that have no session.
         // Setting a thread local is how a context is made current, so restore the previous one.
@@ -398,16 +411,18 @@ private void invalidateIvmBaselineIfQueryUnusable(BaseTableInfo baseTableInfo, T
             MTMVPlanUtil.ensureMTMVQueryUsable(mtmv,
                     MTMVPlanUtil.createMTMVContext(mtmv, MTMVPlanUtil.DISABLE_RULES_WHEN_RUN_MTMV_TASK));
         } catch (Exception e) {
-            LOG.info("Invalidate IVM baseline, the MV query is no longer usable. baseTable={}, mtmv={}, "
-                    + "reason={}", baseTableInfo, mtmv.getName(), e.getMessage());
-            mtmv.invalidateIvmBaseline();
+            LOG.info("Invalidate MV, the MV query is no longer usable. baseTable={}, mtmv={}, reason={}",
+                    baseTableInfo, mtmv.getName(), e.getMessage());
+            mtmv.invalidateWholeMv("The MV query is no longer analyzable: " + baseTableInfo).await();
+            return true;
         } finally {
             if (previousCtx != null) {
                 previousCtx.setThreadLocalInfo();
             } else {
                 ConnectContext.remove();
             }
         }
+        return false;
     }
 
     @Override
@@ -469,8 +484,14 @@ private void processBaseViewChange(BaseTableInfo baseViewInfo, String msgPrefix)
         }
     }
 
+    /**
+     * Puts every MV that reads this base table into {@code SCHEMA_CHANGE}.
+     *
+     * @param checkQueryUsable whether to re-analyze each MV's query first; see
+     *                         {@link #invalidateMvIfQueryUnusable}
+     */
     private void processBaseTableChange(BaseTableInfo baseTableInfo, String msgPrefix,
-            boolean checkIvmQueryUsable) {
+            boolean checkQueryUsable) {
         Set<BaseTableInfo> mtmvsByBaseTable = getMtmvsByBaseTableOneLevelAndFromView(baseTableInfo);
         if (CollectionUtils.isEmpty(mtmvsByBaseTable)) {
             return;
@@ -483,8 +504,11 @@ private void processBaseTableChange(BaseTableInfo baseTableInfo, String msgPrefi
                 LOG.warn(e);
                 continue;
             }
-            if (checkIvmQueryUsable) {
-                invalidateIvmBaselineIfQueryUnusable(baseTableInfo, mtmv);
+            if (checkQueryUsable && invalidateMvIfQueryUnusable(baseTableInfo, mtmv)) {
+                // Invalidated with the reason, which is the more specific of the two messages and the one
+                // this change is worth recording: the state is the same one the generic record below
+                // would set, so writing it too would only bury the reason.
+                continue;
             }
             TableNameInfo tableNameInfo = new TableNameInfo(mtmv.getQualifiedDbName(),
                     mtmv.getName());

fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmInfo.java

@@ -17,12 +17,8 @@
 
 package org.apache.doris.mtmv.ivm;
 
-import com.google.common.base.Preconditions;
 import com.google.gson.annotations.SerializedName;
 
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.Set;
 
 /**
  * Thin persistent IVM metadata stored on MTMV.
@@ -35,16 +31,6 @@ public class IvmInfo {
     @SerializedName("en")
     private boolean enableIvm = false;
 
-    @SerializedName("brr")
-    // Keep an explicit COMPLETE requirement because persisted partition names are a point-in-time snapshot.
-    // For example, if the MV has {p1, p2} when invalidated and partition sync adds p3 before refresh,
-    // COMPLETE must rebuild p3 too.
-    private boolean completeBaselineRebuildRequired;
-
-    @SerializedName("brp")
-    // MV partitions that must be rebuilt before their IVM offsets can be used again.
-    private Set<String> pendingBaselineRebuildPartitions = new HashSet<>();
-
     /** Persisted ivm_use_full_keys flag: true means the MV unique keys include identity key columns. */
     @SerializedName("ukf")
     private boolean useFullKeys = false;
@@ -62,8 +48,6 @@ public IvmInfo() {
 
     public IvmInfo(IvmInfo other) {
         this.enableIvm = other.enableIvm;
-        this.completeBaselineRebuildRequired = other.completeBaselineRebuildRequired;
-        this.pendingBaselineRebuildPartitions = new HashSet<>(other.pendingBaselineRebuildPartitions);
         this.useFullKeys = other.useFullKeys;
         this.planSignature = other.planSignature;
         this.sequencePrefix = other.sequencePrefix;
@@ -77,35 +61,6 @@ public void setEnableIvm(boolean enableIvm) {
         this.enableIvm = enableIvm;
     }
 
-    public boolean isBaselineRebuildRequired() {
-        return completeBaselineRebuildRequired || !pendingBaselineRebuildPartitions.isEmpty();
-    }
-
-    public boolean requiresCompleteBaselineRebuild() {
-        return completeBaselineRebuildRequired;
-    }
-
-    public Set<String> getPendingBaselineRebuildPartitions() {
-        return Collections.unmodifiableSet(new HashSet<>(pendingBaselineRebuildPartitions));
-    }
-
-    public void requireCompleteBaselineRebuild() {
-        completeBaselineRebuildRequired = true;
-        pendingBaselineRebuildPartitions.clear();
-    }
-
-    public void addPendingBaselineRebuildPartitions(Set<String> partitions) {
-        Preconditions.checkArgument(!partitions.isEmpty(), "baseline rebuild partitions can not be empty");
-        if (!completeBaselineRebuildRequired) {
-            pendingBaselineRebuildPartitions.addAll(partitions);
-        }
-    }
-
-    public void clearBaselineRebuild() {
-        completeBaselineRebuildRequired = false;
-        pendingBaselineRebuildPartitions.clear();
-    }
-
     public boolean isUseFullKeys() {
         return useFullKeys;
     }
@@ -134,8 +89,6 @@ public void advanceSequencePrefix() {
     public String toString() {
         return "IvmInfo{"
                 + "enableIvm=" + enableIvm
-                + ", completeBaselineRebuildRequired=" + completeBaselineRebuildRequired
-                + ", pendingBaselineRebuildPartitions=" + pendingBaselineRebuildPartitions
                 + ", useFullKeys=" + useFullKeys
                 + ", planSignature='" + planSignature + '\''
                 + '}';