[feature](fe) Support dropping row policies by roles in batch

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

@@ -366,6 +366,8 @@ supportedDropStatement
     | DROP INDEX (IF EXISTS)? name=identifier ON tableName=multipartIdentifier
         partitionSpec?                                                            #dropIndex
     | DROP RESOURCE (IF EXISTS)? name=identifierOrText                          #dropResource
+    | DROP ROW POLICY FOR ROLE roleNames+=identifierOrText
+        (COMMA roleNames+=identifierOrText)*                                    #dropRowPolicyByRoles
     | DROP ROW POLICY (IF EXISTS)? policyName=identifier
         ON tableName=multipartIdentifier
         (FOR (userIdentify | ROLE roleName=identifier))?                        #dropRowPolicy

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

@@ -744,6 +744,7 @@
 import org.apache.doris.nereids.trees.plans.commands.DropResourceCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropRoleCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropRoleMappingCommand;
+import org.apache.doris.nereids.trees.plans.commands.DropRowPolicyByRolesCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropRowPolicyCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropSqlBlockRuleCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropStageCommand;
@@ -9020,6 +9021,14 @@ public LogicalPlan visitDropRowPolicy(DorisParser.DropRowPolicyContext ctx) {
         return new DropRowPolicyCommand(ifExist, policyName, tableNameInfo, userIdentity, roleName);
     }
 
+    @Override
+    public LogicalPlan visitDropRowPolicyByRoles(DorisParser.DropRowPolicyByRolesContext ctx) {
+        List<String> roleNames = ctx.roleNames.stream()
+                .map(ParseTree::getText)
+                .collect(Collectors.toList());
+        return new DropRowPolicyByRolesCommand(roleNames);
+    }
+
     @Override
     public LogicalPlan visitTransactionBegin(DorisParser.TransactionBeginContext ctx) {
         if (ctx.LABEL() != null) {

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

@@ -443,6 +443,7 @@ public enum PlanType {
     STOP_DATA_SYNC_JOB_COMMAND,
     DROP_RESOURCE_COMMAND,
     DROP_ROW_POLICY_COMMAND,
+    DROP_ROW_POLICY_BY_ROLES_COMMAND,
     TRANSACTION_BEGIN_COMMAND,
     TRANSACTION_COMMIT_COMMAND,
     TRANSACTION_ROLLBACK_COMMAND,

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

@@ -0,0 +1,85 @@
+// 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.StmtType;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+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.StmtExecutor;
+
+import java.util.ArrayList;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * DropRowPolicyByRolesCommand
+ * Drop all row policies bound to the specified roles.
+ * Syntax: DROP ROW POLICY FOR ROLE role1, role2, ...
+ **/
+public class DropRowPolicyByRolesCommand extends DropCommand {
+    private final List<String> roleNames;
+
+    public DropRowPolicyByRolesCommand(List<String> roleNames) {
+        super(PlanType.DROP_ROW_POLICY_BY_ROLES_COMMAND);
+        this.roleNames = roleNames;
+    }
+
+    @Override
+    public void doRun(ConnectContext ctx, StmtExecutor executor) throws Exception {
+        validate(ctx);
+        Set<String> uniqueRoleNames = new LinkedHashSet<>(roleNames);
+        Env.getCurrentEnv().getPolicyMgr().dropRowPoliciesByRoles(new ArrayList<>(uniqueRoleNames));
+    }
+
+    /**
+     * validate
+     */
+    public void validate(ConnectContext ctx) throws AnalysisException {
+        for (String roleName : roleNames) {
+            if (roleName == null || roleName.isEmpty()) {
+                throw new AnalysisException("role name is empty");
+            }
+        }
+        if (!Env.getCurrentEnv().getAccessManager()
+                .checkGlobalPriv(ConnectContext.get(), PrivPredicate.GRANT)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR,
+                    PrivPredicate.GRANT.getPrivs().toString());
+        }
+    }
+
+    public List<String> getRoleNames() {
+        return roleNames;
+    }
+
+    @Override
+    public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+        return visitor.visitDropRowPolicyByRolesCommand(this, context);
+    }
+
+    @Override
+    public StmtType stmtType() {
+        return StmtType.DROP;
+    }
+}

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

@@ -130,6 +130,7 @@
 import org.apache.doris.nereids.trees.plans.commands.DropResourceCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropRoleCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropRoleMappingCommand;
+import org.apache.doris.nereids.trees.plans.commands.DropRowPolicyByRolesCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropRowPolicyCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropSqlBlockRuleCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropStageCommand;
@@ -1310,6 +1311,10 @@ default R visitDropRowPolicyCommand(DropRowPolicyCommand dropRowPolicyCommand, C
         return visitCommand(dropRowPolicyCommand, context);
     }
 
+    default R visitDropRowPolicyByRolesCommand(DropRowPolicyByRolesCommand dropRowPolicyByRolesCommand, C context) {
+        return visitCommand(dropRowPolicyByRolesCommand, context);
+    }
+
     default R visitTransactionBeginCommand(TransactionBeginCommand transactionBeginCommand, C context) {
         return visitCommand(transactionBeginCommand, context);
     }

fe/fe-core/src/main/java/org/apache/doris/policy/PolicyMgr.java

@@ -196,6 +196,22 @@ public void dropPolicy(DropPolicyLog dropPolicyLog, boolean ifExists) throws Ddl
         }
     }
 
