[Enhancement] Optimize logic of materialized view when base table is dropped

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

@@ -406,6 +406,18 @@ public HashMap<Long, AgentBatchTask> unprotectDropTable(long tableId, boolean is
             }
         }
 
+        // process related materialized views
+        if (table.isOlapTable()) {
+            OlapTable olapTable = (OlapTable) table;
+            for (long mvId : olapTable.getRelatedMaterializedViews()) {
+                Table tmpTable = getTable(mvId);
+                if (tmpTable != null) {
+                    MaterializedView mv = (MaterializedView) tmpTable;
+                    mv.setActive(false);
+                }
+            }
+        }
+
         LOG.info("finished dropping table[{}] in db[{}], tableId: {}", table.getName(), getFullName(),
                 table.getId());
         return batchTaskMap;

fe/fe-core/src/main/java/com/starrocks/scheduler/MvTaskRunProcessor.java

@@ -42,6 +42,7 @@
 import com.starrocks.sql.analyzer.ViewDefBuilder;
 import com.starrocks.sql.ast.QueryStatement;
 import com.starrocks.sql.ast.TableRelation;
+import com.starrocks.sql.common.DmlException;
 import com.starrocks.sql.common.ExpressionPartitionUtil;
 import com.starrocks.sql.optimizer.Utils;
 import com.starrocks.sql.parser.SqlParser;
@@ -72,9 +73,10 @@ public void processTaskRun(TaskRunContext context) {
         Database database = GlobalStateMgr.getCurrentState().getDb(context.ctx.getDatabase());
         MaterializedView materializedView = (MaterializedView) database.getTable(mvId);
         if (!materializedView.isActive()) {
-            LOG.warn("Materialized view: {} is not active, " +
-                    "skip sync partition and data with base tables", mvId);
-            return;
+            String errorMsg = String.format("Materialized view: %s, id: %d is not active, " +
+                    "skip sync partition and data with base tables", materializedView.getName(), mvId);
+            LOG.warn(errorMsg);
+            throw new DmlException(errorMsg);
         }
         Set<Long> baseTableIds = materializedView.getBaseTableIds();
         PartitionInfo partitionInfo = materializedView.getPartitionInfo();

fe/fe-core/src/main/java/com/starrocks/sql/common/DmlException.java

@@ -0,0 +1,17 @@
+// This file is licensed under the Elastic License 2.0. Copyright 2021-present, StarRocks Limited.
+package com.starrocks.sql.common;
+
+public class DmlException extends RuntimeException {
+    public DmlException(String message) {
+        super(message);
+    }
+
+    @Override
+    public String getMessage() {
+        String message = super.getMessage();
+        if (message == null && getCause() != null) {
+            message = getCause().getMessage();
+        }
+        return message;
+    }
+}

fe/fe-core/src/test/java/com/starrocks/catalog/MaterializedViewTest.java

@@ -444,4 +444,38 @@ public void testRenameMaterializedView() throws Exception {
         SlotRef slotRef2 = (SlotRef) rightChild;
         Assert.assertEquals("mv_new_name2", slotRef2.getTblNameWithoutAnalyzed().getTbl());
     }
+
+    @Test
+    public void testMvAfterDropBaseTable() throws Exception {
+        FeConstants.runningUnitTest = true;
+        Config.enable_experimental_mv = true;
+        UtFrameUtils.createMinStarRocksCluster();
+        ConnectContext connectContext = UtFrameUtils.createDefaultCtx();
+        StarRocksAssert starRocksAssert = new StarRocksAssert(connectContext);
+        starRocksAssert.withDatabase("test").useDatabase("test")
+                .withTable("CREATE TABLE test.tbl_drop\n" +
+                        "(\n" +
+                        "    k1 date,\n" +
+                        "    k2 int,\n" +
+                        "    v1 int sum\n" +
+                        ")\n" +
+                        "PARTITION BY RANGE(k1)\n" +
+                        "(\n" +
+                        "    PARTITION p1 values [('2022-02-01'),('2022-02-16')),\n" +
+                        "    PARTITION p2 values [('2022-02-16'),('2022-03-01'))\n" +
+                        ")\n" +
+                        "DISTRIBUTED BY HASH(k2) BUCKETS 3\n" +
+                        "PROPERTIES('replication_num' = '1');")
+                .withNewMaterializedView("create materialized view mv_to_check\n" +
+                        "distributed by hash(k2) buckets 3\n" +
+                        "refresh async\n" +
+                        "as select k2, sum(v1) as total from tbl_drop group by k2;");
+        Database testDb = GlobalStateMgr.getCurrentState().getDb("default_cluster:test");
+        MaterializedView mv = ((MaterializedView) testDb.getTable("mv_to_check"));
+        String dropSql = "drop table tbl_drop;";
+        StmtExecutor stmtExecutor = new StmtExecutor(connectContext, dropSql);
+        stmtExecutor.execute();
+        Assert.assertNotNull(mv);
+        Assert.assertFalse(mv.isActive());
+    }
 }

fe/fe-core/src/test/java/com/starrocks/scheduler/MvTaskRunProcessorTest.java

@@ -62,6 +62,12 @@ public static void beforeClass() throws Exception {
                         "DISTRIBUTED BY HASH(k2) BUCKETS 3\n" +
                         "PROPERTIES('replication_num' = '1');")
                 .withNewMaterializedView("create materialized view test.mv1\n" +
+                        "partition by date_trunc('week',k1) \n" +
+                        "distributed by hash(k2)\n" +
+                        "refresh manual\n" +
+                        "properties('replication_num' = '1')\n" +
+                        "as select tbl1.k1, tbl2.k2 from tbl1 join tbl2 on tbl1.k2 = tbl2.k2;")
+                .withNewMaterializedView("create materialized view test.mv_inactive\n" +
                         "partition by date_trunc('week',k1) \n" +
                         "distributed by hash(k2)\n" +
                         "refresh manual\n" +
@@ -118,4 +124,21 @@ public void handleDMLStmt(ExecPlan execPlan, DmlStmt stmt) throws Exception {}
             Assert.fail(e.getMessage());
         }
     }
+
+    @Test
+    public void testInactive() {
+        Database testDb = GlobalStateMgr.getCurrentState().getDb("default_cluster:test");
+        MaterializedView materializedView = ((MaterializedView) testDb.getTable("mv_inactive"));
+        materializedView.setActive(false);
+        Task task = TaskBuilder.buildMvTask(materializedView, testDb.getFullName());
+
+        TaskRun taskRun = TaskRunBuilder.newBuilder(task).build();
+        taskRun.initStatus(UUIDUtil.genUUID().toString(), System.currentTimeMillis());
+        try {
+            taskRun.executeTaskRun();
+            Assert.fail("should not be here. executeTaskRun will throw exception");
+        } catch (Exception e) {
+            Assert.assertTrue(e.getMessage().contains("is not active, skip sync partition and data with base tables"));
+        }
+    }
 }