[feature](ivm) Add the per-partition refresh state and its journal channel

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

@@ -1344,6 +1344,10 @@ public void processAlterMTMV(AlterMTMV alterMTMV, boolean isReplay) {
                     // Live IVM changes are journaled inside MTMV; this branch applies the journal snapshot.
                     mtmv.alterIvmInfo(alterMTMV.getIvmInfo());
                     break;
+                case ALTER_PARTITION_STATES:
+                    // Replay only, like ALTER_IVM_INFO: a live change journals itself from inside MTMV.
+                    mtmv.alterPartitionStates(alterMTMV.getPartitionStates());
+                    break;
                 default:
                     throw new RuntimeException("Unknown type value: " + alterMTMV.getOpType());
             }

fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java

@@ -38,6 +38,7 @@
 import org.apache.doris.mtmv.MTMVPartitionExpander;
 import org.apache.doris.mtmv.MTMVPartitionInfo;
 import org.apache.doris.mtmv.MTMVPartitionInfo.MTMVPartitionType;
+import org.apache.doris.mtmv.MTMVPartitionState;
 import org.apache.doris.mtmv.MTMVPartitionUtil;
 import org.apache.doris.mtmv.MTMVPlanUtil;
 import org.apache.doris.mtmv.MTMVPropertyUtil;
