[draft](ivm) for run the test pipeline

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

@@ -1101,7 +1101,7 @@ refreshSchedule
     ;
 
 refreshMethod
-    : COMPLETE | AUTO
+    : COMPLETE | AUTO | INCREMENTAL
     ;
 
 mvPartition

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

@@ -59,12 +59,15 @@ 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 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__";
@@ -217,6 +220,10 @@ public Column(String name, Type type, boolean isKey, AggregateType aggregateType
                 false, null, null,  Sets.newHashSet(), null);
     }
 
+    public static boolean isIvmHiddenColumn(String columnName) {
+        return StringUtils.startsWith(columnName, IVM_HIDDEN_COLUMN_PREFIX);
+    }
+
     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/java/org/apache/doris/catalog/MTMV.java

@@ -44,9 +44,12 @@
 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;
@@ -56,6 +59,7 @@
 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,7 @@ public MTMV() {
         this.mvPartitionInfo = params.mvPartitionInfo;
         this.relation = params.relation;
         this.refreshSnapshot = new MTMVRefreshSnapshot();
+        this.ivmInfo = new IvmInfo();
         this.envInfo = new EnvInfo(-1L, -1L);
         this.sessionVariables = params.sessionVariables;
         mvRwLock = new ReentrantReadWriteLock(true);
@@ -437,6 +444,21 @@ public MTMVRefreshSnapshot getRefreshSnapshot() {
         return refreshSnapshot;
     }
 
+    public IvmInfo getIvmInfo() {
+        return ivmInfo;
+    }
+
+    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 {
@@ -609,6 +631,9 @@ private void compatibleInternal(CatalogMgr catalogMgr) throws Exception {
     @Override
     public void gsonPostProcess() throws IOException {
         super.gsonPostProcess();
+        if (ivmInfo == null) {
+            ivmInfo = new IvmInfo();
+        }
         Map<String, MTMVRefreshPartitionSnapshot> partitionSnapshots = refreshSnapshot.getPartitionSnapshots();
         compatiblePctSnapshot(partitionSnapshots);
     }

fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java

@@ -57,18 +57,16 @@
 import org.apache.doris.mtmv.MTMVRelatedTableIf;
 import org.apache.doris.mtmv.MTMVRelation;
 import org.apache.doris.mtmv.MTMVUtil;
+import org.apache.doris.mtmv.ivm.IvmRefreshManager;
+import org.apache.doris.mtmv.ivm.IvmRefreshResult;
 import org.apache.doris.nereids.StatementContext;
-import org.apache.doris.nereids.glue.LogicalPlanAdapter;
 import org.apache.doris.nereids.trees.plans.commands.UpdateMvByPartitionCommand;
-import org.apache.doris.qe.AuditLogHelper;
 import org.apache.doris.qe.ConnectContext;
-import org.apache.doris.qe.QueryState.MysqlStateType;
 import org.apache.doris.qe.StmtExecutor;
 import org.apache.doris.system.SystemInfoService;
 import org.apache.doris.thrift.TCell;
 import org.apache.doris.thrift.TRow;
 import org.apache.doris.thrift.TStatusCode;
-import org.apache.doris.thrift.TUniqueId;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
@@ -244,6 +242,21 @@ public void run() throws JobException {
             if (refreshMode == MTMVTaskRefreshMode.NOT_REFRESH) {
                 return;
             }
+            // Attempt IVM refresh for incremental MVs and fall back when the plan is unsupported.
+            if (mtmv.getRefreshInfo().getRefreshMethod() == RefreshMethod.INCREMENTAL) {
+                IvmRefreshManager ivmRefreshManager = new IvmRefreshManager();
+                IvmRefreshResult ivmResult = ivmRefreshManager.doRefresh(mtmv);
+                if (ivmResult.isSuccess()) {
+                    LOG.info("IVM incremental refresh succeeded for mv={}, taskId={}",
+                            mtmv.getName(), getTaskId());
+                    return;
+                }
+                LOG.warn("IVM refresh fell back for mv={}, reason={}, detail={}, taskId={}. "
+                        + "Continuing with partition-based refresh.",
+                        mtmv.getName(), ivmResult.getFallbackReason(),
+                        ivmResult.getDetailMessage(), getTaskId());
+                // TODO: it may cause too many full refresh, need limit full refresh here
+            }
             Map<TableIf, String> tableWithPartKey = getIncrementalTableMap();
             this.completedPartitions = Lists.newCopyOnWriteArrayList();
             int refreshPartitionNum = mtmv.getRefreshPartitionNum();
@@ -321,36 +334,23 @@ private void executeWithRetry(Set<String> execPartitionNames, Map<TableIf, Strin
     private void exec(Set<String> refreshPartitionNames,
             Map<TableIf, String> tableWithPartKey)
             throws Exception {
-        ConnectContext ctx = MTMVPlanUtil.createMTMVContext(mtmv, MTMVPlanUtil.DISABLE_RULES_WHEN_RUN_MTMV_TASK);
+        // Create MTMV context first so that new StatementContext() captures the
+        // correct thread-local ConnectContext (with MTMV disabled rules, etc.).
+        ConnectContext mtmvCtx = MTMVPlanUtil.createMTMVContext(mtmv, MTMVPlanUtil.DISABLE_RULES_WHEN_RUN_MTMV_TASK);
         StatementContext statementContext = new StatementContext();
         for (Entry<MvccTableInfo, MvccSnapshot> entry : snapshots.entrySet()) {
             statementContext.setSnapshot(entry.getKey(), entry.getValue());
         }
-        ctx.setStatementContext(statementContext);
-        TUniqueId queryId = generateQueryId();
-        lastQueryId = DebugUtil.printId(queryId);
         // if SELF_MANAGE mv, only have default partition,  will not have partitionItem, so we give empty set
         UpdateMvByPartitionCommand command = UpdateMvByPartitionCommand
                 .from(mtmv, mtmv.getMvPartitionInfo().getPartitionType() != MTMVPartitionType.SELF_MANAGE
                         ? refreshPartitionNames : Sets.newHashSet(), tableWithPartKey, statementContext);
-        try {
-            executor = new StmtExecutor(ctx, new LogicalPlanAdapter(command, ctx.getStatementContext()));
-            ctx.setExecutor(executor);
-            ctx.setQueryId(queryId);
-            ctx.getState().setNereids(true);
-            command.run(ctx, executor);
-            if (getStatus() == TaskStatus.CANCELED) {
-                // Throwing an exception to interrupt subsequent partition update tasks
-                throw new JobException("task is CANCELED");
-            }
-            if (ctx.getState().getStateType() != MysqlStateType.OK) {
-                throw new JobException(ctx.getState().getErrorMessage());
-            }
-        } finally {
-            if (executor != null) {
-                AuditLogHelper.logAuditLog(ctx, getDummyStmt(refreshPartitionNames),
-                        executor.getParsedStmt(), executor.getQueryStatisticsForAuditLog(), true);
-            }
+        boolean enableIvmNormalMTMVPlan = mtmv.getRefreshInfo().getRefreshMethod() == RefreshMethod.INCREMENTAL;
+        executor = MTMVPlanUtil.executeCommand(mtmvCtx, command, statementContext,
+                getDummyStmt(refreshPartitionNames), enableIvmNormalMTMVPlan);
+        lastQueryId = DebugUtil.printId(executor.getContext().queryId());
+        if (getStatus() == TaskStatus.CANCELED) {
+            throw new JobException("task is CANCELED");
         }
     }
 

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

@@ -17,6 +17,8 @@
 
 package org.apache.doris.mtmv;
 
+import org.apache.doris.mtmv.ivm.IvmNormalizeResult;
+import org.apache.doris.nereids.trees.plans.Plan;
 import org.apache.doris.nereids.trees.plans.commands.info.ColumnDefinition;
 
 import java.util.List;
@@ -25,6 +27,8 @@ public class MTMVAnalyzeQueryInfo {
     private MTMVRelation relation;
     private MTMVPartitionInfo mvPartitionInfo;
     private List<ColumnDefinition> columnDefinitions;
+    // set when IVM normalization is enabled; carries normalizedPlan + aggMeta
+    private IvmNormalizeResult ivmNormalizeResult;
 
     public MTMVAnalyzeQueryInfo(List<ColumnDefinition> columnDefinitions, MTMVPartitionInfo mvPartitionInfo,
             MTMVRelation relation) {
@@ -44,4 +48,17 @@ public MTMVPartitionInfo getMvPartitionInfo() {
     public MTMVRelation getRelation() {
         return relation;
     }
+
+    public IvmNormalizeResult getIvmNormalizeResult() {
+        return ivmNormalizeResult;
+    }
+
+    public void setIvmNormalizeResult(IvmNormalizeResult ivmNormalizeResult) {
+        this.ivmNormalizeResult = ivmNormalizeResult;
+    }
+
+    /** Convenience accessor — returns the normalized plan, or null if IVM is not active. */
+    public Plan getIvmNormalizedPlan() {
+        return ivmNormalizeResult != null ? ivmNormalizeResult.getNormalizedPlan() : null;
+    }
 }