Only inactive related materialized views because of base table/view i…

…s changed in Leader and not replay

Signed-off-by: shuming.li <[email protected]>

fe/fe-core/src/main/java/com/starrocks/alter/AlterJobExecutor.java

@@ -54,7 +54,6 @@
 import com.starrocks.persist.SwapTableOperationLog;
 import com.starrocks.qe.ConnectContext;
 import com.starrocks.server.GlobalStateMgr;
-import com.starrocks.server.LocalMetastore;
 import com.starrocks.sql.analyzer.SemanticException;
 import com.starrocks.sql.ast.AddColumnClause;
 import com.starrocks.sql.ast.AddColumnsClause;
@@ -329,10 +328,10 @@ public Void visitSwapTableClause(SwapTableClause clause, ConnectContext context)
                 }
 
                 // inactive the related MVs
-                LocalMetastore.inactiveRelatedMaterializedView(db, origTable,
-                        MaterializedViewExceptions.inactiveReasonForBaseTableSwapped(origTblName));
-                LocalMetastore.inactiveRelatedMaterializedView(db, olapNewTbl,
-                        MaterializedViewExceptions.inactiveReasonForBaseTableSwapped(newTblName));
+                AlterMVJobExecutor.inactiveRelatedMaterializedView(db, origTable,
+                        MaterializedViewExceptions.inactiveReasonForBaseTableSwapped(origTblName), false);
+                AlterMVJobExecutor.inactiveRelatedMaterializedView(db, olapNewTbl,
+                        MaterializedViewExceptions.inactiveReasonForBaseTableSwapped(newTblName), false);
 
                 SwapTableOperationLog log = new SwapTableOperationLog(db.getId(), origTable.getId(), olapNewTbl.getId());
                 GlobalStateMgr.getCurrentState().getAlterJobMgr().swapTableInternal(log);
@@ -811,7 +810,7 @@ public Void visitAlterViewClause(AlterViewClause alterViewClause, ConnectContext
                 alterViewClause.getColumns(),
                 ctx.getSessionVariable().getSqlMode(), alterViewClause.getComment());
 
-        GlobalStateMgr.getCurrentState().getAlterJobMgr().alterView(alterViewInfo);
+        GlobalStateMgr.getCurrentState().getAlterJobMgr().alterView(alterViewInfo, false);
         GlobalStateMgr.getCurrentState().getEditLog().logModifyViewDef(alterViewInfo);
         return null;
     }

fe/fe-core/src/main/java/com/starrocks/alter/AlterJobMgr.java

@@ -79,7 +79,6 @@
 import com.starrocks.scheduler.Task;
 import com.starrocks.scheduler.TaskBuilder;
 import com.starrocks.server.GlobalStateMgr;
-import com.starrocks.server.LocalMetastore;
 import com.starrocks.sql.analyzer.Analyzer;
 import com.starrocks.sql.analyzer.MaterializedViewAnalyzer;
 import com.starrocks.sql.analyzer.SemanticException;
@@ -508,7 +507,7 @@ public void swapTableInternal(SwapTableOperationLog log) throws DdlException {
         }
     }
 
