[Enhancement] (nereids)implement DropStoragePolicyCommand in nereids

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

@@ -219,6 +219,7 @@ supportedDropStatement
     | DROP ROLE (IF EXISTS)? name=identifier                                    #dropRole
     | DROP SQL_BLOCK_RULE (IF EXISTS)? identifierSeq                            #dropSqlBlockRule
     | DROP USER (IF EXISTS)? userIdentify                                       #dropUser
+    | DROP STORAGE POLICY (IF EXISTS)? name=identifier                          #dropStoragePolicy
     | DROP WORKLOAD GROUP (IF EXISTS)? name=identifierOrText                    #dropWorkloadGroup
     | DROP CATALOG (IF EXISTS)? name=identifier                                 #dropCatalog
     | DROP FILE name=STRING_LITERAL
@@ -699,7 +700,6 @@ unsupportedDropStatement
     | DROP ROW POLICY (IF EXISTS)? policyName=identifier
         ON tableName=multipartIdentifier
         (FOR (userIdentify | ROLE roleName=identifier))?                        #dropRowPolicy
-    | DROP STORAGE POLICY (IF EXISTS)? name=identifier                          #dropStoragePolicy
     | DROP STAGE (IF EXISTS)? name=identifier                                   #dropStage
     ;
 

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

@@ -130,6 +130,7 @@
 import org.apache.doris.nereids.DorisParser.DropRepositoryContext;
 import org.apache.doris.nereids.DorisParser.DropRoleContext;
 import org.apache.doris.nereids.DorisParser.DropSqlBlockRuleContext;
+import org.apache.doris.nereids.DorisParser.DropStoragePolicyContext;
 import org.apache.doris.nereids.DorisParser.DropUserContext;
 import org.apache.doris.nereids.DorisParser.DropWorkloadGroupContext;
 import org.apache.doris.nereids.DorisParser.DropWorkloadPolicyContext;
@@ -541,6 +542,7 @@
 import org.apache.doris.nereids.trees.plans.commands.DropRepositoryCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropRoleCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropSqlBlockRuleCommand;
+import org.apache.doris.nereids.trees.plans.commands.DropStoragePolicyCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropUserCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropWorkloadGroupCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropWorkloadPolicyCommand;
@@ -4995,6 +4997,13 @@ public LogicalPlan visitCreateEncryptkey(CreateEncryptkeyContext ctx) {
                                             stripQuotes(ctx.STRING_LITERAL().getText()));
     }
 
+    @Override
+    public LogicalPlan visitDropStoragePolicy(DropStoragePolicyContext ctx) {
+        String policyName = ctx.name.getText();
+        boolean ifExists = ctx.EXISTS() != null;
+        return new DropStoragePolicyCommand(policyName, ifExists);
+    }
+
     @Override
     public LogicalPlan visitDropEncryptkey(DropEncryptkeyContext ctx) {
         List<String> nameParts = visitMultipartIdentifier(ctx.name);

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

@@ -200,6 +200,7 @@ public enum PlanType {
     EXECUTE_COMMAND,
     DROP_SQL_BLOCK_RULE_COMMAND,
     DROP_USER_COMMAND,
+    DROP_STORAGE_POLICY_COMMAND,
     DROP_WORKLOAD_GROUP_NAME,
     DROP_WORKLOAD_POLICY_COMMAND,
     ADMIN_SET_TABLE_STATUS_COMMAND,

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

@@ -0,0 +1,68 @@
+// 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.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.policy.DropPolicyLog;
+import org.apache.doris.policy.PolicyTypeEnum;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.StmtExecutor;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+/**
+ * drop storage policy command
+ */
+public class DropStoragePolicyCommand extends DropCommand {
+    public static final Logger LOG = LogManager.getLogger(DropStoragePolicyCommand.class);
+    private final boolean ifExists;
+    private final String policyName;
+
+    /**
+     * constructor
+     */
+    public DropStoragePolicyCommand(String policyName, boolean ifExists) {
+        super(PlanType.DROP_STORAGE_POLICY_COMMAND);
+        this.policyName = policyName;
+        this.ifExists = ifExists;
+    }
+
+    @Override
+    public void doRun(ConnectContext ctx, StmtExecutor executor) throws Exception {
+        if (!Env.getCurrentEnv().getAccessManager()
+                .checkGlobalPriv(ConnectContext.get(), PrivPredicate.ADMIN)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR,
+                        PrivPredicate.ADMIN.getPrivs().toString());
+        }
+        // Drop the storage policy
+        DropPolicyLog dropPolicyLog = new DropPolicyLog(PolicyTypeEnum.STORAGE, policyName);
+        Env.getCurrentEnv().getPolicyMgr().dropPolicy(dropPolicyLog, ifExists);
+    }
+
+    @Override
+    public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+        return visitor.visitDropStoragePolicyCommand(this, context);
+    }
+}
+

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

