[Bug] Fix fe meta version problem, make drop meta check code easy to read and add doc content for drop meta check

docs/en/sql-reference/sql-statements/Data Definition/ALTER TABLE.md

@@ -61,7 +61,8 @@ under the License.
         note:
             1) Use a partitioned table to keep at least one partition.
             2) Execute DROP PARTITION For a period of time, the deleted partition can be recovered by the RECOVER statement. See the RECOVER statement for details.
-
+            3) If DROP PARTITION FORCE is executed, the system will not check whether the partition has unfinished transactions, the partition will be deleted directly and cannot be recovered, generally this operation is not recommended
+
     3. Modify the partition properties
         grammar:
             MODIFY PARTITION p1|(p1[, p2, ...]) SET ("key" = "value", ...)

docs/en/sql-reference/sql-statements/Data Definition/DROP DATABASE.md

@@ -31,7 +31,8 @@ Grammar:
 DROP DATABASE [IF EXISTS] db_name;
 
 Explain:
-After executing DROP DATABASE for a period of time, the deleted database can be restored through the RECOVER statement. See RECOVER statement for details
+1) After executing DROP DATABASE for a period of time, the deleted database can be restored through the RECOVER statement. See RECOVER statement for details
+2) If DROP DATABASE FORCE is executed, the system will not check whether the database has unfinished transactions, the database will be deleted directly and cannot be recovered, generally this operation is not recommended
 
 ## example
 1. Delete database db_test

docs/en/sql-reference/sql-statements/Data Definition/DROP TABLE.md

@@ -31,7 +31,8 @@ Grammar:
 DROP TABLE [IF EXISTS] [db_name.]table_name;
 
 Explain:
-After executing DROP TABLE for a period of time, the deleted table can be restored through the RECOVER statement. See RECOVER statement for details
+1) After executing DROP TABLE for a period of time, the deleted table can be restored through the RECOVER statement. See RECOVER statement for details
+2) If DROP TABLE FORCE is executed, the system will not check whether the table has unfinished transactions, the table will be deleted directly and cannot be recovered, generally this operation is not recommended
 
 ## example
 1. Delete a table

docs/zh-CN/sql-reference/sql-statements/Data Definition/ALTER TABLE.md

@@ -57,10 +57,11 @@ under the License.
 
     2. 删除分区
         语法：
-            DROP PARTITION [IF EXISTS] partition_name
+            DROP PARTITION [IF EXISTS] partition_name [FORCE]
         注意：
             1) 使用分区方式的表至少要保留一个分区。
-            2) 执行 DROP PARTITION 一段时间内，可以通过 RECOVER 语句恢复被删除的 partition。详见 RECOVER 语句
+            2) 执行 DROP PARTITION 一段时间内，可以通过 RECOVER 语句恢复被删除的分区。详见 RECOVER 语句
+            3) 如果执行 DROP PARTITION FORCE，则系统不会检查该分区是否存在未完成的事务，分区将直接被删除并且不能被恢复，一般不建议执行此操作
 
     3. 修改分区属性
         语法：

docs/zh-CN/sql-reference/sql-statements/Data Definition/DROP DATABASE.md

@@ -28,15 +28,15 @@ under the License.
 ## description
     该语句用于删除数据库（database）
     语法：
-        DROP DATABASE [IF EXISTS] db_name;
+        DROP DATABASE [IF EXISTS] db_name [FORCE];
 
     说明：
-        执行 DROP DATABASE 一段时间内，可以通过 RECOVER 语句恢复被删除的 database。详见 RECOVER 语句
+        1) 执行 DROP DATABASE 一段时间内，可以通过 RECOVER 语句恢复被删除的数据库。详见 RECOVER 语句
+        2) 如果执行 DROP DATABASE FORCE，则系统不会检查该数据库是否存在未完成的事务，数据库将直接被删除并且不能被恢复，一般不建议执行此操作
 
 ## example
     1. 删除数据库 db_test
         DROP DATABASE db_test;
-
 ## keyword
     DROP,DATABASE
 

docs/zh-CN/sql-reference/sql-statements/Data Definition/DROP TABLE.md

@@ -28,11 +28,12 @@ under the License.
 ## description
     该语句用于删除 table 。
     语法：