-    public void alterView(AlterViewInfo alterViewInfo) {
+    public void alterView(AlterViewInfo alterViewInfo, boolean isReplay) {
         long dbId = alterViewInfo.getDbId();
         long tableId = alterViewInfo.getTableId();
         String inlineViewDef = alterViewInfo.getInlineViewDef();
@@ -530,8 +529,8 @@ public void alterView(AlterViewInfo alterViewInfo) {
             }
             view.setNewFullSchema(newFullSchema);
             view.setComment(comment);
-            LocalMetastore.inactiveRelatedMaterializedView(db, view,
-                    MaterializedViewExceptions.inactiveReasonForBaseViewChanged(viewName));
+            AlterMVJobExecutor.inactiveRelatedMaterializedView(db, view,
+                    MaterializedViewExceptions.inactiveReasonForBaseViewChanged(viewName), isReplay);
             db.dropTable(viewName);
             db.registerTableUnlocked(view);
 

fe/fe-core/src/main/java/com/starrocks/alter/AlterMVJobExecutor.java

@@ -20,13 +20,19 @@
 import com.starrocks.analysis.IntLiteral;
 import com.starrocks.analysis.StringLiteral;
 import com.starrocks.analysis.TableName;
+import com.starrocks.catalog.Column;
 import com.starrocks.catalog.Database;
 import com.starrocks.catalog.MaterializedView;
+import com.starrocks.catalog.MvId;
+import com.starrocks.catalog.MvPlanContext;
+import com.starrocks.catalog.OlapTable;
 import com.starrocks.catalog.Table;
 import com.starrocks.catalog.TableProperty;
 import com.starrocks.catalog.constraint.ForeignKeyConstraint;
 import com.starrocks.catalog.constraint.UniqueConstraint;
+import com.starrocks.common.Config;
 import com.starrocks.common.DdlException;
+import com.starrocks.common.MaterializedViewExceptions;
 import com.starrocks.common.MetaNotFoundException;
 import com.starrocks.common.Pair;
 import com.starrocks.common.util.DynamicPartitionUtil;
@@ -58,14 +64,20 @@
 import com.starrocks.sql.ast.TableRenameClause;
 import com.starrocks.sql.common.DmlException;
 import com.starrocks.sql.optimizer.CachingMvPlanContextBuilder;
+import com.starrocks.sql.optimizer.MvPlanContextBuilder;
+import com.starrocks.sql.optimizer.OptExpression;
 import com.starrocks.sql.optimizer.Utils;
+import com.starrocks.sql.optimizer.operator.scalar.ColumnRefOperator;
 import com.starrocks.sql.optimizer.rule.transformation.materialization.MvUtils;
 import com.starrocks.warehouse.Warehouse;
 import org.apache.commons.lang3.StringUtils;
 import org.threeten.extra.PeriodDuration;
 
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.stream.Collectors;
 
 import static com.starrocks.catalog.TableProperty.INVALID;
 
@@ -475,4 +487,119 @@ private void updateTaskDefinition(MaterializedView materializedView) {
             currentTask.setPostRun(TaskBuilder.getAnalyzeMVStmt(materializedView.getName()));
         }
     }
+
+    /**
+     * Inactive related materialized views because of base table/view is changed or dropped in the leader background.
+     */
+    public static void inactiveRelatedMaterializedView(Database db, Table olapTable, String reason, boolean isReplay) {
+        if (!Config.enable_mv_automatic_inactive_by_base_table_changes) {
+            LOG.warn("Skip to inactive related materialized views because of automatic inactive is disabled, " +
+                    "table:{}, reason:{}", olapTable.getName(), reason);
+            return;
+        }
+        // Only check this in leader and not replay to avoid duplicate inactive
+        if (!GlobalStateMgr.getCurrentState().isLeader() || isReplay) {
+            LOG.warn("Skip to inactive related materialized views because of base table/view {} is " +
+                            "changed or dropped in the leader backgroud, isLeader: {}, isReplay, reason:{}",
+                    olapTable.getName(), GlobalStateMgr.getCurrentState().isLeader(), isReplay, reason);
+            return;
+        }
+        for (MvId mvId : olapTable.getRelatedMaterializedViews()) {
+            MaterializedView mv = (MaterializedView) GlobalStateMgr.getCurrentState().getLocalMetastore()
+                    .getTable(db.getId(), mvId.getId());
+            if (mv != null) {
+                LOG.warn("Inactive MV {}/{} because {}", mv.getName(), mv.getId(), reason);
+                // inactive mv by reason
+                if (mv.isActive()) {
+                    mv.setInactiveAndReason(reason);
+                    // log edit log
+                    String status = AlterMaterializedViewStatusClause.INACTIVE;
+                    GlobalStateMgr.getCurrentState().getAlterJobMgr().
+                            alterMaterializedViewStatus(mv, status, false);
+                } else {
+                    mv.setInactiveAndReason(reason);
+                }
+                // recursive inactive
+                inactiveRelatedMaterializedView(db, mv,
+                        MaterializedViewExceptions.inactiveReasonForBaseTableActive(mv.getName()), false);
+            } else {
+                LOG.info("Ignore materialized view {} does not exists", mvId);
+            }
+        }
+    }
+
+    /**
+     * Inactive related mvs after modified columns have been done. Only inactive mvs after
+     * modified columns have done because the modified process may be failed and in this situation
+     * should not inactive mvs then.
+     */
+    public static void inactiveRelatedMaterializedViews(Database db,
+                                                        OlapTable olapTable,
+                                                        Set<String> modifiedColumns) {
+        if (modifiedColumns == null || modifiedColumns.isEmpty()) {
+            return;
+        }
+        if (!Config.enable_mv_automatic_inactive_by_base_table_changes) {
+            LOG.warn("Skip to inactive related materialized views because of automatic inactive is disabled, " +
+                    "table:{}, modifiedColumns:{}", olapTable.getName(), modifiedColumns);
+            return;
+        }
+        // inactive related asynchronous mvs
+        for (MvId mvId : olapTable.getRelatedMaterializedViews()) {
+            MaterializedView mv = (MaterializedView) GlobalStateMgr.getCurrentState().getLocalMetastore()
+                    .getTable(db.getId(), mvId.getId());
+            if (mv == null) {
+                LOG.warn("Ignore materialized view {} does not exists", mvId);
+                continue;
+
+            }
+            // TODO: support more types for base table's schema change.
+            try {
+                List<MvPlanContext> mvPlanContexts = MvPlanContextBuilder.getPlanContext(mv);
+                for (MvPlanContext mvPlanContext : mvPlanContexts) {
+                    if (mvPlanContext != null) {
+                        OptExpression mvPlan = mvPlanContext.getLogicalPlan();
+                        Set<ColumnRefOperator> usedColRefs = MvUtils.collectScanColumn(mvPlan, scan -> {
+                            if (scan == null) {
+                                return false;
+                            }
+                            Table table = scan.getTable();
+                            return table.getId() == olapTable.getId();
+                        });
+                        Set<String> usedColNames = usedColRefs.stream()
+                                .map(x -> x.getName())
+                                .collect(Collectors.toCollection(() -> new TreeSet<>(String.CASE_INSENSITIVE_ORDER)));
+                        for (String modifiedColumn : modifiedColumns) {
+                            if (usedColNames.contains(modifiedColumn)) {
+                                LOG.warn("Setting the materialized view {}({}) to invalid because " +
+                                                "the column {} of the table {} was modified.", mv.getName(), mv.getId(),
+                                        modifiedColumn, olapTable.getName());
+                                mv.setInactiveAndReason(
+                                        MaterializedViewExceptions.inactiveReasonForColumnChanged(modifiedColumns));
+                            }
+                        }
+                    }
+                }
+            } catch (SemanticException e) {
+                LOG.warn("Get related materialized view {} failed:", mv.getName(), e);
+                LOG.warn("Setting the materialized view {}({}) to invalid because " +
+                                "the columns  of the table {} was modified.", mv.getName(), mv.getId(),
+                        olapTable.getName());
+                mv.setInactiveAndReason(MaterializedViewExceptions.inactiveReasonForColumnChanged(modifiedColumns));
+            } catch (Exception e) {
+                LOG.warn("Get related materialized view {} failed:", mv.getName(), e);
+                // basic check: may lose some situations
+                for (Column mvColumn : mv.getColumns()) {
+                    if (modifiedColumns.contains(mvColumn.getName())) {
+                        LOG.warn("Setting the materialized view {}({}) to invalid because " +
+                                        "the column {} of the table {} was modified.", mv.getName(), mv.getId(),
+                                mvColumn.getName(), olapTable.getName());
+                        mv.setInactiveAndReason(
+                                MaterializedViewExceptions.inactiveReasonForColumnChanged(modifiedColumns));
+                        break;
+                    }
+                }
+            }
+        }
+    }
 }

