[Enhancement] (nereids)implement adminSetFrontendConfigCommand in nereids

fe/fe-core/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4

@@ -597,6 +597,8 @@ supportedAdminStatement
     | ADMIN COMPACT TABLE baseTableRef (WHERE TYPE EQ STRING_LITERAL)?              #adminCompactTable
     | ADMIN CHECK tabletList properties=propertyClause?                             #adminCheckTablets
     | ADMIN SHOW TABLET STORAGE FORMAT VERBOSE?                                     #adminShowTabletStorageFormat
+    | ADMIN SET (FRONTEND | (ALL FRONTENDS)) CONFIG
+        (LEFT_PAREN propertyItemList RIGHT_PAREN)? ALL?                             #adminSetFrontendConfig
     | ADMIN CLEAN TRASH
         (ON LEFT_PAREN backends+=STRING_LITERAL
               (COMMA backends+=STRING_LITERAL)* RIGHT_PAREN)?                       #adminCleanTrash
@@ -617,8 +619,6 @@ supportedRecoverStatement
 
 unsupportedAdminStatement
     : ADMIN SET REPLICA VERSION PROPERTIES LEFT_PAREN propertyItemList RIGHT_PAREN  #adminSetReplicaVersion
-    | ADMIN SET (FRONTEND | (ALL FRONTENDS)) CONFIG
-        (LEFT_PAREN propertyItemList RIGHT_PAREN)? ALL?                             #adminSetFrontendConfig
     | ADMIN SET TABLE name=multipartIdentifier
         PARTITION VERSION properties=propertyClause?                                #adminSetPartitionVersion
     ;

fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java

@@ -203,6 +203,7 @@
 import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.nereids.jobs.load.LabelProcessor;
 import org.apache.doris.nereids.stats.HboPlanStatisticsManager;
+import org.apache.doris.nereids.trees.plans.commands.AdminSetFrontendConfigCommand;
 import org.apache.doris.nereids.trees.plans.commands.AdminSetReplicaStatusCommand;
 import org.apache.doris.nereids.trees.plans.commands.AlterSystemCommand;
 import org.apache.doris.nereids.trees.plans.commands.AlterTableCommand;
@@ -6263,6 +6264,36 @@ public void setConfig(AdminSetConfigStmt stmt) throws Exception {
         }
     }
 
