[Config] Add alter table, broker load, stream load, truncate table disable co…

fe/fe-core/src/main/java/org/apache/doris/common/Config.java

@@ -1387,4 +1387,12 @@ public class Config extends ConfigBase {
      */
     @ConfField(mutable = true, masterOnly = true)
     public static int max_dynamic_partition_num = 500;
+
+    /**
+     * Array config to disable operations
+     * You can set config use command: admin set frontend config ("disable_config"="broker load, alter table, truncate table, stream load");
+     * It can be extended to add more disable operations, currently supports 4 operations.
+     */
+    @ConfField(mutable = true)
+    public static String[] disable_config = {};
 }

fe/fe-core/src/main/java/org/apache/doris/http/rest/LoadAction.java

@@ -17,8 +17,12 @@
 
 package org.apache.doris.http.rest;
 
+import com.google.common.base.Strings;
+import io.netty.handler.codec.http.HttpHeaders;
+import io.netty.handler.codec.http.HttpMethod;
 import org.apache.doris.catalog.Catalog;
 import org.apache.doris.cluster.ClusterNamespace;
+import org.apache.doris.common.Config;
 import org.apache.doris.common.DdlException;
 import org.apache.doris.http.ActionController;
 import org.apache.doris.http.BaseRequest;
@@ -29,17 +33,11 @@
 import org.apache.doris.service.ExecuteEnv;
 import org.apache.doris.system.Backend;
 import org.apache.doris.thrift.TNetworkAddress;
-
-import com.google.common.base.Strings;
-
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
 import java.util.List;
 
-import io.netty.handler.codec.http.HttpHeaders;
-import io.netty.handler.codec.http.HttpMethod;
-
 public class LoadAction extends RestBaseAction {
     private static final Logger LOG = LogManager.getLogger(LoadAction.class);
 
@@ -91,6 +89,14 @@ public void executeWithoutPassword(BaseRequest request, BaseResponse response) t
         if (Strings.isNullOrEmpty(tableName)) {
             throw new DdlException("No table selected.");
         }
+
+        if (isStreamLoad) {
+            for (String config: Config.disable_config) {
+                if (config.equalsIgnoreCase("stream load")) {
+                    throw new DdlException("Stream Load is disabled, please contact Doris Manager.");
+                }
+            }
+        }
 
         String fullDbName = ClusterNamespace.getFullName(clusterName, dbName);
 

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

@@ -113,6 +113,12 @@ public static void execute(Catalog catalog, DdlStmt ddlStmt) throws Exception {
         } else if (ddlStmt instanceof DropMaterializedViewStmt) {
             catalog.dropMaterializedView((DropMaterializedViewStmt) ddlStmt);
         } else if (ddlStmt instanceof AlterTableStmt) {
+            for (String config: Config.disable_config) {
+                if (config.equalsIgnoreCase("alter table")) {
+                    throw new DdlException("Alter Table operation is disabled,"
+                            + " please contact Doris manager;'");
+                }
+            }
             catalog.alterTable((AlterTableStmt) ddlStmt);
         } else if (ddlStmt instanceof AlterViewStmt) {
             catalog.alterView((AlterViewStmt) ddlStmt);
@@ -128,6 +134,14 @@ public static void execute(Catalog catalog, DdlStmt ddlStmt) throws Exception {
                 throw new DdlException("Load job by hadoop cluster is disabled."
                         + " Try using broker load. See 'help broker load;'");
             }
+            if (jobType == EtlJobType.BROKER) {
+                for (String config:Config.disable_config) {
+                    if (config.equalsIgnoreCase("broker load")) {
+                        throw new DdlException("Broker Load is disabled,"
+                                + " please contact Doris manager;'");
+                    }
+                }
+            }
             if (jobType == EtlJobType.HADOOP) {
                 catalog.getLoadManager().createLoadJobV1FromStmt(loadStmt, jobType, System.currentTimeMillis());
             } else {
@@ -207,6 +221,12 @@ public static void execute(Catalog catalog, DdlStmt ddlStmt) throws Exception {
         } else if (ddlStmt instanceof SyncStmt) {
             return;
         } else if (ddlStmt instanceof TruncateTableStmt) {
+            for (String config:Config.disable_config) {
+                if (config.equalsIgnoreCase("truncate table")) {
+                    throw new DdlException("Truncate Table is disabled,"
+                            + " please contact Doris manager;'");
+                }
+            }
             catalog.truncateTable((TruncateTableStmt) ddlStmt);
         } else if (ddlStmt instanceof AdminRepairTableStmt) {
             catalog.getTabletChecker().repairTable((AdminRepairTableStmt) ddlStmt);