[refactor](ivm) Rename IvmInfo.refreshVersion to sequencePrefix

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

@@ -255,8 +255,8 @@ public boolean isIvm() {
         return getIvmInfo().isEnableIvm();
     }
 
-    public long getNextRefreshVersion() {
-        return Config.isCloudMode() ? getNextVersion() : getIvmInfo().getRefreshVersion() + 1;
+    public long getNextSequencePrefix() {
+        return Config.isCloudMode() ? getNextVersion() : getIvmInfo().getSequencePrefix() + 1;
     }
 
     public boolean addTaskResult(AlterMTMV alterMTMV, boolean isReplay) {

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

@@ -155,7 +155,7 @@ __DORIS_IVM_ROW_ID_COL__ | k1 | cnt | sum_v1 | __DORIS_IVM_DML_FACTOR_COL__ | __
 
 ### Semantics
 
-- **Read-only.** No insert transaction is built. Stream offsets, refresh version, and MV metadata
+- **Read-only.** No insert transaction is built. Stream offsets, sequence prefix, and MV metadata
   are never modified.
 - **Idempotent.** Repeating the same dry run returns identical rows as long as the base table data
   has not changed.

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

@@ -57,11 +57,11 @@ class IvmDeltaRewriteState {
     private int nextDeltaScanIndex;
 
     IvmDeltaRewriteState(Map<OlapTable, OlapTableStream> streams,
-            boolean includeExhaustedStreams, long refreshVersion, DataType sequenceType,
+            boolean includeExhaustedStreams, long sequencePrefix, DataType sequenceType,
             Map<OlapTable, List<Long>> windowPartitionIdsByTable) {
         this.streams = new HashMap<>(streams);
         this.includeExhaustedStreams = includeExhaustedStreams;
-        this.sequenceCalculator = IvmSequenceCalculator.create(refreshVersion, sequenceType);
+        this.sequenceCalculator = IvmSequenceCalculator.create(sequencePrefix, sequenceType);
         this.windowPartitionIdsByTable = windowPartitionIdsByTable;
     }
 

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

@@ -67,8 +67,8 @@ public Plan generateIncrRefreshPlan(Plan sinkChild, IvmRewriteResult rewriteResu
                 rewriteContext.isIncludeExhaustedStreams());
         Pair<Plan, List<LogicalProject<?>>> prefixChain = helper.detachAdaptProjectChain(sinkChild);
         Plan rootPlan = prefixChain.first;
-        long refreshVersion = refreshContext.getMtmv().getNextRefreshVersion();
-        IvmDeltaRewriteState rewriteState = createDeltaRewriteState(rootPlan, refreshContext, refreshVersion,
+        long sequencePrefix = refreshContext.getMtmv().getNextSequencePrefix();
+        IvmDeltaRewriteState rewriteState = createDeltaRewriteState(rootPlan, refreshContext, sequencePrefix,
                 rewriteContext.getIncrementalScopePartitionIds());
         Optional<IvmDeltaRewriteResult> deltaResult = rewriteDelta(rootPlan, refreshContext, rewriteState);
         if (!deltaResult.isPresent()) {
@@ -156,7 +156,7 @@ private static Pair<Plan, Map<Slot, Slot>> rewriteSnapshot(Plan plan, IvmDeltaRe
         return IvmDeltaRewriteHelper.INSTANCE.freshPlan(rewritten);
     }
 
-    private IvmDeltaRewriteState createDeltaRewriteState(Plan plan, IvmIncrRefreshContext ctx, long refreshVersion,
+    private IvmDeltaRewriteState createDeltaRewriteState(Plan plan, IvmIncrRefreshContext ctx, long sequencePrefix,
             Map<BaseTableInfo, Set<Long>> scopePartitionIds) {
         Map<OlapTable, OlapTableStream> streams = new HashMap<>();
         // Window limits apply to every base table in the plan, including excluded
@@ -186,7 +186,7 @@ private IvmDeltaRewriteState createDeltaRewriteState(Plan plan, IvmIncrRefreshCo
             }
         }
         applyScopePartitionIds(windowPartitionIdsByTable, planTables, scopePartitionIds);
-        return new IvmDeltaRewriteState(streams, ctx.isIncludeExhaustedStreams(), refreshVersion,
+        return new IvmDeltaRewriteState(streams, ctx.isIncludeExhaustedStreams(), sequencePrefix,
                 DataType.fromCatalogType(ctx.getMtmv().getColumn(Column.SEQUENCE_COL).getType()),
                 windowPartitionIdsByTable);
     }

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

@@ -53,8 +53,9 @@ public class IvmInfo {
     @SerializedName("ps")
     private String planSignature;
 
-    @SerializedName("rv")
-    private long refreshVersion;
+    /** The prefix of the sequence values this MV's rows are stamped with; see IvmSequenceCalculator. */
+    @SerializedName("sp")
+    private long sequencePrefix;
 
     public IvmInfo() {
     }
@@ -65,7 +66,7 @@ public IvmInfo(IvmInfo other) {
         this.pendingBaselineRebuildPartitions = new HashSet<>(other.pendingBaselineRebuildPartitions);
         this.useFullKeys = other.useFullKeys;
         this.planSignature = other.planSignature;
-        this.refreshVersion = other.refreshVersion;
+        this.sequencePrefix = other.sequencePrefix;
     }
 
     public boolean isEnableIvm() {
@@ -121,12 +122,12 @@ public void setPlanSignature(String planSignature) {
         this.planSignature = planSignature;
     }
 
-    public long getRefreshVersion() {
-        return refreshVersion;
+    public long getSequencePrefix() {
+        return sequencePrefix;
     }
 
-    public void advanceRefreshVersion() {
-        refreshVersion++;
+    public void advanceSequencePrefix() {
+        sequencePrefix++;
     }
 
     @Override

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

@@ -33,32 +33,32 @@
 /**
  * Encodes IVM delta ordering into the MTMV sequence column.
  *
- * <p>BIGINT encodes {@code (refresh version, delta index, op)}. LARGEINT additionally encodes
- * a 64-bit binlog sequence: {@code (refresh version, delta index, binlog sequence, op)}.
+ * <p>BIGINT encodes {@code (sequence prefix, delta index, op)}. LARGEINT additionally encodes
+ * a 64-bit binlog sequence: {@code (sequence prefix, delta index, binlog sequence, op)}.
  */
 abstract class IvmSequenceCalculator {
     static final int DELTA_INDEX_BITS = 10;
     private static final int BIGINT_LOW_BITS = DELTA_INDEX_BITS + 1;
     private static final int LARGEINT_BINLOG_BITS = 64;
     private static final int LARGEINT_DELTA_INDEX_SHIFT = LARGEINT_BINLOG_BITS + 1;
-    private static final int LARGEINT_REFRESH_VERSION_SHIFT =
+    private static final int LARGEINT_SEQUENCE_PREFIX_SHIFT =
             LARGEINT_DELTA_INDEX_SHIFT + DELTA_INDEX_BITS;
     private static final int MAX_DELTA_INDEX = 1 << DELTA_INDEX_BITS;
     private static final BigInteger MAX_BINLOG_SEQUENCE =
             BigInteger.ONE.shiftLeft(LARGEINT_BINLOG_BITS).subtract(BigInteger.ONE);
 
-    final long refreshVersion;
+    final long sequencePrefix;
 
-    private IvmSequenceCalculator(long refreshVersion) {
-        this.refreshVersion = refreshVersion;
+    private IvmSequenceCalculator(long sequencePrefix) {
+        this.sequencePrefix = sequencePrefix;
     }
 
-    static IvmSequenceCalculator create(long refreshVersion, DataType sequenceType) {
+    static IvmSequenceCalculator create(long sequencePrefix, DataType sequenceType) {
         if (sequenceType.equals(BigIntType.INSTANCE)) {
-            return new BigIntSequenceCalculator(refreshVersion);
+            return new BigIntSequenceCalculator(sequencePrefix);
         }
         if (sequenceType.equals(LargeIntType.INSTANCE)) {
-            return new LargeIntSequenceCalculator(refreshVersion);
+            return new LargeIntSequenceCalculator(sequencePrefix);
         }
         throw new IvmException(IvmFailureReason.PLAN_REWRITE_FAILED,
                 "unsupported IVM sequence type: " + sequenceType.simpleString());
@@ -87,11 +87,11 @@ final void checkBinlogSequence(BigInteger binlogSequence) {
     }
 
     private static class BigIntSequenceCalculator extends IvmSequenceCalculator {
-        private BigIntSequenceCalculator(long refreshVersion) {
-            super(refreshVersion);
-            if (refreshVersion < 0 || refreshVersion > (Long.MAX_VALUE >>> BIGINT_LOW_BITS)) {
+        private BigIntSequenceCalculator(long sequencePrefix) {
+            super(sequencePrefix);
+            if (sequencePrefix < 0 || sequencePrefix > (Long.MAX_VALUE >>> BIGINT_LOW_BITS)) {
                 throw new IvmException(IvmFailureReason.PLAN_REWRITE_FAILED,
-                        "IVM refresh version exceeds the BIGINT sequence encoding range: " + refreshVersion);
+                        "IVM sequence prefix exceeds the BIGINT sequence encoding range: " + sequencePrefix);
             }
         }
 
@@ -102,26 +102,26 @@ Literal encode(int deltaIndex, BigInteger binlogSequence, boolean positive) {
                 throw new IvmException(IvmFailureReason.PLAN_REWRITE_FAILED,
                         "BIGINT IVM sequence does not support binlog sequence");
             }
-            long sequence = (refreshVersion << BIGINT_LOW_BITS)
+            long sequence = (sequencePrefix << BIGINT_LOW_BITS)
                     | ((long) deltaIndex << 1) | (positive ? 1 : 0);
             return new BigIntLiteral(sequence);
         }
     }
 
     private static class LargeIntSequenceCalculator extends IvmSequenceCalculator {
-        private LargeIntSequenceCalculator(long refreshVersion) {
-            super(refreshVersion);
-            if (refreshVersion < 0 || refreshVersion > (Long.MAX_VALUE >>> BIGINT_LOW_BITS)) {
+        private LargeIntSequenceCalculator(long sequencePrefix) {
+            super(sequencePrefix);
+            if (sequencePrefix < 0 || sequencePrefix > (Long.MAX_VALUE >>> BIGINT_LOW_BITS)) {
                 throw new IvmException(IvmFailureReason.PLAN_REWRITE_FAILED,
-                        "IVM refresh version exceeds the LARGEINT sequence encoding range: " + refreshVersion);
+                        "IVM sequence prefix exceeds the LARGEINT sequence encoding range: " + sequencePrefix);
             }
         }
 
         @Override
         Literal encode(int deltaIndex, BigInteger binlogSequence, boolean positive) {
             checkDeltaIndex(deltaIndex);
             checkBinlogSequence(binlogSequence);
-            BigInteger sequence = BigInteger.valueOf(refreshVersion).shiftLeft(LARGEINT_REFRESH_VERSION_SHIFT)
+            BigInteger sequence = BigInteger.valueOf(sequencePrefix).shiftLeft(LARGEINT_SEQUENCE_PREFIX_SHIFT)
                     .or(BigInteger.valueOf(deltaIndex).shiftLeft(LARGEINT_DELTA_INDEX_SHIFT))
                     .or(binlogSequence.shiftLeft(1));
             if (positive) {

fe/fe-core/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java

@@ -2418,15 +2418,15 @@ private void updateCatalogAfterCommitted(TransactionState transactionState, Data
         // update table stream offset if necessary
         if (!CollectionUtils.isEmpty(transactionState.getStreamUpdateInfos())) {
             updateStreamOffset(transactionState, transactionState.getCommitTime());
-            updateIvmRefreshVersion(transactionState, db);
+            updateIvmSequencePrefix(transactionState, db);
         }
     }
 
-    private void updateIvmRefreshVersion(TransactionState transactionState, Database db) {
+    private void updateIvmSequencePrefix(TransactionState transactionState, Database db) {
         for (Long tableId : transactionState.getTableIdList()) {
             Table table = db.getTableNullable(tableId);
             if (table instanceof MTMV && ((MTMV) table).isIvm()) {
-                ((MTMV) table).getIvmInfo().advanceRefreshVersion();
+                ((MTMV) table).getIvmInfo().advanceSequencePrefix();
             }
         }
     }

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

@@ -70,8 +70,8 @@ class IvmAggDeltaHandlerTest extends IvmDeltaTestBase {
     private AggRewriteResult rewriteAgg(LogicalAggregate<? extends Plan> agg) {
         PlanBundle bundle = normalizeAggPlan(agg);
         MTMV mtmv = buildMtmvFromPlan(bundle.normalizedPlan.getOutput());
-        mtmv.getIvmInfo().advanceRefreshVersion();
-        mtmv.getIvmInfo().advanceRefreshVersion();
+        mtmv.getIvmInfo().advanceSequencePrefix();
+        mtmv.getIvmInfo().advanceSequencePrefix();
         Plan rewritten = new IvmDeltaRewriter().generateIncrRefreshPlan(
                 bundle.normalizedPlan, bundle.rewriteResult,
                 IvmRewriteContext.incremental(mtmv), bundle.connectContext);
@@ -89,8 +89,8 @@ private AggRewriteResult rewriteAggWithIdentityKeys(LogicalAggregate<? extends P
         PlanBundle bundle = normalizeAggPlan(agg);
         bundle.rewriteResult.setIdentityKeySlots(identityKeys);
         MTMV mtmv = buildMtmvFromPlan(bundle.normalizedPlan.getOutput());
-        mtmv.getIvmInfo().advanceRefreshVersion();
-        mtmv.getIvmInfo().advanceRefreshVersion();
+        mtmv.getIvmInfo().advanceSequencePrefix();
+        mtmv.getIvmInfo().advanceSequencePrefix();
         Plan rewritten = new IvmDeltaRewriter().generateIncrRefreshPlan(
                 bundle.normalizedPlan, bundle.rewriteResult,
                 IvmRewriteContext.incremental(mtmv), bundle.connectContext);

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

@@ -43,7 +43,7 @@
 class IvmDeltaRewriteStateTest extends IvmDeltaTestBase {
 
     @Test
-    void testSequenceEncodesRefreshVersionAndDeltaIndex() {
+    void testSequenceEncodesSequencePrefixAndDeltaIndex() {
         IvmDeltaRewriteState state = new IvmDeltaRewriteState(
                 ImmutableMap.of(), false, 7L, BigIntType.INSTANCE, ImmutableMap.of());
 
@@ -172,7 +172,7 @@ private void addTablet(OlapTable table, int partitionIndex, long tabletId) {
     }
 
     @Test
-    void testLargeIntSequenceEncodesRefreshVersionAndDeltaIndex() {
+    void testLargeIntSequenceEncodesSequencePrefixAndDeltaIndex() {
         IvmDeltaRewriteState state = new IvmDeltaRewriteState(
                 ImmutableMap.of(), false, 7L, LargeIntType.INSTANCE, ImmutableMap.of());
 

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

@@ -26,21 +26,31 @@
 
 class IvmInfoTest {
     @Test
-    void testRefreshVersionAdvancesAfterCommittedRefresh() {
+    void testSequencePrefixAdvancesAfterCommittedRefresh() {
         IvmInfo info = new IvmInfo();
 
-        Assertions.assertEquals(0, info.getRefreshVersion());
-        info.advanceRefreshVersion();
-        Assertions.assertEquals(1, info.getRefreshVersion());
+        Assertions.assertEquals(0, info.getSequencePrefix());
+        info.advanceSequencePrefix();
+        Assertions.assertEquals(1, info.getSequencePrefix());
     }
 
     @Test
-    void testRefreshVersionPersistsThroughGson() {
+    void testSequencePrefixPersistsThroughGson() {
         IvmInfo info = new IvmInfo();
-        info.advanceRefreshVersion();
+        info.advanceSequencePrefix();
 
         IvmInfo recovered = GsonUtils.GSON.fromJson(GsonUtils.GSON.toJson(info), IvmInfo.class);
-        Assertions.assertEquals(1, recovered.getRefreshVersion());
+        Assertions.assertEquals(1, recovered.getSequencePrefix());
+    }
+
+    @Test
+    void testSequencePrefixIsPersistedAsSp() {
+        IvmInfo info = new IvmInfo();
+        info.advanceSequencePrefix();
+
+        String json = GsonUtils.GSON.toJson(info);
+        Assertions.assertTrue(json.contains("\"sp\":1"), json);
+        Assertions.assertFalse(json.contains("\"rv\""), json);
     }
 
     @Test
@@ -59,7 +69,7 @@ void testCopyConstructor() {
         info.requireCompleteBaselineRebuild();
         info.setUseFullKeys(true);
         info.setPlanSignature("abc123");
-        info.advanceRefreshVersion();
+        info.advanceSequencePrefix();
 
         IvmInfo copy = new IvmInfo(info);
         info.clearBaselineRebuild();
@@ -68,7 +78,7 @@ void testCopyConstructor() {
         Assertions.assertTrue(copy.isBaselineRebuildRequired());
         Assertions.assertTrue(copy.isUseFullKeys());
         Assertions.assertEquals("abc123", copy.getPlanSignature());
-        Assertions.assertEquals(1, copy.getRefreshVersion());
+        Assertions.assertEquals(1, copy.getSequencePrefix());
     }
 
     @Test

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

@@ -57,9 +57,9 @@ void testLargeIntSequence() {
 
     @Test
     void testLargeIntSequenceUsesFullEncodingRange() {
-        long maxRefreshVersion = Long.MAX_VALUE >>> 11;
+        long maxSequencePrefix = Long.MAX_VALUE >>> 11;
         IvmSequenceCalculator calculator = IvmSequenceCalculator.create(
-                maxRefreshVersion, LargeIntType.INSTANCE);
+                maxSequencePrefix, LargeIntType.INSTANCE);
 
         LargeIntLiteral sequence = (LargeIntLiteral) calculator.encode(
                 1023, BigInteger.ONE.shiftLeft(64).subtract(BigInteger.ONE), true);
@@ -70,12 +70,12 @@ void testLargeIntSequenceUsesFullEncodingRange() {
 
     @Test
     void testSequenceRejectsValuesOutsideEncodingRanges() {
-        long maxRefreshVersion = Long.MAX_VALUE >>> 11;
-        IvmSequenceCalculator.create(maxRefreshVersion, LargeIntType.INSTANCE);
+        long maxSequencePrefix = Long.MAX_VALUE >>> 11;
+        IvmSequenceCalculator.create(maxSequencePrefix, LargeIntType.INSTANCE);
 
-        IvmException refreshVersionException = Assertions.assertThrows(IvmException.class,
-                () -> IvmSequenceCalculator.create(maxRefreshVersion + 1, LargeIntType.INSTANCE));
-        Assertions.assertTrue(refreshVersionException.getMessage().contains("refresh version"));
+        IvmException sequencePrefixException = Assertions.assertThrows(IvmException.class,
+                () -> IvmSequenceCalculator.create(maxSequencePrefix + 1, LargeIntType.INSTANCE));
+        Assertions.assertTrue(sequencePrefixException.getMessage().contains("sequence prefix"));
 
         IvmSequenceCalculator calculator = IvmSequenceCalculator.create(1, LargeIntType.INSTANCE);
         IvmException deltaIndexException = Assertions.assertThrows(IvmException.class,

fe/fe-core/src/test/java/org/apache/doris/transaction/DatabaseTransactionMgrTest.java

@@ -503,7 +503,7 @@ public void testAbortTransactionWithNotFoundException() throws UserException {
     }
 
     @Test
-    public void testUpdateCatalogAfterCommittedAdvancesIvmRefreshVersionForNormalCommitAndReplay()
+    public void testUpdateCatalogAfterCommittedAdvancesIvmSequencePrefixForNormalCommitAndReplay()
             throws Exception {
         DatabaseTransactionMgr masterDbTransMgr = masterTransMgr.getDatabaseTransactionMgr(CatalogTestUtil.testDbId1);
         Database masterDb = masterEnv.getInternalCatalog().getDbOrMetaException(CatalogTestUtil.testDbId1);
@@ -520,7 +520,7 @@ public void testUpdateCatalogAfterCommittedAdvancesIvmRefreshVersionForNormalCom
         method.setAccessible(true);
         method.invoke(masterDbTransMgr, normalCommitTxn, masterDb, false);
 
-        Assertions.assertEquals(1L, normalCommitIvmInfo.getRefreshVersion());
+        Assertions.assertEquals(1L, normalCommitIvmInfo.getSequencePrefix());
         Mockito.verify(normalCommitStream).unprotectedUpdateStreamUpdate(
                 normalCommitTxn.getStreamUpdateInfos().get(0).getUpdate(), normalCommitTxn.getCommitTime());
 
@@ -535,7 +535,7 @@ public void testUpdateCatalogAfterCommittedAdvancesIvmRefreshVersionForNormalCom
                 slaveDb.getId(), 9001L, 10002L, replayStreamId, 456L);
         method.invoke(slaveDbTransMgr, replayTxn, slaveDb, true);
 
-        Assertions.assertEquals(1L, replayIvmInfo.getRefreshVersion());
+        Assertions.assertEquals(1L, replayIvmInfo.getSequencePrefix());
         Mockito.verify(replayStream).unprotectedUpdateStreamUpdate(
                 replayTxn.getStreamUpdateInfos().get(0).getUpdate(), replayTxn.getCommitTime());
     }