Support check committed txns before catalog drop meta, like db, table, partition etc

fe/fe-core/src/main/cup/sql_parser.cup

@@ -235,7 +235,7 @@ terminal String KW_ADD, KW_ADMIN, KW_AFTER, KW_AGGREGATE, KW_ALL, KW_ALTER, KW_A
     KW_DELETE, KW_DISTINCT, KW_DISTINCTPC, KW_DISTINCTPCSA, KW_DISTRIBUTED, KW_DISTRIBUTION, KW_DYNAMIC, KW_BUCKETS, KW_DIV, KW_DOUBLE, KW_DROP, KW_DROPP, KW_DUPLICATE,
     KW_ELSE, KW_END, KW_ENGINE, KW_ENGINES, KW_ENTER, KW_ERRORS, KW_EVENTS, KW_EXCEPT, KW_EXISTS, KW_EXPORT,
     KW_EXTERNAL, KW_EXTRACT,
-    KW_FALSE, KW_FOLLOWER, KW_FOLLOWING, KW_FREE, KW_FROM, KW_FILE, KW_FIRST, KW_FLOAT, KW_FOR, KW_FORMAT, KW_FRONTEND, KW_FRONTENDS, KW_FULL, KW_FUNCTION, KW_FUNCTIONS,
+    KW_FALSE, KW_FOLLOWER, KW_FOLLOWING, KW_FREE, KW_FROM, KW_FILE, KW_FIRST, KW_FLOAT, KW_FOR, KW_FORCE, KW_FORMAT, KW_FRONTEND, KW_FRONTENDS, KW_FULL, KW_FUNCTION, KW_FUNCTIONS,
     KW_GLOBAL, KW_GRANT, KW_GRANTS, KW_GROUP, KW_GROUPING,
     KW_HASH, KW_HAVING, KW_HELP,KW_HLL, KW_HLL_UNION, KW_HOUR, KW_HUB,
     KW_IDENTIFIED, KW_IF, KW_IN, KW_INDEX, KW_INDEXES, KW_INFILE, KW_INSTALL,
@@ -463,6 +463,7 @@ nonterminal String opt_db, procedure_or_function, opt_comment, opt_engine;
 nonterminal ColumnDef.DefaultValue opt_default_value;
 nonterminal Boolean opt_if_exists, opt_if_not_exists;
 nonterminal Boolean opt_external;
+nonterminal Boolean opt_force;
 nonterminal IndexDef.IndexType opt_index_type;
 
 nonterminal ShowAlterStmt.AlterType opt_alter_type;
@@ -877,9 +878,9 @@ alter_table_clause ::=
     {:
         RESULT = new AddPartitionClause(desc, distribution, properties, isTempPartition);
     :}
-    | KW_DROP opt_tmp:isTempPartition KW_PARTITION opt_if_exists:ifExists ident:partitionName
+    | KW_DROP opt_tmp:isTempPartition KW_PARTITION opt_force:force opt_if_exists:ifExists ident:partitionName
     {:
-        RESULT = new DropPartitionClause(ifExists, partitionName, isTempPartition);
+        RESULT = new DropPartitionClause(ifExists, partitionName, isTempPartition, force ? !force : !isTempPartition);
     :}
     | KW_MODIFY KW_PARTITION ident:partitionName KW_SET LPAREN key_value_map:properties RPAREN
     {:
@@ -1570,13 +1571,13 @@ revoke_stmt ::=
 // Drop statement
 drop_stmt ::=
     /* Database */
-    KW_DROP KW_DATABASE opt_if_exists:ifExists ident:db
+    KW_DROP KW_DATABASE opt_force:force opt_if_exists:ifExists ident:db
     {:
-        RESULT = new DropDbStmt(ifExists, db);
+        RESULT = new DropDbStmt(ifExists, db, !force);
     :}
-    | KW_DROP KW_SCHEMA opt_if_exists:ifExists ident:db
+    | KW_DROP KW_SCHEMA opt_force:force opt_if_exists:ifExists ident:db
     {:
-        RESULT = new DropDbStmt(ifExists, db);
+        RESULT = new DropDbStmt(ifExists, db, !force);
     :}
     /* cluster */
     | KW_DROP KW_CLUSTER opt_if_exists:ifExists ident:cluster
@@ -1589,9 +1590,9 @@ drop_stmt ::=
         RESULT = new DropFunctionStmt(functionName, args);
     :}
     /* Table */