+    public void setConfig(AdminSetFrontendConfigCommand command) throws Exception {
+        Map<String, String> configs = command.getConfigs();
+        Preconditions.checkState(configs.size() == 1);
+
+        for (Map.Entry<String, String> entry : configs.entrySet()) {
+            try {
+                setMutableConfigWithCallback(entry.getKey(), entry.getValue());
+            } catch (ConfigException e) {
+                throw new DdlException(e.getMessage());
+            }
+        }
+
+        if (command.isApplyToAll()) {
+            for (Frontend fe : Env.getCurrentEnv().getFrontends(null /* all */)) {
+                if (!fe.isAlive() || fe.getHost().equals(Env.getCurrentEnv().getSelfNode().getHost())) {
+                    continue;
+                }
+
+                TNetworkAddress feAddr = new TNetworkAddress(fe.getHost(), fe.getRpcPort());
+                FEOpExecutor executor = new FEOpExecutor(feAddr, command.getLocalSetStmt(),
+                        ConnectContext.get(), false);
+                executor.execute();
+                if (executor.getStatusCode() != TStatusCode.OK.getValue()) {
+                    throw new DdlException(String.format("failed to apply to fe %s:%s, error message: %s",
+                        fe.getHost(), fe.getRpcPort(), executor.getErrMsg()));
+                }
+            }
+        }
+    }
+
     public void replayBackendReplicasInfo(BackendReplicasInfo backendReplicasInfo) {
         long backendId = backendReplicasInfo.getBackendId();
         List<BackendReplicasInfo.ReplicaReportInfo> replicaInfos = backendReplicasInfo.getReplicaReportInfos();

fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java

@@ -564,6 +564,7 @@
 import org.apache.doris.nereids.trees.plans.commands.AdminCopyTabletCommand;
 import org.apache.doris.nereids.trees.plans.commands.AdminRebalanceDiskCommand;
 import org.apache.doris.nereids.trees.plans.commands.AdminRepairTableCommand;
+import org.apache.doris.nereids.trees.plans.commands.AdminSetFrontendConfigCommand;
 import org.apache.doris.nereids.trees.plans.commands.AdminSetReplicaStatusCommand;
 import org.apache.doris.nereids.trees.plans.commands.AdminSetTableStatusCommand;
 import org.apache.doris.nereids.trees.plans.commands.AlterCatalogCommentCommand;
@@ -5804,6 +5805,17 @@ public LogicalPlan visitAdminSetTableStatus(AdminSetTableStatusContext ctx) {
         return new AdminSetTableStatusCommand(new TableNameInfo(dbTblNameParts), properties);
     }
 
+    @Override
+    public LogicalPlan visitAdminSetFrontendConfig(DorisParser.AdminSetFrontendConfigContext ctx) {
+        Map<String, String> configs = visitPropertyItemList(ctx.propertyItemList());
+        boolean applyToAll = ctx.ALL() != null || ctx.FRONTENDS() != null;
+
+        return new AdminSetFrontendConfigCommand(
+            NodeType.FRONTEND,
+            configs,
+            applyToAll);
+    }
+
     @Override
     public LogicalPlan visitShowFrontends(ShowFrontendsContext ctx) {
         String detail = (ctx.name != null) ? ctx.name.getText() : null;

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/PlanType.java

@@ -325,6 +325,7 @@ public enum PlanType {
     REPLAY_COMMAND,
     ADMIN_REBALANCE_DISK_COMMAND,
     ADMIN_CANCEL_REBALANCE_DISK_COMMAND,
+    ADMIN_SET_FRONTEND_CONFIG_COMMAND,
     CREATE_ENCRYPTKEY_COMMAND,
     CREATE_WORKLOAD_GROUP_COMMAND,
     CREATE_CATALOG_COMMAND,

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

@@ -0,0 +1,126 @@
+// 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.
+
+package org.apache.doris.nereids.trees.plans.commands;
+
+import org.apache.doris.analysis.RedirectStatus;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.ConfigBase;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.common.UserException;
+import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.OriginStatement;
+import org.apache.doris.qe.StmtExecutor;
+import org.apache.doris.system.NodeType;
+
+import com.google.common.collect.Maps;
+
+import java.util.Map;
+
+/**
+ * admin set frontend config ("key" = "value");
+ */
+public class AdminSetFrontendConfigCommand extends Command implements NoForward {
+    private boolean applyToAll;
+    private NodeType type;
+    private Map<String, String> configs;
+    private OriginStatement originStmt;
+
+    private RedirectStatus redirectStatus = RedirectStatus.NO_FORWARD;
+
+    /**
+     * AdminSetFrontendConfigCommand
+     */
+    public AdminSetFrontendConfigCommand(NodeType type, Map<String, String> configs, boolean applyToAll) {
+        super(PlanType.ADMIN_SET_FRONTEND_CONFIG_COMMAND);
+        this.type = type;
+        this.configs = configs;
+        if (this.configs == null) {
+            this.configs = Maps.newHashMap();
+        }
+        this.applyToAll = applyToAll;
+
+        // we have to analyze configs here to determine whether to forward it to master
+        for (String key : this.configs.keySet()) {
+            if (ConfigBase.checkIsMasterOnly(key)) {
+                redirectStatus = RedirectStatus.FORWARD_NO_SYNC;
+                this.applyToAll = false;
+            }
+        }
+    }
+
+    @Override
+    public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
+        validate();
+        originStmt = ctx.getStatementContext().getOriginStatement();
+        Env.getCurrentEnv().setConfig(this);
+    }
+
+    /**
+     * validate
+     */
+    public void validate() throws UserException {
+        if (configs.size() != 1) {
+            throw new AnalysisException("config parameter size is not equal to 1");
+        }
+        // check auth
+        if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ConnectContext.get(), PrivPredicate.ADMIN)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR, "ADMIN");
+        }
+
+        if (type != NodeType.FRONTEND) {
+            throw new AnalysisException("Only support setting Frontend configs now");
+        }
+    }
+
+    @Override
+    public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+        return visitor.visitAdminSetFrontendConfigCommand(this, context);
+    }
+
+    public boolean isApplyToAll() {
+        return applyToAll;
+    }
+
+    public NodeType getNodeTypeType() {
+        return type;
+    }
+
+    public Map<String, String> getConfigs() {
+        return configs;
+    }
+
+    public RedirectStatus getRedirectStatus() {
+        return redirectStatus;
+    }
+
+    /**
+     * getLocalSetStmt
+     */
+    public OriginStatement getLocalSetStmt() {
+        Object[] keyArr = configs.keySet().toArray();
+        String sql = String.format("ADMIN SET FRONTEND CONFIG (\"%s\" = \"%s\");",
+                keyArr[0].toString(), configs.get(keyArr[0].toString()));
+
+        return new OriginStatement(sql, originStmt.idx);
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/visitor/CommandVisitor.java

@@ -26,6 +26,7 @@
 import org.apache.doris.nereids.trees.plans.commands.AdminCopyTabletCommand;
 import org.apache.doris.nereids.trees.plans.commands.AdminRebalanceDiskCommand;
 import org.apache.doris.nereids.trees.plans.commands.AdminRepairTableCommand;
+import org.apache.doris.nereids.trees.plans.commands.AdminSetFrontendConfigCommand;
 import org.apache.doris.nereids.trees.plans.commands.AdminSetReplicaStatusCommand;
 import org.apache.doris.nereids.trees.plans.commands.AdminSetTableStatusCommand;
 import org.apache.doris.nereids.trees.plans.commands.AlterCatalogCommentCommand;
@@ -1240,6 +1241,10 @@ default R visitGrantResourcePrivilegeCommand(GrantResourcePrivilegeCommand comma
         return visitCommand(command, context);
     }
 
+    default R visitAdminSetFrontendConfigCommand(AdminSetFrontendConfigCommand command, C context) {
+        return visitCommand(command, context);
+    }
+
     default R visitRevokeRoleCommand(RevokeRoleCommand revokeRoleCommand, C context) {
         return visitCommand(revokeRoleCommand, context);
     }

fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/AdminSetFrontendConfigCommandTest.java

@@ -0,0 +1,99 @@
+// 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.
+
+package org.apache.doris.nereids.trees.plans.commands;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.CaseSensibility;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.ConfigBase;
+import org.apache.doris.common.PatternMatcher;
+import org.apache.doris.common.PatternMatcherWrapper;
+import org.apache.doris.common.VariableAnnotation;
+import org.apache.doris.nereids.parser.NereidsParser;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.utframe.TestWithFeService;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+public class AdminSetFrontendConfigCommandTest extends TestWithFeService {
+    @Test
+    public void testNormal() throws Exception {
+        String sql = "admin set frontend config(\"alter_table_timeout_second\" = \"60\");";
+        LogicalPlan plan = new NereidsParser().parseSingle(sql);
+        Assertions.assertTrue(plan instanceof AdminSetFrontendConfigCommand);
+        Assertions.assertDoesNotThrow(() -> ((AdminSetFrontendConfigCommand) plan).validate());
+    }
+
+    @Test
+    public void testEmptyConfig() {
+        String sql = "admin set frontend config;";
+        LogicalPlan plan = new NereidsParser().parseSingle(sql);
+        Assertions.assertTrue(plan instanceof AdminSetFrontendConfigCommand);
+        AnalysisException exception = Assertions.assertThrows(AnalysisException.class,
+                () -> ((AdminSetFrontendConfigCommand) plan).validate());
+        Assertions.assertEquals("errCode = 2, detailMessage = config parameter size is not equal to 1",
+                exception.getMessage());
+    }
+
+    @Test
+    public void testExperimentalConfig() throws Exception {
+        // 1. set without experimental
+        boolean enableMtmv = Config.enable_mtmv;
+        String sql = "admin set frontend config('enable_mtmv' = '" + String.valueOf(!enableMtmv) + "');";
+        LogicalPlan plan = new NereidsParser().parseSingle(sql);
+
+        Assertions.assertTrue(plan instanceof AdminSetFrontendConfigCommand);
+        Env.getCurrentEnv().setConfig((AdminSetFrontendConfigCommand) plan);
+        Assertions.assertNotEquals(enableMtmv, Config.enable_mtmv);
+
+        // 2. set with experimental
+        enableMtmv = Config.enable_mtmv;
+        sql = "admin set frontend config('experimental_enable_mtmv' = '" + String.valueOf(!enableMtmv) + "');";
+        plan = new NereidsParser().parseSingle(sql);
+
+        Assertions.assertTrue(plan instanceof AdminSetFrontendConfigCommand);
+        Env.getCurrentEnv().setConfig((AdminSetFrontendConfigCommand) plan);
+        Assertions.assertNotEquals(enableMtmv, Config.enable_mtmv);
+
+        // 3. show config
+        int num = ConfigBase.getConfigNumByVariableAnnotation(VariableAnnotation.EXPERIMENTAL);
+        PatternMatcher matcher = PatternMatcherWrapper.createMysqlPattern("%experimental%",
+                CaseSensibility.CONFIG.getCaseSensibility());
+        List<List<String>> results = ConfigBase.getConfigInfo(matcher);
+        Assertions.assertEquals(num, results.size());
+
+        num = ConfigBase.getConfigNumByVariableAnnotation(VariableAnnotation.DEPRECATED);
+        matcher = PatternMatcherWrapper.createMysqlPattern("%deprecated%",
+            CaseSensibility.CONFIG.getCaseSensibility());
+        results = ConfigBase.getConfigInfo(matcher);
+        Assertions.assertEquals(num, results.size());
+    }
+
+    @Test
+    public void testTrimPropertyKey() throws Exception {
+        String sql = "admin set frontend config(\" alter_table_timeout_second \" = \"60\");";
+        LogicalPlan plan = new NereidsParser().parseSingle(sql);
+
+        Assertions.assertTrue(plan instanceof AdminSetFrontendConfigCommand);
+        Assertions.assertEquals("60", ((AdminSetFrontendConfigCommand) plan).getConfigs().get("alter_table_timeout_second"));
+    }
+}