part2: support drop command

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

@@ -247,7 +247,7 @@ supportedDropStatement
         ((FROM | IN) database=identifier)? properties=propertyClause            #dropFile
     | DROP WORKLOAD POLICY (IF EXISTS)? name=identifierOrText                   #dropWorkloadPolicy
     | DROP REPOSITORY name=identifier                                           #dropRepository
-
+    | DROP DICTIONARY (IF EXISTS)? name=multipartIdentifier                     #dropDictionary
     ;
 
 supportedShowStatement

fe/fe-core/src/main/java/org/apache/doris/dictionary/DictionaryManager.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.dictionary;
 
+import org.apache.commons.compress.utils.Lists;
 import org.apache.doris.catalog.Env;
 import org.apache.doris.common.DdlException;
 import org.apache.doris.common.io.Text;
@@ -35,6 +36,7 @@
 import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
@@ -115,7 +117,8 @@ public Dictionary createDictionary(CreateDictionaryInfo info) throws DdlExceptio
         } finally {
             unlockWrite();
         }
-
+        // The data in BE doesn't always have the same situation with FE(think BE restart). So the data load don't need
+        // transaction safety. It rely on periodic check and update.
         scheduleDataLoad(dictionary);
 
         return dictionary;
@@ -128,6 +131,7 @@ public Dictionary createDictionary(CreateDictionaryInfo info) throws DdlExceptio
      */
     public void dropDictionary(String dbName, String dictName, boolean ifExists) throws DdlException {
         lockWrite();
+        Dictionary dic = null;
         try {
             Map<String, Dictionary> dbDictionaries = dictionaries.get(dbName);
             if (dbDictionaries == null || !dbDictionaries.containsKey(dictName)) {
@@ -136,35 +140,40 @@ public void dropDictionary(String dbName, String dictName, boolean ifExists) thr
                 }
                 return;
             }
-            dbDictionaries.remove(dictName);
-            if (dbDictionaries.isEmpty()) {
-                dictionaries.remove(dbName);
-            }
+            dic = dbDictionaries.remove(dictName);
 
             // Log the drop operation
             Env.getCurrentEnv().getEditLog().logDropDictionary(dbName, dictName);
         } finally {
             unlockWrite();
         }