-    | KW_DROP KW_TABLE opt_if_exists:ifExists table_name:name
+    | KW_DROP KW_TABLE opt_force:force opt_if_exists:ifExists table_name:name
     {:
-        RESULT = new DropTableStmt(ifExists, name);
+        RESULT = new DropTableStmt(ifExists, name, !force);
     :}
     /* User */
     | KW_DROP KW_USER user_identity:userId
@@ -1601,7 +1602,7 @@ drop_stmt ::=
     /* View */
     | KW_DROP KW_VIEW opt_if_exists:ifExists table_name:name
     {:
-        RESULT = new DropTableStmt(ifExists, name, true);
+        RESULT = new DropTableStmt(ifExists, name, true, false);
     :}
     | KW_DROP KW_REPOSITORY ident:repoName
     {:
@@ -2020,6 +2021,17 @@ opt_external ::=
     :}
     ;
 
+opt_force ::=
+     /* empty */
+    {:
+        RESULT = false;
+    :}
+    | KW_FORCE
+    {:
+        RESULT = true;
+    :}
+    ;
+
 // Show statement
 show_stmt ::=
     KW_SHOW show_param:stmt

fe/fe-core/src/main/java/org/apache/doris/alter/Alter.java

@@ -143,7 +143,6 @@ public void processDropMaterializedView(DropMaterializedViewStmt stmt) throws Dd
                 throw new DdlException("Table[" + table.getName() + "]'s state is not NORMAL. "
                         + "Do not allow doing DROP ops");
             }
-
             // drop materialized view
             ((MaterializedViewHandler)materializedViewHandler).processDropMaterializedView(stmt, db, olapTable);
 

fe/fe-core/src/main/java/org/apache/doris/alter/MaterializedViewHandler.java

