From 202cdb2744d486cf583868c6bf122d98146b6ae0 Mon Sep 17 00:00:00 2001
From: zhangdong <493738387@qq.com>
Date: Wed, 8 May 2024 16:03:22 +0800
Subject: [PATCH] [fix](mtmv)fix refresh failed when not use db before create
 MTMV (#34431) (#34522)

when create MTMV,we will save current ctl and db.

when refresh MTMV,will create an ConnectContext, and set same ctl, db to ctx

when db,ctx dropped, task will be failed.

But sometimes deleting a db does not actually have an impact, so changing it to not directly fail. If refreshing the data does cause an error, then giving the user an error message
---
 .../doris/job/extensions/mtmv/MTMVTask.java   | 14 ++-
 .../org/apache/doris/mtmv/MTMVPlanUtil.java   | 43 +++++++--
 .../trees/plans/visitor/TableCollector.java   | 14 +--
 .../cluster/DecommissionBackendTest.java      |  4 +-
 .../mtmv_p0/test_env_db_dropped_mtmv.groovy   | 93 +++++++++++++++++++
 5 files changed, 144 insertions(+), 24 deletions(-)
 create mode 100644 regression-test/suites/mtmv_p0/test_env_db_dropped_mtmv.groovy

diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
index 9e872cf034..8924b229b2 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
@@ -25,6 +25,7 @@ import org.apache.doris.catalog.TableIf;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.DdlException;
 import org.apache.doris.common.FeConstants;
+import org.apache.doris.common.Pair;
 import org.apache.doris.common.UserException;
 import org.apache.doris.common.util.DebugUtil;
 import org.apache.doris.common.util.TimeUtils;
@@ -154,8 +155,8 @@ public class MTMVTask extends AbstractTask {
     @Override
     public void run() throws JobException {
         LOG.info("mtmv task run, taskId: {}", super.getTaskId());
+        ConnectContext ctx = MTMVPlanUtil.createMTMVContext(mtmv);
         try {
-            ConnectContext ctx = MTMVPlanUtil.createMTMVContext(mtmv);
             if (LOG.isDebugEnabled()) {
                 String taskSessionContext = ctx.getSessionVariable().toJson().toJSONString();
                 if (LOG.isDebugEnabled()) {
@@ -201,8 +202,15 @@ public class MTMVTask extends AbstractTask {
             }
         } catch (Throwable e) {
             if (getStatus() == TaskStatus.RUNNING) {
-                LOG.warn("run task failed: ", e);
-                throw new JobException(e);
+                StringBuilder errMsg = new StringBuilder();
+                // when env ctl/db not exist, need give client tips
+                Pair<Boolean, String> pair = MTMVPlanUtil.checkEnvInfo(mtmv.getEnvInfo(), ctx);
+                if (!pair.first) {
+                    errMsg.append(pair.second);
+                }
+                errMsg.append(e.getMessage());
+                LOG.warn("run task failed: ", errMsg.toString());
+                throw new JobException(errMsg.toString(), e);
             } else {
                 // if status is not `RUNNING`,maybe the task was canceled, therefore, it is a normal situation
                 LOG.info("task [{}] interruption running, because status is [{}]", getTaskId(), getStatus());
diff --git a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPlanUtil.java b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPlanUtil.java
index d23c7534f8..16eab00dca 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPlanUtil.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPlanUtil.java
@@ -19,11 +19,12 @@ package org.apache.doris.mtmv;
 
 import org.apache.doris.analysis.StatementBase;
 import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.DatabaseIf;
 import org.apache.doris.catalog.Env;
 import org.apache.doris.catalog.MTMV;
 import org.apache.doris.catalog.TableIf;
 import org.apache.doris.catalog.TableIf.TableType;
-import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.Pair;
 import org.apache.doris.datasource.CatalogIf;
 import org.apache.doris.mysql.privilege.Auth;
 import org.apache.doris.nereids.NereidsPlanner;
@@ -48,26 +49,54 @@ import java.util.Set;
 
 public class MTMVPlanUtil {
 
-    public static ConnectContext createMTMVContext(MTMV mtmv) throws AnalysisException {
+    public static ConnectContext createMTMVContext(MTMV mtmv) {
         ConnectContext ctx = new ConnectContext();
         ctx.setEnv(Env.getCurrentEnv());
         ctx.setQualifiedUser(Auth.ADMIN_USER);
         ctx.setCurrentUserIdentity(UserIdentity.ADMIN);
         ctx.getState().reset();
         ctx.setThreadLocalInfo();
-        CatalogIf catalog = Env.getCurrentEnv().getCatalogMgr()
-                .getCatalogOrAnalysisException(mtmv.getEnvInfo().getCtlId());
-        ctx.changeDefaultCatalog(catalog.getName());
-        ctx.setDatabase(catalog.getDbOrAnalysisException(mtmv.getEnvInfo().getDbId()).getFullName());
         ctx.getSessionVariable().enableFallbackToOriginalPlanner = false;
+        ctx.getSessionVariable().enableNereidsDML = true;
         Optional<String> workloadGroup = mtmv.getWorkloadGroup();
         if (workloadGroup.isPresent()) {
             ctx.getSessionVariable().setWorkloadGroup(workloadGroup.get());
         }
-        ctx.getSessionVariable().enableNereidsDML = true;
+        // switch catalog;
+        CatalogIf catalog = Env.getCurrentEnv().getCatalogMgr().getCatalog(mtmv.getEnvInfo().getCtlId());
+        // if catalog not exist, it may not have any impact, so there is no error and it will be returned directly
+        if (catalog == null) {
+            return ctx;
+        }
+        ctx.changeDefaultCatalog(catalog.getName());
+        // use db
+        Optional<? extends DatabaseIf<? extends TableIf>> databaseIf = catalog.getDb(mtmv.getEnvInfo().getDbId());
+        // if db not exist, it may not have any impact, so there is no error and it will be returned directly
+        if (!databaseIf.isPresent()) {
+            return ctx;
+        }
+        ctx.setDatabase(databaseIf.get().getFullName());
         return ctx;
     }
 
+    public static Pair<Boolean, String> checkEnvInfo(EnvInfo envInfo, ConnectContext ctx) {
+        if (envInfo.getCtlId() != ctx.getCurrentCatalog().getId()) {
+            return Pair.of(false, String.format(
+                    "The catalog selected when creating the materialized view was %s, "
+                            + "but now this catalog has been deleted. "
+                            + "Please recreate the materialized view.",
+                    envInfo.getCtlId()));
+        }
+        if (envInfo.getDbId() != ctx.getCurrentDbId()) {
+            return Pair.of(false, String.format(
+                    "The database selected when creating the materialized view was %s, "
+                            + "but now this database has been deleted. "
+                            + "Please recreate the materialized view.",
+                    envInfo.getDbId()));
+        }
+        return Pair.of(true, "");
+    }
+
     public static MTMVRelation generateMTMVRelation(MTMV mtmv, ConnectContext ctx) {
         // Should not make table without data to empty relation when analyze the related table,
         // so add disable rules
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/visitor/TableCollector.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/visitor/TableCollector.java
index 3736ab60b0..94b1eb89c2 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/visitor/TableCollector.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/visitor/TableCollector.java
@@ -20,7 +20,6 @@ package org.apache.doris.nereids.trees.plans.visitor;
 import org.apache.doris.catalog.MTMV;
 import org.apache.doris.catalog.TableIf;
 import org.apache.doris.catalog.TableIf.TableType;
-import org.apache.doris.common.AnalysisException;
 import org.apache.doris.mtmv.MTMVCache;
 import org.apache.doris.mtmv.MTMVPlanUtil;
 import org.apache.doris.nereids.trees.plans.Plan;
@@ -73,17 +72,8 @@ public class TableCollector extends DefaultPlanVisitor<Plan, TableCollectorConte
         if (!context.isExpand()) {
             return;
         }
-        try {
-            MTMVCache expandedMv = MTMVCache.from(mtmv, MTMVPlanUtil.createMTMVContext(mtmv));
-            expandedMv.getLogicalPlan().accept(this, context);
-        } catch (AnalysisException e) {
-            LOG.error(String.format(
-                    "table collector expand fail, mtmv name is %s, targetTableTypes is %s",
-                    mtmv.getName(), context.targetTableTypes), e);
-            throw new org.apache.doris.nereids.exceptions.AnalysisException(
-                    String.format("expand mv and collect table fail, mv name is %s, mv sql is %s",
-                            mtmv.getName(), mtmv.getQuerySql()), e);
-        }
+        MTMVCache expandedMv = MTMVCache.from(mtmv, MTMVPlanUtil.createMTMVContext(mtmv));
+        expandedMv.getLogicalPlan().accept(this, context);
     }
 
     /**
diff --git a/fe/fe-core/src/test/java/org/apache/doris/cluster/DecommissionBackendTest.java b/fe/fe-core/src/test/java/org/apache/doris/cluster/DecommissionBackendTest.java
index e50a7ba632..8006cc7a57 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/cluster/DecommissionBackendTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/cluster/DecommissionBackendTest.java
@@ -283,14 +283,14 @@ public class DecommissionBackendTest extends TestWithFeService {
                 + "DISTRIBUTED BY HASH(`c2`) BUCKETS 1\n"
                 + ";");
 
-        createMvByNereids("create materialized view db4.mv1 BUILD IMMEDIATE REFRESH COMPLETE ON MANUAL\n"
+        createMvByNereids("create materialized view db4.mv1 BUILD DEFERRED REFRESH COMPLETE ON MANUAL\n"
                 + "DISTRIBUTED BY RANDOM BUCKETS 1\n"
                 + "as "
                 + "select t1.c1, t2.c2, t2.k4 "
                 + "from db4.table1 t1 "
                 + "inner join db4.table2 t2 on t1.c1= t2.c2;");
 
-        createMvByNereids("create materialized view db4.mv2 BUILD IMMEDIATE REFRESH COMPLETE ON MANUAL\n"
+        createMvByNereids("create materialized view db4.mv2 BUILD DEFERRED REFRESH COMPLETE ON MANUAL\n"
                 + "DISTRIBUTED BY HASH(c1) BUCKETS 20 \n"
                 + "PROPERTIES ( 'colocate_with' = 'foo', 'replication_num' = '3' ) "
                 + "as "
diff --git a/regression-test/suites/mtmv_p0/test_env_db_dropped_mtmv.groovy b/regression-test/suites/mtmv_p0/test_env_db_dropped_mtmv.groovy
new file mode 100644
index 0000000000..11063ba0e5
--- /dev/null
+++ b/regression-test/suites/mtmv_p0/test_env_db_dropped_mtmv.groovy
@@ -0,0 +1,93 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_env_db_dropped_mtmv") {
+    def tableName = "t_test_env_db_dropped_mtmv_user"
+    def mvName = "test_env_db_dropped_mtmv"
+    def dbName1 = "test_env_db_dropped_mtmv_db1"
+    def dbName2 = "test_env_db_dropped_mtmv_db2"
+    sql """drop database if exists `${dbName1}`"""
+    sql """drop database if exists `${dbName2}`"""
+    sql """create database `${dbName1}`"""
+    sql """create database `${dbName2}`"""
+    sql """use `${dbName1}`"""
+
+    sql """
+        CREATE TABLE IF NOT EXISTS ${dbName2}.${tableName} (
+            event_day DATE,
+            id BIGINT,
+            username VARCHAR(20)
+        )
+        DISTRIBUTED BY HASH(id) BUCKETS 10 
+        PROPERTIES (
+            "replication_num" = "1"
+        );
+        """
+
+    sql """
+        CREATE MATERIALIZED VIEW ${dbName2}.${mvName}
+        BUILD DEFERRED REFRESH COMPLETE ON MANUAL
+        DISTRIBUTED BY RANDOM BUCKETS 2
+        PROPERTIES ('replication_num' = '1') 
+        AS 
+        SELECT * FROM ${dbName2}.${tableName};
+    """
+    def jobName = getJobName(dbName2, mvName);
+
+    // table and mv in db2, so drop db1, should success
+    sql """drop database if exists `${dbName1}`"""
+    sql """
+            REFRESH MATERIALIZED VIEW ${dbName2}.${mvName} AUTO;
+        """
+    waitingMTMVTaskFinished(jobName)
+
+    sql """
+        DROP MATERIALIZED VIEW ${dbName2}.${mvName};
+    """
+
+    sql """create database `${dbName1}`"""
+    sql """use `${dbName1}`"""
+    sql """
+        CREATE TABLE IF NOT EXISTS ${tableName} (
+            event_day DATE,
+            id BIGINT,
+            username VARCHAR(20)
+        )
+        DISTRIBUTED BY HASH(id) BUCKETS 10
+        PROPERTIES (
+            "replication_num" = "1"
+        );
+        """
+    sql """
+        CREATE MATERIALIZED VIEW ${dbName2}.${mvName}
+        BUILD DEFERRED REFRESH COMPLETE ON MANUAL
+        DISTRIBUTED BY RANDOM BUCKETS 2
+        PROPERTIES ('replication_num' = '1')
+        AS
+        SELECT * FROM ${tableName};
+    """
+    jobName = getJobName(dbName2, mvName);
+    sql """drop database if exists `${dbName1}`"""
+    // table in db1, has been dropped, will error
+    sql """
+            REFRESH MATERIALIZED VIEW ${dbName2}.${mvName} AUTO;
+        """
+    waitingMTMVTaskFinishedNotNeedSuccess(jobName)
+    def msg = sql """select ErrorMsg from tasks('type'='mv') where JobName = '${jobName}' order by CreateTime DESC limit 1"""
+    logger.info(msg.toString())
+    assertTrue(msg.toString().contains("has been deleted"))
+}