@@ -67,6 +67,7 @@
 import org.apache.doris.nereids.trees.plans.commands.DropRepositoryCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropRoleCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropSqlBlockRuleCommand;
+import org.apache.doris.nereids.trees.plans.commands.DropStoragePolicyCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropUserCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropWorkloadGroupCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropWorkloadPolicyCommand;
@@ -401,6 +402,10 @@ default R visitSetDefaultStorageVault(SetDefaultStorageVaultCommand setDefaultSt
         return visitCommand(setDefaultStorageVaultCommand, context);
     }
 
+    default R visitDropStoragePolicyCommand(DropStoragePolicyCommand dropStoragePolicyCommand, C context) {
+        return visitCommand(dropStoragePolicyCommand, context);
+    }
+
     default R visitRefreshCatalogCommand(RefreshCatalogCommand refreshCatalogCommand, C context) {
         return visitCommand(refreshCatalogCommand, context);
     }

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

@@ -171,11 +171,7 @@ public void addPolicy(Policy policy) throws UserException {
         }
     }
 
-    /**
-     * Drop policy through stmt.
-     **/
-    public void dropPolicy(DropPolicyStmt stmt) throws DdlException, AnalysisException {
-        DropPolicyLog dropPolicyLog = DropPolicyLog.fromDropStmt(stmt);
+    public void dropPolicy(DropPolicyLog dropPolicyLog, boolean ifExists) throws DdlException, AnalysisException {
         if (dropPolicyLog.getType() == PolicyTypeEnum.STORAGE) {
             List<Database> databases = Env.getCurrentEnv().getInternalCatalog().getDbs();
             for (Database db : databases) {
@@ -198,7 +194,7 @@ public void dropPolicy(DropPolicyStmt stmt) throws DdlException, AnalysisExcepti
         writeLock();
         try {
             if (!existPolicy(dropPolicyLog)) {
-                if (stmt.isIfExists()) {
+                if (ifExists) {
                     return;
                 }
                 throw new DdlException("the policy " + dropPolicyLog.getPolicyName() + " not exist");
@@ -210,6 +206,14 @@ public void dropPolicy(DropPolicyStmt stmt) throws DdlException, AnalysisExcepti
         }
     }
 
+    /**
+     * Drop policy through stmt.
+     **/
+    public void dropPolicy(DropPolicyStmt stmt) throws DdlException, AnalysisException {
+        DropPolicyLog dropPolicyLog = DropPolicyLog.fromDropStmt(stmt);
+        dropPolicy(dropPolicyLog, stmt.isIfExists());
+    }
+
     /**
      * Check whether the policy exist.
      *

regression-test/suites/nereids_p0/test_drop_storage_policy_command.groovy

@@ -0,0 +1,63 @@
+// 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_drop_storage_policy_command", "drop,storage_policy") {
+    String policyName = "test_drop_storage_policy";
+    String resourceName = "test_drop_storage_policy_resource";
+    try {
+        // Drop existing storage policy and resource if they exist before creating new ones
+        try_sql("DROP STORAGE POLICY IF EXISTS ${policyName}")
+        try_sql("DROP RESOURCE IF EXISTS ${resourceName}")
+        // Create a new resource to be used in the storage policy
+        sql """
+            CREATE RESOURCE IF NOT EXISTS "${resourceName}"
+            PROPERTIES(
+                "type"="s3",
+                "AWS_ENDPOINT" = "${getS3Endpoint()}",
+                "AWS_REGION" = "${getS3Region()}",
+                "AWS_ROOT_PATH" = "regression/cooldown",
+                "AWS_ACCESS_KEY" = "${getS3AK()}",
+                "AWS_SECRET_KEY" = "${getS3SK()}",
+                "AWS_MAX_CONNECTIONS" = "50",
+                "AWS_REQUEST_TIMEOUT_MS" = "3000",
+                "AWS_CONNECTION_TIMEOUT_MS" = "1000",
+                "AWS_BUCKET" = "${getS3BucketName()}",
+                "s3_validity_check" = "false"
+            );
+        """
+
+        // Create a new storage policy to test the SHOW STORAGE POLICY command
+        sql """
+            CREATE STORAGE POLICY IF NOT EXISTS ${policyName}
+            PROPERTIES(
+                "storage_resource" = "${resourceName}",
+                "cooldown_ttl" = "300"
+            )
+        """
+
+        checkNereidsExecute("DROP STORAGE POLICY IF EXISTS ${policyName}")
+
+    } catch (Exception e) {
+        // Log any exceptions that occur during testing
+        log.error("Failed to execute SHOW STORAGE POLICY command", e)
+        throw e
+    } finally {
+        // Clean up by dropping the storage policy and resource if they still exist
+        try_sql("DROP STORAGE POLICY IF EXISTS ${policyName}")
+        try_sql("DROP RESOURCE IF EXISTS ${resourceName}")
+    }
+}