-        DROP TABLE [IF EXISTS] [db_name.]table_name;
+        DROP TABLE [IF EXISTS] [db_name.]table_name [FORCE];
 
     说明：
-        执行 DROP TABLE 一段时间内，可以通过 RECOVER 语句恢复被删除的 table。详见 RECOVER 语句
-
+        1) 执行 DROP TABLE 一段时间内，可以通过 RECOVER 语句恢复被删除的表。详见 RECOVER 语句
+        2) 如果执行 DROP TABLE FORCE，则系统不会检查该表是否存在未完成的事务，表将直接被删除并且不能被恢复，一般不建议执行此操作
+
 ## example
     1. 删除一个 table
         DROP TABLE my_table;

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

@@ -885,7 +885,7 @@ alter_table_clause ::=
     :}
     | KW_DROP opt_tmp:isTempPartition KW_PARTITION opt_force:force opt_if_exists:ifExists ident:partitionName
     {:
-        RESULT = new DropPartitionClause(ifExists, partitionName, isTempPartition, force ? !force : !isTempPartition);
+        RESULT = new DropPartitionClause(ifExists, partitionName, isTempPartition, force ? force : isTempPartition);
     :}
     | KW_MODIFY KW_PARTITION ident:partitionName KW_SET LPAREN key_value_map:properties RPAREN
     {:
@@ -1578,7 +1578,7 @@ drop_stmt ::=
     /* Database */
     KW_DROP KW_DATABASE opt_force:force opt_if_exists:ifExists ident:db
     {:
-        RESULT = new DropDbStmt(ifExists, db, !force);
+        RESULT = new DropDbStmt(ifExists, db, force);
     :}
     | KW_DROP KW_SCHEMA opt_force:force opt_if_exists:ifExists ident:db
     {:
@@ -1597,7 +1597,7 @@ drop_stmt ::=
     /* Table */
     | KW_DROP KW_TABLE opt_force:force opt_if_exists:ifExists table_name:name
     {:
-        RESULT = new DropTableStmt(ifExists, name, !force);
+        RESULT = new DropTableStmt(ifExists, name, force);
     :}
     /* User */
     | KW_DROP KW_USER user_identity:userId

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

@@ -20,7 +20,6 @@
 import org.apache.doris.catalog.Catalog;
 import org.apache.doris.catalog.InfoSchemaDb;
 import org.apache.doris.cluster.ClusterNamespace;
-import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.ErrorCode;
 import org.apache.doris.common.ErrorReport;
 import org.apache.doris.common.UserException;
@@ -33,12 +32,12 @@
 public class DropDbStmt extends DdlStmt {
     private boolean ifExists;
     private String dbName;
-    private boolean needCheckCommittedTxns;
+    private boolean forceDrop;
 
-    public DropDbStmt(boolean ifExists, String dbName, boolean needCheckCommittedTxns) {
+    public DropDbStmt(boolean ifExists, String dbName, boolean forceDrop) {
         this.ifExists = ifExists;
         this.dbName = dbName;
-        this.needCheckCommittedTxns = needCheckCommittedTxns;
+        this.forceDrop = forceDrop;
     }
 
     public boolean isSetIfExists() {
@@ -49,12 +48,12 @@ public String getDbName() {
         return this.dbName;
     }
 
-    public boolean isNeedCheckCommittedTxns() {
-        return this.needCheckCommittedTxns;
+    public boolean isForceDrop() {
+        return this.forceDrop;
     }
 
     @Override
-    public void analyze(Analyzer analyzer) throws AnalysisException, UserException {
+    public void analyze(Analyzer analyzer) throws UserException {
         super.analyze(analyzer);
         if (Strings.isNullOrEmpty(dbName)) {
             ErrorReport.reportAnalysisException(ErrorCode.ERR_WRONG_DB_NAME, dbName);

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

@@ -32,15 +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;
+    private boolean forceDrop;
 
-    public DropPartitionClause(boolean ifExists, String partitionName, boolean isTempPartition, boolean needCheckCommittedTxns) {
+    public DropPartitionClause(boolean ifExists, String partitionName, boolean isTempPartition, boolean forceDrop) {
         super(AlterOpType.DROP_PARTITION);
         this.ifExists = ifExists;
         this.partitionName = partitionName;
         this.isTempPartition = isTempPartition;
         this.needTableStable = false;
-        this.needCheckCommittedTxns = needCheckCommittedTxns;
+        this.forceDrop = forceDrop;
     }
 
     public boolean isSetIfExists() {
@@ -55,8 +55,8 @@ public boolean isTempPartition() {
         return isTempPartition;
     }
 
-    public boolean isNeedCheckCommittedTxns() {
-        return needCheckCommittedTxns;
+    public boolean isForceDrop() {
+        return forceDrop;
     }
 
     @Override

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

@@ -18,7 +18,6 @@
 package org.apache.doris.analysis;
 
 import org.apache.doris.catalog.Catalog;
-import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.ErrorCode;
 import org.apache.doris.common.ErrorReport;
 import org.apache.doris.common.UserException;
@@ -32,20 +31,20 @@ public class DropTableStmt extends DdlStmt {
     private boolean ifExists;
     private final TableName tableName;
     private final boolean isView;
-    private boolean needCheckCommittedTxns;
+    private boolean forceDrop;
 
-    public DropTableStmt(boolean ifExists, TableName tableName, boolean needCheckCommittedTxns) {
+    public DropTableStmt(boolean ifExists, TableName tableName, boolean forceDrop) {
         this.ifExists = ifExists;
         this.tableName = tableName;
         this.isView = false;
-        this.needCheckCommittedTxns = needCheckCommittedTxns;
+        this.forceDrop = forceDrop;
     }
 
-    public DropTableStmt(boolean ifExists, TableName tableName, boolean isView, boolean needCheckCommittedTxns) {
+    public DropTableStmt(boolean ifExists, TableName tableName, boolean isView, boolean forceDrop) {
         this.ifExists = ifExists;
         this.tableName = tableName;
         this.isView = isView;
-        this.needCheckCommittedTxns = needCheckCommittedTxns;
+        this.forceDrop = forceDrop;
     }
 
     public boolean isSetIfExists() {
@@ -64,12 +63,12 @@ public boolean isView() {
         return isView;
     }
 
-    public boolean isNeedCheckCommittedTxns() {
-        return this.needCheckCommittedTxns;
+    public boolean isForceDrop() {
+        return this.forceDrop;
     }
 
     @Override
-    public void analyze(Analyzer analyzer) throws AnalysisException, UserException {
+    public void analyze(Analyzer analyzer) throws UserException {
         if (Strings.isNullOrEmpty(tableName.getDb())) {
             tableName.setDb(analyzer.getDefaultDb());
         }

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

@@ -166,6 +166,7 @@
 import org.apache.doris.persist.ClusterInfo;
 import org.apache.doris.persist.ColocatePersistInfo;
 import org.apache.doris.persist.DatabaseInfo;
+import org.apache.doris.persist.DropDbInfo;
 import org.apache.doris.persist.DropInfo;
 import org.apache.doris.persist.DropLinkDbAndUpdateDbInfo;
 import org.apache.doris.persist.DropPartitionInfo;
@@ -2678,7 +2679,7 @@ public void dropDb(DropDbStmt stmt) throws DdlException {
             Database db = this.fullNameToDb.get(dbName);
             db.writeLock();
             try {
-                if (stmt.isNeedCheckCommittedTxns()) {
+                if (!stmt.isForceDrop()) {
                     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)," +
@@ -2720,8 +2721,8 @@ public void dropDb(DropDbStmt stmt) throws DdlException {
 
                 // save table names for recycling
                 Set<String> tableNames = db.getTableNamesWithLock();
-                unprotectDropDb(db, !stmt.isNeedCheckCommittedTxns());
-                if (stmt.isNeedCheckCommittedTxns()) {
+                unprotectDropDb(db, stmt.isForceDrop());
+                if (!stmt.isForceDrop()) {
                     Catalog.getCurrentRecycleBin().recycleDatabase(db, tableNames);
                 }
             } finally {
@@ -2733,12 +2734,13 @@ 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, !stmt.isNeedCheckCommittedTxns());
+            DropDbInfo info = new DropDbInfo(dbName, stmt.isForceDrop());
+            editLog.logDropDb(info);
         } finally {
             unlock();
         }
 
-        LOG.info("finish drop database[{}], is force : {}", dbName, !stmt.isNeedCheckCommittedTxns());
+        LOG.info("finish drop database[{}], is force : {}", dbName, stmt.isForceDrop());
     }
 
     public void unprotectDropDb(Database db, boolean isForeDrop) {
@@ -3333,7 +3335,7 @@ public void dropPartition(Database db, OlapTable olapTable, DropPartitionClause
         if (isTempPartition) {
             olapTable.dropTempPartition(partitionName, true);
         } else {
-            if (clause.isNeedCheckCommittedTxns()) {
+            if (!clause.isForceDrop()) {
                 Partition partition = olapTable.getPartition(partitionName);
                 if (partition != null) {
                     if (Catalog.getCurrentCatalog().getGlobalTransactionMgr().existCommittedTxns(db.getId(), olapTable.getId(), partition.getId())) {
@@ -3343,14 +3345,14 @@ public void dropPartition(Database db, OlapTable olapTable, DropPartitionClause
                     }
                 }
             }
-            olapTable.dropPartition(db.getId(), partitionName, !clause.isNeedCheckCommittedTxns());
+            olapTable.dropPartition(db.getId(), partitionName, clause.isForceDrop());
         }
 
         // log
-        DropPartitionInfo info = new DropPartitionInfo(db.getId(), olapTable.getId(), partitionName, isTempPartition, !clause.isNeedCheckCommittedTxns());
+        DropPartitionInfo info = new DropPartitionInfo(db.getId(), olapTable.getId(), partitionName, isTempPartition, clause.isForceDrop());
         editLog.logDropPartition(info);
 
-        LOG.info("succeed in droping partition[{}], is temp : {}, is force : {}", partitionName, isTempPartition, !clause.isNeedCheckCommittedTxns());
+        LOG.info("succeed in droping partition[{}], is temp : {}, is force : {}", partitionName, isTempPartition, clause.isForceDrop());
     }
 
     public void replayDropPartition(DropPartitionInfo info) {
@@ -4305,21 +4307,21 @@ public void dropTable(DropTableStmt stmt) throws DdlException {
                 }
             }
 
-            if (stmt.isNeedCheckCommittedTxns()) {
+            if (!stmt.isForceDrop()) {
                 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());
+            unprotectDropTable(db, table.getId(), stmt.isForceDrop());
+            DropInfo info = new DropInfo(db.getId(), table.getId(), -1L, stmt.isForceDrop());
             editLog.logDropTable(info);
         } finally {
             db.writeUnlock();
         }
 
-        LOG.info("finished dropping table: {} from db: {}, is force: {}", tableName, dbName, !stmt.isNeedCheckCommittedTxns());
+        LOG.info("finished dropping table: {} from db: {}, is force: {}", tableName, dbName, stmt.isForceDrop());
     }
 
     public boolean unprotectDropTable(Database db, long tableId, boolean isForceDrop) {

fe/fe-core/src/main/java/org/apache/doris/persist/DropDbInfo.java

@@ -56,7 +56,7 @@ private void readFields(DataInput in) throws IOException {
     }
 
     public static DropDbInfo read(DataInput in) throws IOException {
-        if (Catalog.getCurrentCatalogJournalVersion() < FeMetaVersion.VERSION_88) {
+        if (Catalog.getCurrentCatalogJournalVersion() < FeMetaVersion.VERSION_89) {
             DropDbInfo info = new DropDbInfo();
             info.readFields(in);
             return info;

fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java

@@ -889,8 +889,8 @@ public void logCreateDb(Database db) {
         logEdit(OperationType.OP_CREATE_DB, db);
     }
 
-    public void logDropDb(String dbName, boolean isForceDrop) {
-        logEdit(OperationType.OP_DROP_DB, new Text(dbName));
+    public void logDropDb(DropDbInfo dropDbInfo) {
+        logEdit(OperationType.OP_DROP_DB, dropDbInfo);
     }
 
     public void logEraseDb(long dbId) {

fe/fe-core/src/test/java/org/apache/doris/persist/DropDbInfoTest.java

@@ -17,7 +17,6 @@
 
 package org.apache.doris.persist;
 
-import org.apache.doris.catalog.Catalog;
 import org.apache.doris.common.FeMetaVersion;
 import org.apache.doris.meta.MetaContext;
 import org.junit.Assert;