+    public void dropRowPoliciesByRoles(List<String> roleNames) throws DdlException {
+        writeLock();
+        try {
+            for (String roleName : roleNames) {
+                DropPolicyLog dropPolicyLog = new DropPolicyLog(
+                        null, null, null, PolicyTypeEnum.ROW, null, null, roleName);
+                if (existPolicy(dropPolicyLog)) {
+                    unprotectedDrop(dropPolicyLog);
+                    Env.getCurrentEnv().getEditLog().logDropPolicy(dropPolicyLog);
+                }
+            }
+        } finally {
+            writeUnlock();
+        }
+    }
+
     /**
      * Check whether the policy exist.
      *

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

@@ -0,0 +1,106 @@
+// 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.jmockit.Deencapsulation;
+import org.apache.doris.mysql.privilege.AccessControllerManager;
+import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.nereids.parser.NereidsParser;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.utframe.TestWithFeService;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+public class DropRowPolicyByRolesCommandTest extends TestWithFeService {
+    private ConnectContext connectContext;
+    private Env env;
+    private AccessControllerManager accessControllerManager;
+
+    private void runBefore() throws IOException {
+        connectContext = createDefaultCtx();
+        env = Env.getCurrentEnv();
+        accessControllerManager = env.getAccessManager();
+    }
+
+    @Test
+    public void testValidateNormal() throws Exception {
+        runBefore();
+        AccessControllerManager spyAcm = Mockito.spy(accessControllerManager);
+        Mockito.doReturn(true).when(spyAcm).checkGlobalPriv(
+                Mockito.nullable(ConnectContext.class), Mockito.eq(PrivPredicate.GRANT));
+        Deencapsulation.setField(env, "accessManager", spyAcm);
+        DropRowPolicyByRolesCommand command = new DropRowPolicyByRolesCommand(
+                Arrays.asList("role1", "role2"));
+        Assertions.assertDoesNotThrow(() -> command.validate(connectContext));
+    }
+
+    @Test
+    public void testValidateNoPrivilege() throws Exception {
+        runBefore();
+        AccessControllerManager spyAcm = Mockito.spy(accessControllerManager);
+        Mockito.doReturn(false).when(spyAcm).checkGlobalPriv(
+                Mockito.nullable(ConnectContext.class), Mockito.eq(PrivPredicate.GRANT));
+        Deencapsulation.setField(env, "accessManager", spyAcm);
+        DropRowPolicyByRolesCommand command = new DropRowPolicyByRolesCommand(
+                Arrays.asList("role1"));
+        Assertions.assertThrows(Exception.class, () -> command.validate(connectContext));
+    }
+
+    @Test
+    public void testValidateEmptyRoleName() throws Exception {
+        runBefore();
+        AccessControllerManager spyAcm = Mockito.spy(accessControllerManager);
+        Mockito.doReturn(true).when(spyAcm).checkGlobalPriv(
+                Mockito.nullable(ConnectContext.class), Mockito.eq(PrivPredicate.GRANT));
+        Deencapsulation.setField(env, "accessManager", spyAcm);
+        DropRowPolicyByRolesCommand command = new DropRowPolicyByRolesCommand(
+                Arrays.asList(""));
+        Assertions.assertThrows(Exception.class, () -> command.validate(connectContext));
+    }
+
+    @Test
+    public void testGetRoleNames() {
+        DropRowPolicyByRolesCommand command = new DropRowPolicyByRolesCommand(
+                Arrays.asList("role1", "role2", "role3"));
+        Assertions.assertEquals(Arrays.asList("role1", "role2", "role3"), command.getRoleNames());
+    }
+
+    @Test
+    public void testParseBacktickRoleName() {
+        NereidsParser parser = new NereidsParser();
+        String[] sqls = {
+            "DROP ROW POLICY FOR ROLE role1",
+            "DROP ROW POLICY FOR ROLE role1, role2",
+            "DROP ROW POLICY FOR ROLE `test-role`",
+            "DROP ROW POLICY FOR ROLE `test-role`, role2",
+        };
+        for (String sql : sqls) {
+            Assertions.assertDoesNotThrow(() -> {
+                Plan plan = parser.parseSingle(sql);
+                Assertions.assertInstanceOf(DropRowPolicyByRolesCommand.class, plan);
+            }, "Failed to parse: " + sql);
+        }
+    }
+}