fe/fe-core/src/main/java/com/starrocks/alter/LakeTableAsyncFastSchemaChangeJob.java

@@ -45,7 +45,6 @@
 import java.util.Set;
 import java.util.stream.Collectors;
 
-import static com.starrocks.sql.optimizer.rule.transformation.materialization.MvUtils.inactiveRelatedMaterializedViews;
 import static java.util.Objects.requireNonNull;
 
 /**
@@ -141,7 +140,7 @@ private void updateCatalogUnprotected(Database db, LakeTable table) {
         table.rebuildFullSchema();
 
         // If modified columns are already done, inactive related mv
-        inactiveRelatedMaterializedViews(db, table, droppedOrModifiedColumns);
+        AlterMVJobExecutor.inactiveRelatedMaterializedViews(db, table, droppedOrModifiedColumns);
     }
 
     @Override

fe/fe-core/src/main/java/com/starrocks/alter/SchemaChangeHandler.java

@@ -148,8 +148,6 @@
 import javax.annotation.Nullable;
 import javax.validation.constraints.NotNull;
 
-import static com.starrocks.sql.optimizer.rule.transformation.materialization.MvUtils.inactiveRelatedMaterializedViews;
-
 public class SchemaChangeHandler extends AlterHandler {
 
     private static final Logger LOG = LogManager.getLogger(SchemaChangeHandler.class);
@@ -2864,7 +2862,7 @@ public void modifyTableAddOrDrop(Database db, OlapTable olapTable,
             olapTable.rebuildFullSchema();
 
             // If modified columns are already done, inactive related mv
-            inactiveRelatedMaterializedViews(db, olapTable, modifiedColumns);
+            AlterMVJobExecutor.inactiveRelatedMaterializedViews(db, olapTable, modifiedColumns);
 
             if (!isReplay) {
                 TableAddOrDropColumnsInfo info = new TableAddOrDropColumnsInfo(db.getId(), olapTable.getId(),

fe/fe-core/src/main/java/com/starrocks/alter/SchemaChangeJobV2.java

@@ -126,8 +126,6 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 
-import static com.starrocks.sql.optimizer.rule.transformation.materialization.MvUtils.inactiveRelatedMaterializedViews;
-
 /*
  * Version 2 of SchemaChangeJob.
  * This is for replacing the old SchemaChangeJob
@@ -788,7 +786,7 @@ protected void runRunningJob() throws AlterCancelException {
             onFinished(tbl);
 
             // If schema changes include fields which defined in related mv, set those mv state to inactive.
-            inactiveRelatedMaterializedViews(db, tbl, modifiedColumns);
+            AlterMVJobExecutor.inactiveRelatedMaterializedViews(db, tbl, modifiedColumns);
 
             pruneMeta();
             tbl.onReload();

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

@@ -47,6 +47,7 @@
 import com.staros.proto.FileCacheInfo;
 import com.staros.proto.FilePathInfo;
 import com.starrocks.alter.AlterJobV2Builder;
+import com.starrocks.alter.AlterMVJobExecutor;
 import com.starrocks.alter.OlapTableAlterJobV2Builder;
 import com.starrocks.alter.OlapTableRollupJobBuilder;
 import com.starrocks.alter.OptimizeJobV2Builder;
@@ -95,7 +96,6 @@
 import com.starrocks.qe.ConnectContext;
 import com.starrocks.qe.OriginStatement;
 import com.starrocks.server.GlobalStateMgr;
-import com.starrocks.server.LocalMetastore;
 import com.starrocks.server.RunMode;
 import com.starrocks.server.TemporaryTableMgr;
 import com.starrocks.sql.analyzer.AnalyzeState;
@@ -3288,8 +3288,8 @@ public void onDrop(Database db, boolean force, boolean replay) {
         // in recycle bin,
         // which make things easier.
         dropAllTempPartitions();
-        LocalMetastore.inactiveRelatedMaterializedView(db, this,
-                MaterializedViewExceptions.inactiveReasonForBaseTableNotExists(getName()));
+        AlterMVJobExecutor.inactiveRelatedMaterializedView(db, this,
+                MaterializedViewExceptions.inactiveReasonForBaseTableNotExists(getName()), replay);
         if (!replay && hasAutoIncrementColumn()) {
             sendDropAutoIncrementMapTask();
         }

fe/fe-core/src/main/java/com/starrocks/common/Config.java

@@ -3099,6 +3099,9 @@ public class Config extends ConfigBase {
      */
     @ConfField(mutable = true)
     public static boolean enable_mv_automatic_active_check = true;