@@ -701,7 +701,7 @@ public void processBatchDropRollup(List<AlterClause> dropRollupClauses, Database
 
     public void processDropMaterializedView(DropMaterializedViewStmt dropMaterializedViewStmt, Database db,
             OlapTable olapTable) throws DdlException, MetaNotFoundException {
-        db.writeLock();
+        Preconditions.checkState(db.isWriteLockHeldByCurrentThread());
         try {
             String mvName = dropMaterializedViewStmt.getMvName();
             // Step1: check drop mv index operation
@@ -710,16 +710,14 @@ public void processDropMaterializedView(DropMaterializedViewStmt dropMaterialize
             long mvIndexId = dropMaterializedView(mvName, olapTable);
             // Step3: log drop mv operation
             EditLog editLog = Catalog.getCurrentCatalog().getEditLog();
-            editLog.logDropRollup(new DropInfo(db.getId(), olapTable.getId(), mvIndexId));
+            editLog.logDropRollup(new DropInfo(db.getId(), olapTable.getId(), mvIndexId, false));
             LOG.info("finished drop materialized view [{}] in table [{}]", mvName, olapTable.getName());
         } catch (MetaNotFoundException e) {
             if (dropMaterializedViewStmt.isIfExists()) {
                 LOG.info(e.getMessage());
             } else {
                 throw e;
             }
-        } finally {
-            db.writeUnlock();
         }
     }
 

fe/fe-core/src/main/java/org/apache/doris/analysis/AlterTableStmt.java

@@ -48,7 +48,6 @@ public List<AlterClause> getOps() {
         return ops;
     }
 
-
     @Override
     public void analyze(Analyzer analyzer) throws UserException {
         super.analyze(analyzer);

fe/fe-core/src/main/java/org/apache/doris/analysis/DropDbStmt.java

@@ -33,10 +33,12 @@
 public class DropDbStmt extends DdlStmt {
     private boolean ifExists;
     private String dbName;
+    private boolean needCheckCommittedTxns;
 
-    public DropDbStmt(boolean ifExists, String dbName) {
+    public DropDbStmt(boolean ifExists, String dbName, boolean needCheckCommittedTxns) {
         this.ifExists = ifExists;
         this.dbName = dbName;
+        this.needCheckCommittedTxns = needCheckCommittedTxns;
     }
 
     public boolean isSetIfExists() {
@@ -47,6 +49,10 @@ public String getDbName() {
         return this.dbName;
     }
 
+    public boolean isNeedCheckCommittedTxns() {
+        return this.needCheckCommittedTxns;
+    }
+
     @Override
     public void analyze(Analyzer analyzer) throws AnalysisException, UserException {
         super.analyze(analyzer);

fe/fe-core/src/main/java/org/apache/doris/analysis/DropPartitionClause.java

@@ -32,13 +32,15 @@ public class DropPartitionClause extends AlterTableClause {
     private String partitionName;
     // true if this is to drop a temp partition
     private boolean isTempPartition;
+    private boolean needCheckCommittedTxns;
 
-    public DropPartitionClause(boolean ifExists, String partitionName, boolean isTempPartition) {
+    public DropPartitionClause(boolean ifExists, String partitionName, boolean isTempPartition, boolean needCheckCommittedTxns) {
         super(AlterOpType.DROP_PARTITION);
         this.ifExists = ifExists;
         this.partitionName = partitionName;
         this.isTempPartition = isTempPartition;
         this.needTableStable = false;
+        this.needCheckCommittedTxns = needCheckCommittedTxns;
     }
 
     public boolean isSetIfExists() {
@@ -53,6 +55,10 @@ public boolean isTempPartition() {
         return isTempPartition;
     }
 
+    public boolean isNeedCheckCommittedTxns() {
+        return needCheckCommittedTxns;
+    }
+
     @Override
     public void analyze(Analyzer analyzer) throws AnalysisException {
         if (Strings.isNullOrEmpty(partitionName)) {

fe/fe-core/src/main/java/org/apache/doris/analysis/DropTableStmt.java

@@ -32,17 +32,20 @@ public class DropTableStmt extends DdlStmt {
     private boolean ifExists;
     private final TableName tableName;
     private final boolean isView;
+    private boolean needCheckCommittedTxns;
 
-    public DropTableStmt(boolean ifExists, TableName tableName) {
+    public DropTableStmt(boolean ifExists, TableName tableName, boolean needCheckCommittedTxns) {
         this.ifExists = ifExists;
         this.tableName = tableName;
         this.isView = false;
+        this.needCheckCommittedTxns = needCheckCommittedTxns;
     }
 
-    public DropTableStmt(boolean ifExists, TableName tableName, boolean isView) {
+    public DropTableStmt(boolean ifExists, TableName tableName, boolean isView, boolean needCheckCommittedTxns) {
         this.ifExists = ifExists;
         this.tableName = tableName;
         this.isView = isView;
+        this.needCheckCommittedTxns = needCheckCommittedTxns;
     }
 
     public boolean isSetIfExists() {
@@ -61,6 +64,10 @@ public boolean isView() {
         return isView;
     }
 
+    public boolean isNeedCheckCommittedTxns() {
+        return this.needCheckCommittedTxns;
+    }
+
     @Override
     public void analyze(Analyzer analyzer) throws AnalysisException, UserException {
         if (Strings.isNullOrEmpty(tableName.getDb())) {

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

@@ -2678,6 +2678,13 @@ public void dropDb(DropDbStmt stmt) throws DdlException {
             Database db = this.fullNameToDb.get(dbName);
             db.writeLock();
             try {
+                if (stmt.isNeedCheckCommittedTxns()) {
+                    if (Catalog.getCurrentCatalog().getGlobalTransactionMgr().existCommittedTxns(db.getId(), null, null)) {
+                       throw new DdlException("There are still some transactions in the COMMITTED state waiting to be completed. " +
+                               "The database [" + dbName +"] cannot be dropped. If you want to forcibly drop(cannot be recovered)," +
+                               " please use \"DROP database FORCE\".");
+                    }
+                }
                 if (db.getDbState() == DbState.LINK && dbName.equals(db.getAttachDb())) {
                     // We try to drop a hard link.
                     final DropLinkDbAndUpdateDbInfo info = new DropLinkDbAndUpdateDbInfo();
@@ -2713,8 +2720,10 @@ public void dropDb(DropDbStmt stmt) throws DdlException {
 
                 // save table names for recycling
                 Set<String> tableNames = db.getTableNamesWithLock();
-                unprotectDropDb(db);
-                Catalog.getCurrentRecycleBin().recycleDatabase(db, tableNames);
+                unprotectDropDb(db, !stmt.isNeedCheckCommittedTxns());
+                if (stmt.isNeedCheckCommittedTxns()) {
+                    Catalog.getCurrentRecycleBin().recycleDatabase(db, tableNames);
+                }
             } finally {
                 db.writeUnlock();
             }
@@ -2724,17 +2733,17 @@ public void dropDb(DropDbStmt stmt) throws DdlException {
             fullNameToDb.remove(db.getFullName());
             final Cluster cluster = nameToCluster.get(db.getClusterName());
             cluster.removeDb(dbName, db.getId());
-            editLog.logDropDb(dbName);
+            editLog.logDropDb(dbName, !stmt.isNeedCheckCommittedTxns());
         } finally {
             unlock();
         }
 
-        LOG.info("finish drop database[{}]", dbName);
+        LOG.info("finish drop database[{}], is force : {}", dbName, !stmt.isNeedCheckCommittedTxns());
     }
 
-    public void unprotectDropDb(Database db) {
+    public void unprotectDropDb(Database db, boolean isForeDrop) {
         for (Table table : db.getTables()) {
-            unprotectDropTable(db, table.getId());
+            unprotectDropTable(db, table.getId(), isForeDrop);
         }
     }
 
@@ -2754,15 +2763,17 @@ public void replayDropLinkDb(DropLinkDbAndUpdateDbInfo info) {
         }
     }
 
-    public void replayDropDb(String dbName) throws DdlException {
+    public void replayDropDb(String dbName, boolean isForceDrop) throws DdlException {
         tryLock(true);
         try {
             Database db = fullNameToDb.get(dbName);
             db.writeLock();
             try {
                 Set<String> tableNames = db.getTableNamesWithLock();
-                unprotectDropDb(db);
-                Catalog.getCurrentRecycleBin().recycleDatabase(db, tableNames);
+                unprotectDropDb(db, isForceDrop);
+                if (!isForceDrop) {
+                    Catalog.getCurrentRecycleBin().recycleDatabase(db, tableNames);
+                }
             } finally {
                 db.writeUnlock();
             }
@@ -3322,14 +3333,24 @@ public void dropPartition(Database db, OlapTable olapTable, DropPartitionClause
         if (isTempPartition) {
             olapTable.dropTempPartition(partitionName, true);
         } else {
-            olapTable.dropPartition(db.getId(), partitionName);
+            if (clause.isNeedCheckCommittedTxns()) {
+                Partition partition = olapTable.getPartition(partitionName);
+                if (partition != null) {
+                    if (Catalog.getCurrentCatalog().getGlobalTransactionMgr().existCommittedTxns(db.getId(), olapTable.getId(), partition.getId())) {
+                        throw new DdlException("There are still some transactions in the COMMITTED state waiting to be completed." +
+                                " The partition [" + partitionName + "] cannot be dropped. If you want to forcibly drop(cannot be recovered)," +
+                                " please use \"DROP partition FORCE\".");
+                    }
+                }
+            }
+            olapTable.dropPartition(db.getId(), partitionName, !clause.isNeedCheckCommittedTxns());
         }
 
         // log
-        DropPartitionInfo info = new DropPartitionInfo(db.getId(), olapTable.getId(), partitionName, isTempPartition);
+        DropPartitionInfo info = new DropPartitionInfo(db.getId(), olapTable.getId(), partitionName, isTempPartition, !clause.isNeedCheckCommittedTxns());
         editLog.logDropPartition(info);
 
-        LOG.info("succeed in droping partition[{}]", partitionName);
+        LOG.info("succeed in droping partition[{}], is temp : {}, is force : {}", partitionName, isTempPartition, !clause.isNeedCheckCommittedTxns());
     }
 
     public void replayDropPartition(DropPartitionInfo info) {
@@ -3340,7 +3361,7 @@ public void replayDropPartition(DropPartitionInfo info) {
             if (info.isTempPartition()) {
                 olapTable.dropTempPartition(info.getPartitionName(), true);
             } else {
-                olapTable.dropPartition(info.getDbId(), info.getPartitionName());
+                olapTable.dropPartition(info.getDbId(), info.getPartitionName(), info.isForceDrop());
             }
         } finally {
             db.writeUnlock();
@@ -4284,18 +4305,24 @@ public void dropTable(DropTableStmt stmt) throws DdlException {
                 }
             }
 
-            unprotectDropTable(db, table.getId());
-
-            DropInfo info = new DropInfo(db.getId(), table.getId(), -1L);
+            if (stmt.isNeedCheckCommittedTxns()) {
+                if (Catalog.getCurrentCatalog().getGlobalTransactionMgr().existCommittedTxns(db.getId(), table.getId(), null)) {
+                    throw new DdlException("There are still some transactions in the COMMITTED state waiting to be completed. " +
+                            "The table [" + tableName +"] cannot be dropped. If you want to forcibly drop(cannot be recovered)," +
+                            " please use \"DROP table FORCE\".");
+                }
+            }
+            unprotectDropTable(db, table.getId(), !stmt.isNeedCheckCommittedTxns());
+            DropInfo info = new DropInfo(db.getId(), table.getId(), -1L, !stmt.isNeedCheckCommittedTxns());
             editLog.logDropTable(info);
         } finally {
             db.writeUnlock();
         }
 
-        LOG.info("finished dropping table: {} from db: {}", tableName, dbName);
+        LOG.info("finished dropping table: {} from db: {}, is force: {}", tableName, dbName, !stmt.isNeedCheckCommittedTxns());
     }
 
-    public boolean unprotectDropTable(Database db, long tableId) {
+    public boolean unprotectDropTable(Database db, long tableId, boolean isForceDrop) {
         Table table = db.getTable(tableId);
         // delete from db meta
         if (table == null) {
@@ -4311,17 +4338,18 @@ public boolean unprotectDropTable(Database db, long tableId) {
         }
 
         db.dropTable(table.getName());
-
-        Catalog.getCurrentRecycleBin().recycleTable(db.getId(), table);
+        if (!isForceDrop) {
+            Catalog.getCurrentRecycleBin().recycleTable(db.getId(), table);
+        }
 
         LOG.info("finished dropping table[{}] in db[{}]", table.getName(), db.getFullName());
         return true;
     }
 
-    public void replayDropTable(Database db, long tableId) {
+    public void replayDropTable(Database db, long tableId, boolean isForceDrop) {
         db.writeLock();
         try {
-            unprotectDropTable(db, tableId);
+            unprotectDropTable(db, tableId, isForceDrop);
         } finally {
             db.writeUnlock();
         }

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

@@ -589,11 +589,7 @@ public void addPartition(Partition partition) {
         nameToPartition.put(partition.getName(), partition);
     }
 
-    public Partition dropPartition(long dbId, String partitionName) {
-        return dropPartition(dbId, partitionName, false);
-    }
-
-    public Partition dropPartition(long dbId, String partitionName, boolean isRestore) {
+    public Partition dropPartition(long dbId, String partitionName, boolean isForceDrop) {
         Partition partition = nameToPartition.get(partitionName);
         if (partition != null) {
             idToPartition.remove(partition.getId());
@@ -602,7 +598,7 @@ public Partition dropPartition(long dbId, String partitionName, boolean isRestor
             Preconditions.checkState(partitionInfo.getType() == PartitionType.RANGE);
             RangePartitionInfo rangePartitionInfo = (RangePartitionInfo) partitionInfo;
 
-            if (!isRestore) {
+            if (!isForceDrop) {
                 // recycle partition
                 Catalog.getCurrentRecycleBin().recyclePartition(dbId, id, partition,
                                           rangePartitionInfo.getRange(partition.getId()),