[draft](ivm) Add mv ivm test for pipeline

fe/fe-catalog/src/main/java/org/apache/doris/catalog/Column.java

@@ -51,12 +51,23 @@ public class Column implements GsonPostProcessable {
     public static final String HIDDEN_COLUMN_PREFIX = "__DORIS_";
     // all shadow indexes should have this prefix in name
     public static final String SHADOW_NAME_PREFIX = "__doris_shadow_";
+    public static final String IVM_HIDDEN_COLUMN_PREFIX = "__DORIS_IVM_";
     // NOTE: you should name hidden column start with '__DORIS_' !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
     public static final String DELETE_SIGN = "__DORIS_DELETE_SIGN__";
     public static final String WHERE_SIGN = "__DORIS_WHERE_SIGN__";
     public static final String SEQUENCE_COL = "__DORIS_SEQUENCE_COL__";
     public static final String ROWID_COL = "__DORIS_ROWID_COL__";
     public static final String GLOBAL_ROWID_COL = "__DORIS_GLOBAL_ROWID_COL__";
+    public static final String IVM_ROW_ID_COL = "__DORIS_IVM_ROW_ID_COL__";
+    public static final String IVM_AGG_COUNT_COL = "__DORIS_IVM_AGG_COUNT_COL__";
+    public static final String IVM_DML_FACTOR_COL = "__DORIS_IVM_DML_FACTOR_COL__";
+    public static final String IVM_DELTA_GROUP_COUNT_COL = "__DORIS_IVM_DELTA_GROUP_COUNT_COL__";
+    /**
+     * Name of the binlog operation column in the base table that indicates insert (0) vs delete (1).
+     * When present, IVM delta rewrite derives dml_factor from this column instead of using the
+     * hardcoded literal 1. Follows the same convention as {@link #DELETE_SIGN}: 0 = insert, 1 = delete.
+     */
+    public static final String BINLOG_OPERATION_COL = "binlog_op";
     public static final String ROW_STORE_COL = "__DORIS_ROW_STORE_COL__";
     public static final String VERSION_COL = "__DORIS_VERSION_COL__";
     public static final String SKIP_BITMAP_COL = "__DORIS_SKIP_BITMAP_COL__";
@@ -209,6 +220,7 @@ public Column(String name, Type type, boolean isKey, AggregateType aggregateType
                 false, null, null,  Sets.newHashSet(), null);
     }
 