+    @ConfField(mutable = true, comment = "Whether to enable the automatic related materialized views since of base table's " +
+            "schema changes")
+    public static boolean enable_mv_automatic_inactive_by_base_table_changes = true;
 
     @ConfField(mutable = true, comment = "The max retry times for base table change when refreshing materialized view")
     public static int max_mv_check_base_table_change_retry_times = 10;

fe/fe-core/src/main/java/com/starrocks/persist/EditLog.java

@@ -344,7 +344,7 @@ public void loadJournal(GlobalStateMgr globalStateMgr, JournalEntity journal)
                 }
                 case OperationType.OP_MODIFY_VIEW_DEF: {
                     AlterViewInfo info = (AlterViewInfo) journal.getData();
-                    globalStateMgr.getAlterJobMgr().alterView(info);
+                    globalStateMgr.getAlterJobMgr().alterView(info, true);
                     break;
                 }
                 case OperationType.OP_RENAME_PARTITION_V2: {

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

@@ -3312,8 +3312,8 @@ public void renameTable(Database db, Table table, TableRenameClause tableRenameC
 
         db.dropTable(oldTableName);
         db.registerTableUnlocked(olapTable);
-        inactiveRelatedMaterializedView(db, olapTable,
-                MaterializedViewExceptions.inactiveReasonForBaseTableRenamed(oldTableName));
+        AlterMVJobExecutor.inactiveRelatedMaterializedView(db, olapTable,
+                MaterializedViewExceptions.inactiveReasonForBaseTableRenamed(oldTableName), false);
 
         TableInfo tableInfo = TableInfo.createForTableRename(db.getId(), olapTable.getId(), newTableName);
         GlobalStateMgr.getCurrentState().getEditLog().logTableRename(tableInfo);
@@ -3329,23 +3329,6 @@ public void alterTableComment(Database db, Table table, AlterTableCommentClause
         table.setComment(clause.getNewComment());
     }
 
-    public static void inactiveRelatedMaterializedView(Database db, Table olapTable, String reason) {
-        for (MvId mvId : olapTable.getRelatedMaterializedViews()) {
-            MaterializedView mv = (MaterializedView) GlobalStateMgr.getCurrentState().getLocalMetastore()
-                    .getTable(db.getId(), mvId.getId());
-            if (mv != null) {
-                LOG.warn("Inactive MV {}/{} because {}", mv.getName(), mv.getId(), reason);
-                mv.setInactiveAndReason(reason);
-
-                // recursive inactive
-                inactiveRelatedMaterializedView(db, mv,
-                        MaterializedViewExceptions.inactiveReasonForBaseTableActive(mv.getName()));
-            } else {
-                LOG.info("Ignore materialized view {} does not exists", mvId);
-            }
-        }
-    }
-
     public void replayRenameTable(TableInfo tableInfo) {
         long dbId = tableInfo.getDbId();
         long tableId = tableInfo.getTableId();
@@ -3360,9 +3343,6 @@ public void replayRenameTable(TableInfo tableInfo) {
             db.dropTable(tableName);
             table.setName(newTableName);
             db.registerTableUnlocked(table);
-            inactiveRelatedMaterializedView(db, table,
-                    MaterializedViewExceptions.inactiveReasonForBaseTableRenamed(tableName));
-
             LOG.info("replay rename table[{}] to {}, tableId: {}", tableName, newTableName, table.getId());
         } finally {
             locker.unLockDatabase(db.getId(), LockType.WRITE);