@@ -103,6 +104,19 @@ public class MTMV extends OlapTable {
     private MTMVRefreshSnapshot refreshSnapshot;
     @SerializedName("ii")
     private IvmInfo ivmInfo;
+    /**
+     * The refresh epoch of every MV partition, keyed by MV partition name.
+     *
+     * <p>Deliberately on MTMV rather than inside {@link IvmInfo}: the field is shared, the behaviour is
+     * not. Both kinds of MV carry it, but only an IVM MV ever populates it -- alignment, invalidation,
+     * the ADD_TASK payload and ALTER_PARTITION_STATES are all no-ops for a non-IVM MV, so for one an
+     * empty map is the complete answer.
+     *
+     * <p>Null means the same thing and has the same two causes: an image written before the field
+     * existed, and a non-IVM MV. {@link #gsonPostProcess()} turns it into an empty map on load.
+     */
+    @SerializedName("pst")
+    private Map<String, MTMVPartitionState> partitionStates;
     // Should update after every fresh, not persist
     // Cache with SessionVarGuardExpr: used when query session variables differ from MV creation variables
     private MTMVCache cacheWithGuard;
@@ -290,6 +304,11 @@ public boolean addTaskResult(AlterMTMV alterMTMV, boolean isReplay) {
                 // Replay the final IVM state; ADD_TASK does not change schemaChangeVersion.
                 ivmInfo = new IvmInfo(alterMTMV.getIvmInfo());
             }
+            if (isReplay && alterMTMV.getPartitionStates() != null) {
+                // A journal written before the field existed carries no state at all: leave the
+                // partition states alone rather than clearing them.
+                partitionStates = MTMVPartitionState.copyOf(alterMTMV.getPartitionStates());
+            }
             if (task.getStatus() == TaskStatus.SUCCESS) {
                 this.status.setState(MTMVState.NORMAL);
                 this.status.setSchemaChangeDetail(null);
@@ -323,6 +342,10 @@ public boolean addTaskResult(AlterMTMV alterMTMV, boolean isReplay) {
             }
             if (ivmInfo.isEnableIvm()) {
                 alterMTMV.setIvmInfo(ivmInfo);
+                // Same condition as ivmInfo, so the journal of a non-IVM MV stays byte-for-byte what it
+                // was. The map is null until the states are first aligned, and a payload without the
+                // member means the same as one carrying an empty map.
+                alterMTMV.setPartitionStates(partitionStates);
             }
             editLogItem = submitAlterLog(alterMTMV);
         } finally {
@@ -598,6 +621,33 @@ public void alterIvmInfo(IvmInfo ivmInfo) {
         }
     }
 
+    /**
+     * Read under the MV lock, like {@link #getIvmInfo()}: the map may be null before
+     * {@link #gsonPostProcess()} has run, and a reader must never see a half-applied replay payload.
+     */
+    public Map<String, MTMVPartitionState> getPartitionStates() {
+        writeMvLock();
+        try {
+            if (partitionStates == null) {
+                partitionStates = Maps.newLinkedHashMap();
+            }
+            return partitionStates;
+        } finally {
+            writeMvUnlock();
+        }
+    }
+
+    // ALTER_PARTITION_STATES replay applies a detached snapshot here, mirroring alterIvmInfo(). Live
+    // invalidation changes submit their journal from the mutating method instead.
+    public void alterPartitionStates(Map<String, MTMVPartitionState> partitionStates) {
+        writeMvLock();
+        try {
+            this.partitionStates = MTMVPartitionState.copyOf(partitionStates);
+        } finally {
+            writeMvUnlock();
+        }
+    }
+
     public void invalidateIvmBaseline() {
         EditLogItem editLogItem;
         writeMvLock();
@@ -997,6 +1047,11 @@ public void gsonPostProcess() throws IOException {
         if (ivmInfo == null) {
             ivmInfo = new IvmInfo();
         }
+        if (partitionStates == null) {
+            // An image written before the field existed deserializes it as null, and so does a non-IVM MV.
+            // Both mean "no state", so an empty map is the whole answer.
+            partitionStates = Maps.newLinkedHashMap();
+        }
         if (refreshInfo != null && refreshInfo.getRefreshMethod() == null) {
             LOG.warn("MTMV {} has unknown refresh method, marking as schema change", name);
             status.setState(MTMVState.SCHEMA_CHANGE);

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

@@ -22,5 +22,6 @@ public enum MTMVAlterOpType {
     ALTER_STATUS,
     ALTER_PROPERTY,
     ADD_TASK,
-    ALTER_IVM_INFO;
+    ALTER_IVM_INFO,
+    ALTER_PARTITION_STATES;
 }

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

@@ -0,0 +1,98 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.mtmv;
+
+import com.google.gson.annotations.SerializedName;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+
+/**
+ * The per-partition refresh state of one MV partition.
+ *
+ * <p>{@code refreshEpoch} says which generation of data the partition currently holds, {@code
+ * latestEpoch} says which generation it must hold. A partition whose {@code latestEpoch} is ahead of
+ * its {@code refreshEpoch} is dirty: it holds rows read before a metadata-only change of a base table
+ * (a dropped / truncated / replaced / recovered partition emits no row binlog), so those rows can no
+ * longer be removed incrementally and the partition has to be rebuilt.
+ *
+ * <p>Keyed by MV partition name in {@code MTMV.partitionStates}. The name is deliberately the only
+ * identity: an MV partition is rewritten by {@code INSERT OVERWRITE} on every refresh and gets a new
+ * partition id each time, so an id would stop matching as soon as the partition is refreshed.
+ *
+ * <p>The two values are plain {@code long}s rather than atomics because this is a persisted DTO: it is
+ * serialized into the alter journal, so it has to stay a plain bean.
+ */
+public class MTMVPartitionState {
+    /** The generation of the data this MV partition currently holds; 0 means it was never refreshed. */
+    @SerializedName("re")
+    private long refreshEpoch;
+
+    /** The generation the data must reach; starts at 1 and grows on every invalidation. */
+    @SerializedName("le")
+    private long latestEpoch;
+
+    public MTMVPartitionState() {
+    }
+
+    public MTMVPartitionState(long refreshEpoch, long latestEpoch) {
+        this.refreshEpoch = refreshEpoch;
+        this.latestEpoch = latestEpoch;
+    }
+
+    public MTMVPartitionState(MTMVPartitionState other) {
+        this.refreshEpoch = other.refreshEpoch;
+        this.latestEpoch = other.latestEpoch;
+    }
+
+    /**
+     * Deep-copies a state map, or returns null for null.
+     *
+     * <p>The journal needs this on both sides. A payload is serialized by the journal thread, which
+     * runs after the submitting thread released the MV lock, so a payload that shared state with the
+     * live map could be written out half-mutated. The replay path goes through the same helper so that
+     * both sides of the journal follow one rule instead of two.
+     */
+    public static Map<String, MTMVPartitionState> copyOf(Map<String, MTMVPartitionState> states) {
+        if (states == null) {
+            return null;
+        }
+        Map<String, MTMVPartitionState> copy = new LinkedHashMap<>();
+        for (Entry<String, MTMVPartitionState> entry : states.entrySet()) {
+            copy.put(entry.getKey(), new MTMVPartitionState(entry.getValue()));
+        }
+        return copy;
+    }
+
+    public long getRefreshEpoch() {
+        return refreshEpoch;
+    }
+
+    public void setRefreshEpoch(long refreshEpoch) {
+        this.refreshEpoch = refreshEpoch;
+    }
+
+    public long getLatestEpoch() {
+        return latestEpoch;
+    }
+
+    public void setLatestEpoch(long latestEpoch) {
+        this.latestEpoch = latestEpoch;
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/persist/AlterMTMV.java

@@ -22,6 +22,7 @@
 import org.apache.doris.common.io.Writable;
 import org.apache.doris.job.extensions.mtmv.MTMVTask;
 import org.apache.doris.mtmv.MTMVAlterOpType;
+import org.apache.doris.mtmv.MTMVPartitionState;
 import org.apache.doris.mtmv.MTMVRefreshInfo;
 import org.apache.doris.mtmv.MTMVRefreshPartitionSnapshot;
 import org.apache.doris.mtmv.MTMVRelation;
@@ -58,6 +59,8 @@ public class AlterMTMV implements Writable {
     private Map<String, MTMVRefreshPartitionSnapshot> partitionSnapshots;
     @SerializedName("ii")
     private IvmInfo ivmInfo;
+    @SerializedName("pst")
+    private Map<String, MTMVPartitionState> partitionStates;
 
     public AlterMTMV(TableNameInfo mvName, MTMVRefreshInfo refreshInfo, MTMVAlterOpType opType) {
         this.mvName = Objects.requireNonNull(mvName, "require mvName object");
@@ -148,6 +151,14 @@ public void setIvmInfo(IvmInfo ivmInfo) {
         this.ivmInfo = ivmInfo == null ? null : new IvmInfo(ivmInfo);
     }
 
+    public Map<String, MTMVPartitionState> getPartitionStates() {
+        return partitionStates;
+    }
+
+    public void setPartitionStates(Map<String, MTMVPartitionState> partitionStates) {
+        this.partitionStates = MTMVPartitionState.copyOf(partitionStates);
+    }
+
     @Override
     public String toString() {
         return "AlterMTMV{"

fe/fe-core/src/test/java/org/apache/doris/mtmv/AlterMTMVTest.java

@@ -37,6 +37,8 @@
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
+import java.util.LinkedHashMap;
+import java.util.Map;
 import java.util.Set;
 
 
@@ -403,6 +405,44 @@ public void testAlterIvmInfoPersistence() throws Exception {
         Assertions.assertEquals(schemaChangeVersion, mtmv.getSchemaChangeVersion());
     }
 
+    @Test
+    public void testReplayAlterPartitionStates() throws Exception {
+        Config.enable_table_stream = true;
+        createDatabaseAndUse("alter_partition_states_test");
+        createTable("CREATE TABLE alter_partition_states_test.states_base (k1 int, v1 int)\n"
+                + "DUPLICATE KEY(k1)\n"
+                + "DISTRIBUTED BY HASH(k1) BUCKETS 1\n"
+                + "PROPERTIES ('replication_num' = '1', 'binlog.enable' = 'true', 'binlog.format' = 'ROW')");
+        createMvByNereids("CREATE MATERIALIZED VIEW states_mv\n"
+                + " BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL\n"
+                + " DISTRIBUTED BY RANDOM BUCKETS 2\n"
+                + " PROPERTIES ('replication_num' = '1')\n"
+                + " AS SELECT k1, v1 FROM states_base");
+
+        MTMV mtmv = (MTMV) Env.getCurrentInternalCatalog()
+                .getDb("alter_partition_states_test").get()
+                .getTableOrMetaException("states_mv");
+        String partitionName = mtmv.getPartitionNames().iterator().next();
+
+        MTMVPartitionState state = new MTMVPartitionState(0, 1);
+        Map<String, MTMVPartitionState> states = new LinkedHashMap<>();
+        states.put(partitionName, state);
+        TableNameInfo tableName = new TableNameInfo(mtmv.getQualifiedDbName(), mtmv.getName());
+        AlterMTMV replayAlter = new AlterMTMV(tableName, MTMVAlterOpType.ALTER_PARTITION_STATES);
+        replayAlter.setPartitionStates(states);
+        // The live map keeps moving after the payload was taken; the payload must not follow it.
+        state.setLatestEpoch(7);
+        // The MV starts without any state, so only the replayed payload can put it there.
+        mtmv.alterPartitionStates(Map.of());
+
+        Env.getCurrentEnv().getAlterInstance().processAlterMTMV(replayAlter, true);
+
+        Map<String, MTMVPartitionState> replayed = mtmv.getPartitionStates();
+        Assertions.assertEquals(Set.of(partitionName), replayed.keySet());
+        Assertions.assertEquals(0, replayed.get(partitionName).getRefreshEpoch());
+        Assertions.assertEquals(1, replayed.get(partitionName).getLatestEpoch());
+    }
+
     @Test
     public void testCreateIncrementalMtmvAutoCreatesStream() throws Exception {
         createDatabaseAndUse("stream_test");