+
     public Column(String name, Type type, boolean isKey, AggregateType aggregateType, boolean isAllowNull,
                   String defaultValue, String comment, boolean visible, int colUniqueId) {
         this(name, type, isKey, aggregateType, isAllowNull, -1, defaultValue, comment, visible, null, colUniqueId, null,

fe/fe-core/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4

@@ -82,7 +82,8 @@ materializedViewStatement
         (BUCKETS (INTEGER_VALUE | AUTO))?)?
         propertyClause?
         AS? query                                                                               #createMTMV
-    | REFRESH MATERIALIZED VIEW mvName=multipartIdentifier (partitionSpec | COMPLETE | AUTO)    #refreshMTMV
+    | REFRESH MATERIALIZED VIEW mvName=multipartIdentifier
+        (partitionSpec | COMPLETE | AUTO | INCREMENTAL | PARTITIONS)                            #refreshMTMV
     | ALTER MATERIALIZED VIEW mvName=multipartIdentifier ((RENAME newName=identifier)
         | (REFRESH (refreshMethod | refreshTrigger | refreshMethod refreshTrigger))
         | REPLACE WITH MATERIALIZED VIEW newName=identifier propertyClause?
@@ -1114,7 +1115,7 @@ refreshSchedule
     ;
 
 refreshMethod
-    : COMPLETE | AUTO
+    : COMPLETE | AUTO | INCREMENTAL
     ;
 
 mvPartition

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

@@ -1263,6 +1263,9 @@ public void processAlterMTMV(AlterMTMV alterMTMV, boolean isReplay) {
                         mtmv.compatible(Env.getCurrentEnv().getCatalogMgr());
                     }
                     break;
+                case ALTER_IVM_INFO:
+                    mtmv.alterIvmInfo(alterMTMV.getIvmInfo());
+                    break;
                 default:
                     throw new RuntimeException("Unknown type value: " + alterMTMV.getOpType());
             }

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

@@ -160,6 +160,8 @@
 import org.apache.doris.mtmv.MTMVService;
 import org.apache.doris.mtmv.MTMVStatus;
 import org.apache.doris.mtmv.MTMVUtil;
+import org.apache.doris.mtmv.ivm.IvmInfo;
+import org.apache.doris.mtmv.ivm.IvmUtil;
 import org.apache.doris.mysql.authenticate.AuthenticateType;
 import org.apache.doris.mysql.authenticate.AuthenticatorManager;
 import org.apache.doris.mysql.privilege.AccessControllerManager;
@@ -3755,16 +3757,29 @@ public static String getMTMVDdl(MTMV mtmv) throws AnalysisException {
                     "get table read lock timeout, database=" + mtmv.getDBName() + ",table=" + mtmv.getName());
         }
         try {
+            boolean isIvm = mtmv.isIvm();
             StringBuilder sb = new StringBuilder("CREATE MATERIALIZED VIEW ");
             sb.append(mtmv.getName());
-            addColNameAndComment(mtmv, sb);
+            addColNameAndComment(mtmv, sb, isIvm);
             sb.append("\n");
             sb.append(mtmv.getRefreshInfo());
-            addMTMVKeyInfo(mtmv, sb);
+            if (!isIvm) {
+                addMTMVKeyInfo(mtmv, sb);
+            }
             addTableComment(mtmv, sb);
             addMTMVPartitionInfo(mtmv, sb);
             DistributionInfo distributionInfo = mtmv.getDefaultDistributionInfo();
-            sb.append("\n").append(distributionInfo.toSql());
+            if (isIvm) {
+                // IVM internally rewrites distribution to HASH(__DORIS_IVM_ROW_ID_COL__),
+                // which is a hidden column invisible to users.  Output DISTRIBUTED BY RANDOM
+                // with the same bucket count / auto-bucket setting, so the DDL is re-executable
+                // and preserves the bucket configuration.  On re-creation, the IVM pipeline
+                // will rewrite RANDOM to HASH(row_id) again automatically.
+                sb.append("\n").append(new RandomDistributionInfo(
+                        distributionInfo.getBucketNum(), distributionInfo.getAutoBucket()).toSql());
+            } else {
+                sb.append("\n").append(distributionInfo.toSql());
+            }
             // properties
             sb.append("\nPROPERTIES (\n");
             addOlapTablePropertyInfo(mtmv, sb, false, false, null);
@@ -3807,13 +3822,22 @@ private static void addMTMVPartitionInfo(MTMV mtmv, StringBuilder sb) throws Ana
     }
 
     private static void addColNameAndComment(TableIf tableIf, StringBuilder sb) {
+        addColNameAndComment(tableIf, sb, false);
+    }
+
+    private static void addColNameAndComment(TableIf tableIf, StringBuilder sb, boolean filterIvmHiddenCols) {
         sb.append("\n(");
         List<Column> columns = tableIf.getBaseSchema();
+        boolean first = true;
         for (int i = 0; i < columns.size(); i++) {
-            if (i != 0) {
+            Column column = columns.get(i);
+            if (filterIvmHiddenCols && IvmUtil.isIvmHiddenColumn(column.getName())) {
+                continue;
+            }
+            if (!first) {
                 sb.append(",");
             }
-            Column column = columns.get(i);
+            first = false;
             sb.append(column.getName());
             if (!StringUtils.isEmpty(column.getComment())) {
                 sb.append(" comment '");
@@ -3934,14 +3958,16 @@ private static void addOlapTablePropertyInfo(OlapTable olapTable, StringBuilder
         }
 
         // unique key table with merge on write, always print this property for unique table
-        if (olapTable.getKeysType() == KeysType.UNIQUE_KEYS) {
+        // but hide it for IVM materialized views (internal physical detail)
+        boolean isIvmMtmv = olapTable instanceof MTMV && ((MTMV) olapTable).isIvm();
+        if (olapTable.getKeysType() == KeysType.UNIQUE_KEYS && !isIvmMtmv) {
             sb.append(",\n\"").append(PropertyAnalyzer.ENABLE_UNIQUE_KEY_MERGE_ON_WRITE).append("\" = \"");
             sb.append(olapTable.getEnableUniqueKeyMergeOnWrite()).append("\"");
         }
 
         // enable_unique_key_skip_bitmap, always print this property for merge-on-write unique table
         if (olapTable.getKeysType() == KeysType.UNIQUE_KEYS && olapTable.getEnableUniqueKeyMergeOnWrite()
-                && olapTable.getEnableUniqueKeySkipBitmap()) {
+                && olapTable.getEnableUniqueKeySkipBitmap() && !isIvmMtmv) {
             sb.append(",\n\"").append(PropertyAnalyzer.ENABLE_UNIQUE_KEY_SKIP_BITMAP_COLUMN).append("\" = \"");
             sb.append(olapTable.getEnableUniqueKeySkipBitmap()).append("\"");
         }
@@ -7508,6 +7534,12 @@ public void addMTMVTaskResult(TableNameInfo mvName, MTMVTask task, MTMVRelation
         this.alter.processAlterMTMV(alter, false);
     }
 
+    public void alterMTMVIvmInfo(TableNameInfo mvName, IvmInfo ivmInfo) {
+        AlterMTMV alter = new AlterMTMV(mvName, MTMVAlterOpType.ALTER_IVM_INFO);
+        alter.setIvmInfo(ivmInfo);
+        this.alter.processAlterMTMV(alter, false);
+    }
+
     // Ensure the env is ready, otherwise throw an exception.
     public void checkReadyOrThrow() throws Exception {
         if (isReady()) {

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

@@ -35,6 +35,7 @@
 import org.apache.doris.mtmv.MTMVPartitionInfo.MTMVPartitionType;
 import org.apache.doris.mtmv.MTMVPartitionUtil;
 import org.apache.doris.mtmv.MTMVPlanUtil;
+import org.apache.doris.mtmv.MTMVPropertyUtil;
 import org.apache.doris.mtmv.MTMVRefreshEnum.MTMVRefreshState;
 import org.apache.doris.mtmv.MTMVRefreshEnum.MTMVState;
 import org.apache.doris.mtmv.MTMVRefreshInfo;
@@ -44,18 +45,21 @@
 import org.apache.doris.mtmv.MTMVRelation;
 import org.apache.doris.mtmv.MTMVSnapshotIf;
 import org.apache.doris.mtmv.MTMVStatus;
+import org.apache.doris.mtmv.ivm.IvmInfo;
+import org.apache.doris.mtmv.ivm.IvmUtil;
 import org.apache.doris.nereids.rules.analysis.SessionVarGuardRewriter;
 import org.apache.doris.qe.ConnectContext;
 
+import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
-import com.google.common.collect.Sets;
 import com.google.gson.annotations.SerializedName;
 import org.apache.commons.collections4.MapUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
 import java.io.IOException;
+import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Optional;
@@ -87,6 +91,8 @@ public class MTMV extends OlapTable {
     private MTMVPartitionInfo mvPartitionInfo;
     @SerializedName("rs")
     private MTMVRefreshSnapshot refreshSnapshot;
+    @SerializedName("ii")
+    private IvmInfo ivmInfo;
     // Should update after every fresh, not persist
     // Cache with SessionVarGuardExpr: used when query session variables differ from MV creation variables
     private MTMVCache cacheWithGuard;
@@ -120,6 +126,8 @@ public MTMV() {
         this.mvPartitionInfo = params.mvPartitionInfo;
         this.relation = params.relation;
         this.refreshSnapshot = new MTMVRefreshSnapshot();
+        this.ivmInfo = new IvmInfo();
+        this.ivmInfo.setEnableIvm(params.enableIvm);
         this.envInfo = new EnvInfo(-1L, -1L);
         this.sessionVariables = params.sessionVariables;
         mvRwLock = new ReentrantReadWriteLock(true);
@@ -207,6 +215,10 @@ public void processBaseViewChange(String schemaChangeDetail) {
         }
     }
 
+    public boolean isIvm() {
+        return getIvmInfo().isEnableIvm();
+    }
+
     public boolean addTaskResult(MTMVTask task, MTMVRelation relation,
             Map<String, MTMVRefreshPartitionSnapshot> partitionSnapshots, boolean isReplay) {
         MTMVCache mtmvCacheWithGuard = null;
@@ -334,36 +346,20 @@ public int getRefreshPartitionNum() {
     }
 
     public Set<TableNameInfo> getExcludedTriggerTables() {
-        Set<TableNameInfo> res = Sets.newHashSet();
         readMvLock();
         try {
-            if (StringUtils.isEmpty(mvProperties.get(PropertyAnalyzer.PROPERTIES_EXCLUDED_TRIGGER_TABLES))) {
-                return res;
-            }
-            String[] split = mvProperties.get(PropertyAnalyzer.PROPERTIES_EXCLUDED_TRIGGER_TABLES).split(",");
-            for (String alias : split) {
-                res.add(new TableNameInfo(alias));
-            }
-            return res;
+            return MTMVPropertyUtil.parseTableNameInfos(
+                    mvProperties.get(PropertyAnalyzer.PROPERTIES_EXCLUDED_TRIGGER_TABLES));
         } finally {
             readMvUnlock();
         }
     }
 
     public Set<TableNameInfo> getQueryRewriteConsistencyRelaxedTables() {
-        Set<TableNameInfo> res = Sets.newHashSet();
         readMvLock();
         try {
-            String stillRewrittenTables
-                    = mvProperties.get(PropertyAnalyzer.ASYNC_MV_QUERY_REWRITE_CONSISTENCY_RELAXED_TABLES);
-            if (StringUtils.isEmpty(stillRewrittenTables)) {
-                return res;
-            }
-            String[] split = stillRewrittenTables.split(",");
-            for (String alias : split) {
-                res.add(new TableNameInfo(alias));
-            }
-            return res;
+            return MTMVPropertyUtil.parseTableNameInfos(
+                    mvProperties.get(PropertyAnalyzer.ASYNC_MV_QUERY_REWRITE_CONSISTENCY_RELAXED_TABLES));
         } finally {
             readMvUnlock();
         }
@@ -437,6 +433,38 @@ public MTMVRefreshSnapshot getRefreshSnapshot() {
         return refreshSnapshot;
     }
 
+    public IvmInfo getIvmInfo() {
+        writeMvLock();
+        try {
+            if (ivmInfo == null) {
+                ivmInfo = new IvmInfo();
+            }
+            return ivmInfo;
+        } finally {
+            writeMvUnlock();
+        }
+    }
+
+    public void alterIvmInfo(IvmInfo ivmInfo) {
+        writeMvLock();
+        try {
+            this.ivmInfo = ivmInfo;
+        } finally {
+            writeMvUnlock();
+        }
+    }
+
+    public List<String> getInsertedColumnNames()  {
+        List<Column> columns = getBaseSchema(true);
+        List<String> columnNames = Lists.newArrayListWithExpectedSize(columns.size());
+        for (Column column : columns) {
+            if (column.isVisible() || IvmUtil.isIvmHiddenColumn(column.getName())) {
+                columnNames.add(column.getName());
+            }
+        }
+        return columnNames;
+    }
+
     public long getSchemaChangeVersion() {
         readMvLock();
         try {
@@ -612,6 +640,9 @@ public void gsonPostProcess() throws IOException {
         if (sessionVariables == null) {
             sessionVariables = Maps.newHashMap();
         }
+        if (ivmInfo == null) {
+            ivmInfo = new IvmInfo();
+        }
         Map<String, MTMVRefreshPartitionSnapshot> partitionSnapshots = refreshSnapshot.getPartitionSnapshots();
         compatiblePctSnapshot(partitionSnapshots);
     }

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

@@ -3468,6 +3468,17 @@ public long getVisibleVersion() throws RpcException {
         }
     }
 
+    /**
+     * Returns the visible TSO (timestamp ordering) for this table.
+     *
+     * <p>MOCK: currently returns {@link #getVisibleVersion()} directly. This will be
+     * replaced with a real TSO-based API once the storage layer supports it.
+     * Each partition's visible version and TSO will be in one-to-one correspondence.
+     */
+    public long getVisibleTso() throws RpcException {
+        return getVisibleVersion();
+    }
+
     // Get the table versions in batch.
     public static List<Long> getVisibleVersionInBatch(List<OlapTable> tables) {
         if (tables.isEmpty()) {

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

@@ -52,6 +52,7 @@ public OlapTableParams(boolean isTemporary) {
     public static class MTMVParams extends BuildParams {
         public MTMVRefreshInfo refreshInfo;
         public String querySql;
+        public boolean enableIvm;
         public Map<String, String> mvProperties;
         public MTMVPartitionInfo mvPartitionInfo;
         public MTMVRelation relation;
@@ -155,6 +156,14 @@ public OlapTableFactory withMvProperties(Map<String, String> mvProperties) {
         return this;
     }
 
+    public OlapTableFactory withEnableIvm(boolean enableIvm) {
+        Preconditions.checkState(params instanceof MTMVParams, "Invalid argument for "
+                + params.getClass().getSimpleName());
+        MTMVParams mtmvParams = (MTMVParams) params;
+        mtmvParams.enableIvm = enableIvm;
+        return this;
+    }
+
     private OlapTableFactory withRefreshInfo(MTMVRefreshInfo refreshInfo) {
         Preconditions.checkState(params instanceof MTMVParams, "Invalid argument for "
                 + params.getClass().getSimpleName());
@@ -194,6 +203,7 @@ public OlapTableFactory withExtraParams(CreateTableInfo createTableInfo) {
         } else {
             CreateMTMVInfo createMTMVInfo = (CreateMTMVInfo) createTableInfo;
             return withRefreshInfo(createMTMVInfo.getRefreshInfo())
+                .withEnableIvm(createMTMVInfo.isEnableIvm())
                 .withQuerySql(createMTMVInfo.getQuerySql())
                 .withMvProperties(createMTMVInfo.getMvProperties())
                 .withMvPartitionInfo(createMTMVInfo.getMvPartitionInfo())

fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java

@@ -1612,7 +1612,10 @@ public void addPartition(Database db, String tableName, AddPartitionOp addPartit
             partitionInfo.createAndCheckPartitionItem(singlePartitionDesc, isTempPartition);
 
             // get distributionInfo
-            List<Column> baseSchema = olapTable.getBaseSchema();
+            // Use full schema (including hidden columns) so that distribution columns
+            // such as __DORIS_IVM_ROW_ID_COL__ (hidden but used as hash distribution key
+            // for IVM materialized views) can be resolved correctly.
+            List<Column> baseSchema = olapTable.getBaseSchema(true);
             DistributionInfo defaultDistributionInfo = olapTable.getDefaultDistributionInfo();
             if (distributionDesc != null) {
                 distributionInfo = distributionDesc.toDistributionInfo(baseSchema);