[enhance](mtmv) not allow modify data of MTMV (#35870)

fe/fe-core/src/main/java/org/apache/doris/analysis/InsertOverwriteTableStmt.java

@@ -18,10 +18,14 @@
 package org.apache.doris.analysis;
 
 import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.MTMV;
+import org.apache.doris.catalog.TableIf;
+import org.apache.doris.common.DdlException;
 import org.apache.doris.common.ErrorCode;
 import org.apache.doris.common.ErrorReport;
 import org.apache.doris.common.UserException;
 import org.apache.doris.common.util.InternalDatabaseUtil;
+import org.apache.doris.mtmv.MTMVUtil;
 import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.qe.ConnectContext;
 
@@ -62,6 +66,10 @@ public String getTbl() {
         return target.getTblName().getTbl();
     }
 
+    public String getCtl() {
+        return target.getTblName().getCtl();
+    }
+
     public QueryStmt getQueryStmt() {
         return source.getQueryStmt();
     }
@@ -84,6 +92,11 @@ public boolean isAutoDetectPartition() {
     public void analyze(Analyzer analyzer) throws UserException {
         target.getTblName().analyze(analyzer);
         InternalDatabaseUtil.checkDatabase(getDb(), ConnectContext.get());
+        TableIf tableIf = Env.getCurrentEnv().getCatalogMgr().getCatalogOrAnalysisException(getCtl())
+                .getDbOrAnalysisException(getDb()).getTableOrAnalysisException(getTbl());
+        if (tableIf instanceof MTMV && !MTMVUtil.allowModifyMTMVData(ConnectContext.get())) {
+            throw new DdlException("Not allowed to perform current operation on async materialized view");
+        }
         if (!Env.getCurrentEnv().getAccessManager()
                 .checkTblPriv(ConnectContext.get(), target.getTblName().getCtl(), getDb(), getTbl(),
                         PrivPredicate.LOAD)) {

fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java

@@ -136,6 +136,7 @@
 import org.apache.doris.common.util.Util;
 import org.apache.doris.datasource.es.EsRepository;
 import org.apache.doris.event.DropPartitionEvent;
+import org.apache.doris.mtmv.MTMVUtil;
 import org.apache.doris.nereids.trees.plans.commands.info.DropMTMVInfo;
 import org.apache.doris.nereids.trees.plans.commands.info.TableNameInfo;
 import org.apache.doris.persist.AlterDatabasePropertyInfo;
@@ -3090,6 +3091,9 @@ public void truncateTable(TruncateTableStmt truncateTableStmt) throws DdlExcepti
         OlapTable olapTable = db.getOlapTableOrDdlException(dbTbl.getTbl());
 
         long rowsToTruncate = 0;
+        if (olapTable instanceof MTMV && !MTMVUtil.allowModifyMTMVData(ConnectContext.get())) {
+            throw new DdlException("Not allowed to perform current operation on async materialized view");
+        }
 
         BinlogConfig binlogConfig;
         olapTable.readLock();

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

@@ -58,6 +58,7 @@ public static ConnectContext createMTMVContext(MTMV mtmv) {
         ctx.setThreadLocalInfo();
         ctx.getSessionVariable().enableFallbackToOriginalPlanner = false;
         ctx.getSessionVariable().enableNereidsDML = true;
+        ctx.getSessionVariable().allowModifyMaterializedViewData = true;
         Optional<String> workloadGroup = mtmv.getWorkloadGroup();
         if (workloadGroup.isPresent()) {
             ctx.getSessionVariable().setWorkloadGroup(workloadGroup.get());

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

@@ -20,6 +20,7 @@
 import org.apache.doris.catalog.Database;
 import org.apache.doris.catalog.Env;
 import org.apache.doris.catalog.MTMV;
+import org.apache.doris.catalog.Table;
 import org.apache.doris.catalog.TableIf;
 import org.apache.doris.catalog.TableIf.TableType;
 import org.apache.doris.common.AnalysisException;
@@ -33,7 +34,11 @@
 import org.apache.doris.nereids.trees.expressions.literal.DateV2Literal;
 import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
 import org.apache.doris.nereids.trees.expressions.literal.VarcharLiteral;
+import org.apache.doris.qe.ConnectContext;
 
+import org.apache.commons.collections.CollectionUtils;
+
+import java.util.List;
 import java.util.Optional;
 import java.util.Set;
 
@@ -126,4 +131,24 @@ public static long getExprTimeSec(org.apache.doris.analysis.LiteralExpr expr, Op
                             expr.getStringValue(), dateFormat));
         }
     }
+
+    public static boolean allowModifyMTMVData(ConnectContext ctx) {
+        if (ctx == null) {
+            return false;
+        }
+        return ctx.getSessionVariable().isAllowModifyMaterializedViewData();
+    }
+
+    public static void checkModifyMTMVData(Database db, List<Long> tableIdList, ConnectContext ctx)
+            throws AnalysisException {
+        if (CollectionUtils.isEmpty(tableIdList)) {
+            return;
+        }
+        for (long tableId : tableIdList) {
+            Optional<Table> table = db.getTable(tableId);
+            if (table.isPresent() && table.get() instanceof MTMV && !MTMVUtil.allowModifyMTMVData(ctx)) {
+                throw new AnalysisException("Not allowed to perform current operation on async materialized view");
+            }
+        }
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java

@@ -18,6 +18,7 @@
 package org.apache.doris.nereids.trees.plans.commands.insert;
 
 import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.MTMV;
 import org.apache.doris.catalog.OlapTable;
 import org.apache.doris.catalog.TableIf;
 import org.apache.doris.common.ErrorCode;
@@ -26,6 +27,7 @@
 import org.apache.doris.common.util.InternalDatabaseUtil;
 import org.apache.doris.datasource.hive.HMSExternalTable;
 import org.apache.doris.insertoverwrite.InsertOverwriteUtil;
+import org.apache.doris.mtmv.MTMVUtil;
 import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.nereids.NereidsPlanner;
 import org.apache.doris.nereids.analyzer.UnboundHiveTableSink;
@@ -110,6 +112,9 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
             throw new AnalysisException("insert into overwrite only support OLAP and HMS table."
                     + " But current table type is " + targetTableIf.getType());
         }
+        if (targetTableIf instanceof MTMV && !MTMVUtil.allowModifyMTMVData(ctx)) {
+            throw new AnalysisException("Not allowed to perform current operation on async materialized view");
+        }
         this.logicalQuery = (LogicalPlan) InsertUtils.normalizePlan(logicalQuery, targetTableIf);
 
         LogicalPlanAdapter logicalPlanAdapter = new LogicalPlanAdapter(logicalQuery, ctx.getStatementContext());

fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java

@@ -524,6 +524,9 @@ public class SessionVariable implements Serializable, Writable {
     public static final String ENABLE_MATERIALIZED_VIEW_REWRITE
             = "enable_materialized_view_rewrite";
 
+    public static final String ALLOW_MODIFY_MATERIALIZED_VIEW_DATA
+            = "allow_modify_materialized_view_data";
+
     public static final String MATERIALIZED_VIEW_REWRITE_ENABLE_CONTAIN_EXTERNAL_TABLE
             = "materialized_view_rewrite_enable_contain_external_table";
 
@@ -1688,6 +1691,11 @@ public void setEnableLeftZigZag(boolean enableLeftZigZag) {
                     "Whether to enable materialized view rewriting based on struct info"})
     public boolean enableMaterializedViewRewrite = false;
 
+    @VariableMgr.VarAttr(name = ALLOW_MODIFY_MATERIALIZED_VIEW_DATA, needForward = true,
+            description = {"是否允许修改物化视图的数据",
+                    "Is it allowed to modify the data of the materialized view"})
+    public boolean allowModifyMaterializedViewData = false;
+
     @VariableMgr.VarAttr(name = MATERIALIZED_VIEW_REWRITE_ENABLE_CONTAIN_EXTERNAL_TABLE, needForward = true,
             description = {"基于结构信息的透明改写，是否使用包含外表的物化视图",
                     "Whether to use a materialized view that contains the foreign table "
@@ -3794,6 +3802,14 @@ public boolean isEnableMaterializedViewRewrite() {
         return enableMaterializedViewRewrite;
     }
 
+    public void setEnableMaterializedViewRewrite(boolean enableMaterializedViewRewrite) {
+        this.enableMaterializedViewRewrite = enableMaterializedViewRewrite;
+    }
+
+    public boolean isAllowModifyMaterializedViewData() {
+        return allowModifyMaterializedViewData;
+    }
+
     public boolean isMaterializedViewRewriteEnableContainExternalTable() {
         return materializedViewRewriteEnableContainExternalTable;
     }

fe/fe-core/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java

@@ -52,6 +52,7 @@
 import org.apache.doris.datasource.InternalCatalog;
 import org.apache.doris.event.DataChangeEvent;
 import org.apache.doris.metric.MetricRepo;
+import org.apache.doris.mtmv.MTMVUtil;
 import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.persist.BatchRemoveTransactionsOperationV2;
 import org.apache.doris.persist.CleanLabelOperationLog;
@@ -345,6 +346,7 @@ public long beginTransaction(List<Long> tableIdList, String label, TUniqueId req
         if (!coordinator.isFromInternal) {
             InternalDatabaseUtil.checkDatabase(db.getFullName(), ConnectContext.get());
         }
+        MTMVUtil.checkModifyMTMVData(db, tableIdList, ConnectContext.get());
         checkDatabaseDataQuota();
         Preconditions.checkNotNull(coordinator);
         Preconditions.checkNotNull(label);

regression-test/suites/mtmv_p0/test_modify_data_mtmv.groovy

@@ -0,0 +1,83 @@
+// 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.
+
+import org.junit.Assert;
+
+suite("test_modify_data_mtmv","mtmv") {
+    String suiteName = "test_modify_data_mtmv"
+    String tableName = "${suiteName}_table"
+    String mvName = "${suiteName}_mv"
+    sql """drop table if exists `${tableName}`"""
+    sql """drop materialized view if exists ${mvName};"""
+
+    sql """
+        CREATE TABLE ${tableName}
+        (
+            k1 TINYINT,
+            k2 INT not null
+        )
+        DISTRIBUTED BY HASH(k2) BUCKETS 2
+        PROPERTIES (
+            "replication_num" = "1"
+        );
+        """
+    sql """
+        CREATE MATERIALIZED VIEW ${mvName}
+        BUILD DEFERRED REFRESH AUTO ON MANUAL
+        DISTRIBUTED BY RANDOM BUCKETS 2
+        PROPERTIES (
+        'replication_num' = '1'
+        )
+        AS
+        SELECT * from ${tableName};
+        """
+
+    sql """
+        insert into ${tableName} values(1,1),(2,2),(3,3);
+        """
+    sql """
+        REFRESH MATERIALIZED VIEW ${mvName} AUTO
+        """
+    waitingMTMVTaskFinishedByMvName(mvName)
+
+    // insert into mtmv
+    test {
+        sql """insert into ${mvName} values(1,1)"""
+        exception "Not allowed"
+    }
+
+    // delete from mtmv
+    test {
+        sql """delete from ${mvName} where k2=1"""
+        exception "Not allowed"
+    }
+
+    // truncate table
+    test {
+        sql """truncate table ${mvName}"""
+        exception "Not allowed"
+    }
+
+    // insert overwrite
+    test {
+        sql """insert overwrite table ${mvName} values(2,2)"""
+        exception "Not allowed"
+    }
+
+    sql """drop table if exists `${tableName}`"""
+    sql """drop materialized view if exists ${mvName};"""
+}