[Enhancement] IVM: re-derive the maintenance query at refresh instead of freezing it (backport #74881)

fe/fe-core/src/main/java/com/starrocks/catalog/MaterializedView.java

@@ -881,8 +881,12 @@ public String getTaskDefinition() {
         return formatInsertSql("insert overwrite");
     }
 
-    public String getIVMTaskDefinition() {
-        return formatInsertSql("INSERT INTO");
+    public String getTaskDefinition(String selectSql) {
+        return formatInsertSql("insert overwrite", selectSql);
+    }
+
+    public String getIVMTaskDefinition(String selectSql) {
+        return formatInsertSql("INSERT INTO", selectSql);
     }
 
     /**
@@ -892,10 +896,14 @@ public String getIVMTaskDefinition() {
      * which {@code InsertAnalyzer} rejects if listed explicitly.
      */
     private String formatInsertSql(String insertKeyword) {
+        return formatInsertSql(insertKeyword, getMVQueryDefinedSql());
+    }
+
+    private String formatInsertSql(String insertKeyword, String selectSql) {
         String targetSpec = hasAutoIncrementColumn()
                 ? String.format("`%s` (%s)", getName(), queryProducedColumnList())
                 : String.format("`%s`", getName());
-        return String.format("%s %s %s", insertKeyword, targetSpec, getMVQueryDefinedSql());
+        return String.format("%s %s %s", insertKeyword, targetSpec, selectSql);
     }
 
     /**

fe/fe-core/src/main/java/com/starrocks/scheduler/mv/MVRefreshSchemaChecker.java

@@ -16,7 +16,6 @@
 
 import com.google.common.base.Strings;
 import com.starrocks.catalog.BaseTableInfo;
-import com.starrocks.catalog.Column;
 import com.starrocks.catalog.Database;
 import com.starrocks.catalog.MaterializedView;
 import com.starrocks.catalog.Table;
@@ -25,20 +24,18 @@
 import com.starrocks.server.CatalogMgr;
 import com.starrocks.server.GlobalStateMgr;
 import com.starrocks.sql.analyzer.Analyzer;
-import com.starrocks.sql.analyzer.AnalyzerUtils;
 import com.starrocks.sql.analyzer.Field;
 import com.starrocks.sql.analyzer.SemanticException;
+import com.starrocks.sql.analyzer.mv.IvmRefreshDefinition;
+import com.starrocks.sql.analyzer.mv.IvmSchemaCompat;
 import com.starrocks.sql.ast.QueryStatement;
 import com.starrocks.sql.ast.StatementBase;
 import com.starrocks.sql.optimizer.rule.transformation.materialization.MvUtils;
 import com.starrocks.sql.parser.SqlParser;
 import com.starrocks.statistic.StatisticUtils;
-import com.starrocks.type.ScalarType;
-import com.starrocks.type.Type;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
-import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 
@@ -83,10 +80,10 @@ public static void checkExternalBaseSchemaCompat(MaterializedView mv) {
             return;
         }
 
-        // Use the same query refresh actually runs: ivmDefineSql for IVM (post-rewrite, with
-        // synthetic state-column expressions like sum_state_merge(...)), and viewDefineSql
-        // for PCT via the fallback in getMVQueryDefinedSql.
-        String selectSql = mv.getMVQueryDefinedSql();
+        // IVM derives its rewritten query (with hidden __ROW_ID__ / __AGG_STATE_* columns) at
+        // refresh time, so the schema check must compare against that same rewrite, not the
+        // original user query. PCT compares against viewDefineSql directly.
+        String selectSql = mv.getViewDefineSql();
         if (Strings.isNullOrEmpty(selectSql)) {
             return;
         }
@@ -127,79 +124,34 @@ private static boolean isLikelyDriftException(SemanticException e) {
         return msg.contains("cannot be resolved")
                 || msg.contains("column schema not compatible")
                 || msg.contains("base table schema changed for columns")
-                || msg.contains("No matching function with signature");
+                || msg.contains("No matching function with signature")
+                // IVM re-derive incompatibility (row-id strategy flip / no longer IVM-rewritable) is a
+                // drop-and-recreate condition -- inactivate with the reason, don't retry forever.
+                || msg.contains("row-id strategy")
+                || msg.contains("is not an IVM query");
     }
 
     private static void analyzeAndCompareSchema(MaterializedView mv, ConnectContext context, String selectSql) {
         Optional<Database> mvDb = GlobalStateMgr.getCurrentState().getLocalMetastore().mayGetDb(mv.getDbId());
         mvDb.map(Database::getFullName).ifPresent(context::setDatabase);
 
-        List<StatementBase> statements = SqlParser.parse(selectSql, context.getSessionVariable());
-        if (statements.isEmpty() || !(statements.get(0) instanceof QueryStatement)) {
-            return;
-        }
-        QueryStatement queryStmt = (QueryStatement) statements.get(0);
-        Analyzer.analyze(queryStmt, context);
-
-        // Align by position, not name: stored MV cols and analyzer Fields are both in the
-        // SELECT-list output order, so position i pairs them. Name lookup would break MVs
-        // created with explicit column aliases — CREATE MV (x, y) AS SELECT k, v — where the
-        // stored column is named x but the analyzer Field is named k. Position is the only
-        // stable correspondence.
-        // Hidden stored cols (IVM __ROW_ID__, __AGG_STATE_<func>) keep their position but are
-        // skipped — their analyzer-derived nullability diverges from stored NOT NULL on
-        // unchanged schemas. Drift that affects the user-visible projection is caught via
-        // "cannot be resolved"; a base widening that keeps the user-visible output type
-        // unchanged but mutates the hidden state type is not detected here.
-        List<Field> derivedFields = queryStmt.getQueryRelation().getRelationFields().getAllFields();
-        List<Column> orderedAll = mv.getOrderedOutputColumns(true);
-        if (derivedFields.size() != orderedAll.size()) {
-            // Stored col count must match analyzer output of the persisted SELECT. A mismatch
-            // implies base schema changed in a way that altered the SELECT's output arity
-            // (DROP of a SELECT * column expanded at CREATE time would normally throw
-            // "cannot be resolved" first; this is a defensive catch).
-            throw new SemanticException(MaterializedViewExceptions.inactiveReasonForColumnChanged(
-                    Collections.singleton("column count " + orderedAll.size() + " vs " + derivedFields.size())));
-        }
-        for (int i = 0; i < orderedAll.size(); i++) {
-            Column existed = orderedAll.get(i);
-            if (existed.isHidden()) {
-                continue;
-            }
-            Field derivedField = derivedFields.get(i);
-            Type derivedNormalized = AnalyzerUtils.transformTableColumnType(derivedField.getType(), false);
-            if (!isColumnCompatible(existed, derivedNormalized)) {
-                // Render derived as a Column only for the error message — gives the canonical
-                // OLAP `(name type NULL/NOT NULL ...)` format.
-                Column derived = new Column(existed.getName(), derivedNormalized, derivedField.isNullable());
-                String reason = MaterializedViewExceptions.inactiveReasonForColumnNotCompatible(
-                        existed.toString(), derived.toString());
-                throw new SemanticException(reason);
+        QueryStatement queryStmt;
+        // Require BOTH an IVM refresh mode AND the __ROW_ID__ schema -- a real IVM MV has both. Mode alone
+        // misfires on a non-IVM-eligible AUTO MV (plain PCT schema); the __ROW_ID__ column alone misfires
+        // on a PCT MV that merely outputs a column named __ROW_ID__ (not a reserved name). Either misfire
+        // makes derive() throw -> the healthy MV is falsely inactivated.
+        if (mv.getCurrentRefreshMode().isIncrementalOrAuto() && mv.getRowIdStrategy() != null) {
+            queryStmt = IvmRefreshDefinition.deriveRewrittenQuery(context, mv);
+        } else {
+            List<StatementBase> statements = SqlParser.parse(selectSql, context.getSessionVariable());
+            if (statements.isEmpty() || !(statements.get(0) instanceof QueryStatement)) {
+                return;
             }
+            queryStmt = (QueryStatement) statements.get(0);
+            Analyzer.analyze(queryStmt, context);
         }
-    }
 
-    // matchesType recurses through ARRAY / MAP / STRUCT — catches inner-type drift like
-    // STRUCT<a INT> → STRUCT<a BIGINT> (Spark/Trino ALTER on iceberg) that PrimitiveType.equals
-    // misses because non-scalars all return INVALID_TYPE. CHAR/VARCHAR widths stay
-    // interchangeable to avoid false positives on iceberg `string`.
-    // Nullability skipped: analyzer's NULL on rewritten IVM SELECT diverges from stored NOT NULL.
-    // Decimal precision: matchesType only buckets by primitive (DECIMAL32/64/128/256), so
-    // DECIMAL(10,2) → DECIMAL(18,2) (both DECIMAL64, same scale) would pass — overflow at
-    // refresh INSERT instead. Explicit precision check guards against this.
-    // Visible for MVRefreshSchemaCheckerTest.
-    static boolean isColumnCompatible(Column existed, Type derivedType) {
-        if (!existed.getType().matchesType(derivedType)) {
-            return false;
-        }
-        if (existed.getType().isDecimalOfAnyVersion() && derivedType.isDecimalOfAnyVersion()) {
-            ScalarType existedScalar = (ScalarType) existed.getType();
-            ScalarType derivedScalar = (ScalarType) derivedType;
-            if (existedScalar.getScalarPrecision() != derivedScalar.getScalarPrecision()
-                    || existedScalar.getScalarScale() != derivedScalar.getScalarScale()) {
-                return false;
-            }
-        }
-        return true;
+        List<Field> derivedFields = queryStmt.getQueryRelation().getRelationFields().getAllFields();
+        IvmSchemaCompat.compare(derivedFields, mv);
     }
 }

fe/fe-core/src/main/java/com/starrocks/scheduler/mv/ivm/MVIVMRefreshProcessor.java

@@ -57,6 +57,7 @@
 import com.starrocks.sql.analyzer.PlannerMetaLocker;
 import com.starrocks.sql.analyzer.SemanticException;
 import com.starrocks.sql.analyzer.mv.IVMAnalyzer;
+import com.starrocks.sql.analyzer.mv.IvmRefreshDefinition;
 import com.starrocks.sql.ast.InsertStmt;
 import com.starrocks.sql.ast.QueryStatement;
 import com.starrocks.sql.ast.TableRelation;
@@ -495,21 +496,23 @@ private InsertStmt prepareRefreshPlan() throws AnalysisException, LockTimeoutExc
                     .collect(Collectors.toSet());
             changeDefaultConnectContextIfNeeded(ctx, baseTables);
 
-            InsertStmt insertStmt = null;
+            // Lock skeleton from the original query: parsed (not analyzed) only to collect the target MV +
+            // base-table locks. Re-derive inside the lock since its analyze must hold the lock.
+            InsertStmt lockSkeleton;
             try (Timer ignored = Tracers.watchScope("MVRefreshParser")) {
-                // generate insert statement from defined query
-                insertStmt = generateInsertAst(ctx, PCellSortedSet.of(), mv.getIVMTaskDefinition());
+                lockSkeleton = generateInsertAst(ctx, PCellSortedSet.of(), mv.getIVMTaskDefinition(mv.getViewDefineSql()));
             }
 
-            PlannerMetaLocker locker = new PlannerMetaLocker(ctx, insertStmt);
+            PlannerMetaLocker locker = new PlannerMetaLocker(ctx, lockSkeleton);
             if (!locker.tryLock(Config.mv_refresh_try_lock_timeout_ms, TimeUnit.MILLISECONDS)) {
                 throw new LockTimeoutException("Failed to lock database in prepareRefreshPlan");
             }
+            InsertStmt insertStmt;
             try (ConnectContext.ScopeGuard guard = ctx.bindScope()) {
-                // analyze the insert statement
                 try (Timer ignored = Tracers.watchScope("MVRefreshAnalyzer")) {
+                    String derivedSelectSql = IvmRefreshDefinition.derive(ctx, mv);
+                    insertStmt = generateInsertAst(ctx, PCellSortedSet.of(), mv.getIVMTaskDefinition(derivedSelectSql));
                     analyzeInsertStmt(insertStmt);
-                    // build the insert plan
                     insertStmt = buildInsertPlan(insertStmt);
                     ctx.setExecutionId(UUIDUtil.toTUniqueId(ctx.getQueryId()));
                 }

fe/fe-core/src/main/java/com/starrocks/scheduler/mv/pct/MVPCTRefreshProcessor.java

@@ -50,6 +50,7 @@
 import com.starrocks.sql.StatementPlanner;
 import com.starrocks.sql.analyzer.AstToSQLBuilder;
 import com.starrocks.sql.analyzer.PlannerMetaLocker;
+import com.starrocks.sql.analyzer.mv.IvmRefreshDefinition;
 import com.starrocks.sql.ast.InsertStmt;
 import com.starrocks.sql.ast.StatementBase;
 import com.starrocks.sql.common.PCellSortedSet;
@@ -231,6 +232,14 @@ private InsertStmt prepareRefreshPlan(PCellSortedSet mvToRefreshedPartitions,
 
         MVPCTRefreshPlanBuilder planBuilder = new MVPCTRefreshPlanBuilder(db, mv, mvContext, mvPctRefreshPartitioner);
         try {
+            // An IVM MV's full rebuild must INSERT the rewritten query (hidden __ROW_ID__/__AGG_STATE
+            // columns), re-derived inside the lock. Require BOTH an IVM mode AND the __ROW_ID__ schema:
+            // mode alone misfires on a non-IVM AUTO MV, __ROW_ID__ alone on a PCT MV that merely outputs a
+            // column named __ROW_ID__. Either misfire fails this terminal PCT refresh.
+            if (mv.getCurrentRefreshMode().isIncrementalOrAuto() && mv.getRowIdStrategy() != null) {
+                insertStmt = generateInsertAst(ctx, mvToRefreshedPartitions,
+                        mv.getTaskDefinition(IvmRefreshDefinition.derive(ctx, mv)));
+            }
             // Analyze and prepare a partition & Rebuild insert statement by
             // considering to-refresh partitions of ref tables/ mv
             try (Timer ignored = Tracers.watchScope("MVRefreshAnalyzer")) {

fe/fe-core/src/main/java/com/starrocks/server/LocalMetastore.java

@@ -3275,7 +3275,9 @@ public void createMaterializedView(CreateMaterializedViewStatement stmt)
         materializedView.setViewDefineSql(stmt.getInlineViewDef());
         materializedView.setSimpleDefineSql(stmt.getSimpleViewDef());
         materializedView.setOriginalViewDefineSql(stmt.getOriginalViewDefineSql());
-        materializedView.setIvmDefineSql(stmt.getIvmViewDef());
+        if (!Strings.isNullOrEmpty(stmt.getIvmViewDef())) {
+            materializedView.setIvmDefineSql(stmt.getIvmViewDef());
+        }
         materializedView.setOriginalDBName(stmt.getOriginalDBName());
         // current refresh mode
         materializedView.setCurrentRefreshMode(stmt.getCurrentRefreshMode());

fe/fe-core/src/main/java/com/starrocks/sql/analyzer/MaterializedViewAnalyzer.java

@@ -397,7 +397,6 @@ public Void visitCreateMaterializedViewStatement(CreateMaterializedViewStatement
                     queryStatement = result.queryStatement();
                     // re-analyze again
                     Analyzer.analyze(queryStatement, context);
-                    statement.setIvmViewDef(AstToSQLBuilder.buildSimple(queryStatement));
                     statement.setQueryStatement(queryStatement);
                     // All incremental MVs are PK tables; the row-id strategy decides how __ROW_ID__ is sourced.
                     statement.setKeysType(KeysType.PRIMARY_KEYS);

fe/fe-core/src/main/java/com/starrocks/sql/analyzer/mv/IVMAnalyzer.java

@@ -159,6 +159,8 @@ private static List<Expr> normalizeGroupKeys(List<Expr> keys) {
     private final ConnectContext connectContext;
     private final CreateMaterializedViewStatement statement;
     private final QueryStatement queryStatement;
+    // null at CREATE -> deduce per keys; non-null at refresh -> use the MV's pinned version.
+    private Integer pinnedEncodeRowIdVersion;
 
     public IVMAnalyzer(ConnectContext connectContext,
                        CreateMaterializedViewStatement statement,
@@ -181,9 +183,20 @@ public static boolean isTableTypeIVMSupported(Table.TableType tableType) {
      * - If incremental refresh is supported, the result must not be none.
      */
     public Optional<IVMAnalyzeResult> rewrite(MaterializedView.RefreshMode refreshMode) {
+        return rewriteInternal(refreshMode, null, refreshMode.isIncremental());
+    }
+
+    public Optional<IVMAnalyzeResult> rewriteForRefresh(MaterializedView.RefreshMode refreshMode,
+                                                        int pinnedEncodeRowIdVersion) {
+        return rewriteInternal(refreshMode, pinnedEncodeRowIdVersion, false);
+    }
+
+    private Optional<IVMAnalyzeResult> rewriteInternal(MaterializedView.RefreshMode refreshMode,
+                                                       Integer pinnedVersion, boolean runTrial) {
         if (!refreshMode.isIncremental() && !refreshMode.isAuto()) {
             return Optional.empty();
         }
+        this.pinnedEncodeRowIdVersion = pinnedVersion;
 
         try {
             QueryRelation queryRelation = queryStatement.getQueryRelation();
@@ -195,8 +208,8 @@ public Optional<IVMAnalyzeResult> rewrite(MaterializedView.RefreshMode refreshMo
                     : RowIdStrategy.AUTO_INCREMENT;
             // Trial-rewrite catches drift the analyzer-level checks can't: e.g. a new logical
             // operator without a matching IvmDelta*Rule, or a combinator's metadata that no
-            // longer matches the BE state-union path. INCREMENTAL only.
-            if (refreshMode.isIncremental()) {
+            // longer matches the BE state-union path. CREATE only; refresh builds the real plan next.
+            if (runTrial) {
                 IvmTrialRewriter.runTrial(connectContext, statement, queryStatement);
             }
             IVMAnalyzeResult result = IVMAnalyzeResult.of(queryStatement, strategy, refreshMode);
@@ -413,7 +426,9 @@ private boolean checkAggregate(SelectRelation selectRelation) throws AnalysisExc
 
         // Build the row ID from the group keys, normalized like the refresh aggregate (see normalizeGroupKeys).
         List<Expr> rowIdKeys = normalizeGroupKeys(groupByExprs);
-        int encodeRowIdVersion = IvmOpUtils.deduceEncodeRowIdVersion(rowIdKeys);
+        int encodeRowIdVersion = (pinnedEncodeRowIdVersion != null)
+                ? IvmOpUtils.getEncodeRowIdVersionChecked(pinnedEncodeRowIdVersion)
+                : IvmOpUtils.deduceEncodeRowIdVersion(rowIdKeys);
         if (statement != null) {
             statement.setEncodeRowIdVersion(encodeRowIdVersion);
         }