+        // The data in BE doesn't always have the same situation with FE(think FE crash). But we have periodic report
+        // so that we can drop unknown dictionary at that time.
+        scheduleDataUnload(dic);
     }
 
     /**
      * Drop all dictionaries in a database. Used when dropping a database.
      */
     public void dropDbDictionaries(String dbName) {
         lockWrite();
+        List<Dictionary> droppedDictionaries = Lists.newArrayList();
         try {
             // pop and save item from dictionaries
             Map<String, Dictionary> dbDictionaries = dictionaries.remove(dbName);
             // Log the drop operation
             if (dbDictionaries != null) {
                 for (Map.Entry<String, Dictionary> entry : dbDictionaries.entrySet()) {
+                    droppedDictionaries.add(entry.getValue());
                     Env.getCurrentEnv().getEditLog().logDropDictionary(dbName, entry.getKey());
                 }
             }
         } finally {
             unlockWrite();
         }
+        for (Dictionary dictionary : droppedDictionaries) {
+            scheduleDataUnload(dictionary);
+        }
     }
 
     /**
@@ -215,6 +224,9 @@ private void checkAndUpdateDictionaries() {
         // 3. Handle any errors or inconsistencies
     }
 
+    /// data load and unload are also used for dictionary data check and update to keep data consistency between BE
+    /// and FE. So they are not private.
+
     public void scheduleDataLoad(Dictionary dictionary) {
         // TODO: Implement data load scheduling logic
         // This should:
@@ -223,6 +235,14 @@ public void scheduleDataLoad(Dictionary dictionary) {
         // 3. Monitor the task progress
     }
 
+    public void scheduleDataUnload(Dictionary dictionary) {
+        // TODO: Implement data unload scheduling logic
+        // This should:
+        // 1. Create an unload task
+        // 2. Submit the task to a task executor
+        // 3. Monitor the task progress
+    }
+
     public void replayCreateDictionary(CreateDictionaryPersistInfo info) {
         Dictionary dictionary = info.getDictionary();
         lockWrite();

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

@@ -132,6 +132,7 @@
 import org.apache.doris.nereids.DorisParser.DropCatalogRecycleBinContext;
 import org.apache.doris.nereids.DorisParser.DropColumnClauseContext;
 import org.apache.doris.nereids.DorisParser.DropConstraintContext;
+import org.apache.doris.nereids.DorisParser.DropDictionaryContext;
 import org.apache.doris.nereids.DorisParser.DropEncryptkeyContext;
 import org.apache.doris.nereids.DorisParser.DropFileContext;
 import org.apache.doris.nereids.DorisParser.DropIndexClauseContext;
@@ -524,6 +525,7 @@
 import org.apache.doris.nereids.trees.plans.commands.DropCatalogRecycleBinCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropCatalogRecycleBinCommand.IdType;
 import org.apache.doris.nereids.trees.plans.commands.DropConstraintCommand;
+import org.apache.doris.nereids.trees.plans.commands.DropDictionaryCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropEncryptkeyCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropFileCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropJobCommand;
@@ -5403,4 +5405,22 @@ public LogicalPlan visitCreateDictionary(CreateDictionaryContext ctx) {
         return new CreateDictionaryCommand(ctx.EXISTS() != null, // if not exists
                 dbName, dictName, sCatalogName, sDbName, sTableName, columns, properties, layoutType);
     }
+
+    public LogicalPlan visitDropDictionary(DropDictionaryContext ctx) {
+        List<String> nameParts = visitMultipartIdentifier(ctx.name);
+        if (nameParts.size() == 0 || nameParts.size() > 2) {
+            throw new AnalysisException("Dictionary name should be [db.]dictionary_name");
+        }
+        String dbName;
+        String dictName;
+        if (nameParts.size() == 1) { // only dict name
+            dbName = null;
+            dictName = nameParts.get(0);
+        } else {
+            dbName = nameParts.get(0);
+            dictName = nameParts.get(1);
+        }
+
+        return new DropDictionaryCommand(dbName, dictName, ctx.EXISTS() != null);
+    }
 }

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

@@ -137,7 +137,8 @@ public enum PlanType {
     CREATE_POLICY_COMMAND,
     CREATE_TABLE_COMMAND,
     CREATE_DICTIONARY_COMMAND,
-            CREATE_SQL_BLOCK_RULE_COMMAND,
+    DROP_DICTIONARY_COMMAND,
+    CREATE_SQL_BLOCK_RULE_COMMAND,
     DELETE_COMMAND,
     EXPLAIN_COMMAND,
     EXPORT_COMMAND,

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

@@ -0,0 +1,72 @@
+// 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.nereids.exceptions.AnalysisException;
+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 org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+/**
+ * Command for dropping a dictionary.
+ */
+public class DropDictionaryCommand extends Command {
+    private static final Logger LOG = LogManager.getLogger(DropDictionaryCommand.class);
+
+    private String dbName;
+
+    private final String dictName;
+
+    private final boolean ifExists;
+
+    public DropDictionaryCommand(String dbName, String dictName, boolean ifExists) {
+        super(PlanType.DROP_DICTIONARY_COMMAND);
+        this.dbName = dbName;
+        this.dictName = dictName;
+        this.ifExists = ifExists;
+    }
+
+    @Override
+    public StmtType stmtType() {
+        return StmtType.DDL;
+    }
+
+    @Override
+    public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+        return visitor.visitDropDictionaryCommand(this, context);
+    }
+
+    @Override
+    public void run(ConnectContext ctx, StmtExecutor executor) {
+        if (dbName == null) { // use current database
+            dbName = ctx.getDatabase();
+        }
+        try {
+            ctx.getEnv().getDictionaryManager().dropDictionary(dbName, dictName, ifExists);
+        } catch (Exception e) {
+            String msg = "Failed to drop dictionary " + dictName + " in database " + dbName;
+            LOG.warn(msg, e);
+            throw new AnalysisException(msg + ": " + e.getMessage());
+        }
+    }
+}

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

@@ -61,6 +61,7 @@
 import org.apache.doris.nereids.trees.plans.commands.DropCatalogCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropCatalogRecycleBinCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropConstraintCommand;
+import org.apache.doris.nereids.trees.plans.commands.DropDictionaryCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropEncryptkeyCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropFileCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropJobCommand;
@@ -259,6 +260,10 @@ default R visitDropConstraintCommand(DropConstraintCommand dropConstraintCommand
         return visitCommand(dropConstraintCommand, context);
     }
 
+    default R visitDropDictionaryCommand(DropDictionaryCommand dropDictionaryCommand, C context) {
+        return visitCommand(dropDictionaryCommand, context);
+    }
+
     default R visitDropJobCommand(DropJobCommand dropJobCommand, C context) {
         return visitCommand(dropJobCommand, context);
     }

regression-test/suites/dictionary_p0/test_ddl.groovy

@@ -193,6 +193,11 @@ suite("test_ddl") {
         """
         exception "Dictionary dic1 already exists in database test"
     }
+
+    // test drop
+    sql "drop dictionary dic1"
+    origin_res = sql "show dictionaries"
+    assertTrue(origin_res.size() == 1)
 
     // drop databases
     sql "use mysql"