ddl: Fix unstable DROP TABLE/FLASHBACK TABLE/RECOVER TABLE

dbms/src/Debug/MockSchemaGetter.h

@@ -39,6 +39,11 @@ struct MockSchemaGetter
         return MockTiDB::instance().getTableInfoByID(table_id);
     }
 
+    static std::pair<TiDB::TableInfoPtr, bool> getTableInfoAndCheckMvcc(DatabaseID db_id, TableID table_id)
+    {
+        return {getTableInfo(db_id, table_id), false};
+    }
+
     static std::tuple<TiDB::DBInfoPtr, TiDB::TableInfoPtr> getDatabaseAndTableInfo(DatabaseID db_id, TableID table_id)
     {
         return std::make_tuple(getDatabase(db_id), getTableInfo(db_id, table_id));

dbms/src/TiDB/Schema/SchemaBuilder.h

@@ -57,7 +57,7 @@ struct SchemaBuilder
 
     void dropAllSchema();
 
-    bool applyTable(DatabaseID database_id, TableID logical_table_id, TableID physical_table_id);
+    bool applyTable(DatabaseID database_id, TableID logical_table_id, TableID physical_table_id, bool force);
 
 private:
     void applyDropSchema(DatabaseID schema_id);
@@ -69,13 +69,16 @@ struct SchemaBuilder
 
     void applyCreateSchema(const TiDB::DBInfoPtr & db_info);
 
-    void applyCreateStorageInstance(const TiDB::DBInfoPtr & db_info, const TiDB::TableInfoPtr & table_info);
+    void applyCreateStorageInstance(
+        const TiDB::DBInfoPtr & db_info,
+        const TiDB::TableInfoPtr & table_info,
+        bool is_tombstone);
 
     void applyDropTable(DatabaseID database_id, TableID table_id);
 
     void applyRecoverTable(DatabaseID database_id, TiDB::TableID table_id);
-
-    void applyRecoverPhysicalTable(const TiDB::DBInfoPtr & db_info, const TiDB::TableInfoPtr & table_info);
+    void applyRecoverLogicalTable(const TiDB::DBInfoPtr & db_info, const TiDB::TableInfoPtr & table_info);
+    bool tryRecoverPhysicalTable(const TiDB::DBInfoPtr & db_info, const TiDB::TableInfoPtr & table_info);
 
     /// Parameter schema_name should be mapped.
     void applyDropPhysicalTable(const String & db_name, TableID table_id);

dbms/src/TiDB/Schema/SchemaGetter.cpp

@@ -283,40 +283,39 @@ TiDB::DBInfoPtr SchemaGetter::getDatabase(DatabaseID db_id)
 }
 
 template <bool mvcc_get>
-TiDB::TableInfoPtr SchemaGetter::getTableInfoImpl(DatabaseID db_id, TableID table_id)
+std::pair<TiDB::TableInfoPtr, bool> SchemaGetter::getTableInfoImpl(DatabaseID db_id, TableID table_id)
 {
     String db_key = getDBKey(db_id);
-    if (!checkDBExists(db_key))
-    {
-        LOG_ERROR(log, "The database does not exist, database_id={}", db_id);
-        return nullptr;
-    }
+    // Note: Do not check the existence of `db_key` here, otherwise we can not
+    //       get the table info after database is dropped.
     String table_key = getTableKey(table_id);
     String table_info_json = TxnStructure::hGet(snap, db_key, table_key);
+    bool get_by_mvcc = false;
     if (table_info_json.empty())
     {
         if constexpr (!mvcc_get)
         {
-            return nullptr;
+            return {nullptr, false};
         }
 
         LOG_WARNING(log, "The table is dropped in TiKV, try to get the latest table_info, table_id={}", table_id);
         table_info_json = TxnStructure::mvccGet(snap, db_key, table_key);
+        get_by_mvcc = true;
         if (table_info_json.empty())
         {
             LOG_ERROR(
                 log,
                 "The table is dropped in TiKV, and the latest table_info is still empty, it should be GCed, "
                 "table_id={}",
                 table_id);
-            return nullptr;
+            return {nullptr, get_by_mvcc};
         }
     }
     LOG_DEBUG(log, "Get Table Info from TiKV, table_id={} {}", table_id, table_info_json);
-    return std::make_shared<TiDB::TableInfo>(table_info_json, keyspace_id);
+    return {std::make_shared<TiDB::TableInfo>(table_info_json, keyspace_id), get_by_mvcc};
 }
-template TiDB::TableInfoPtr SchemaGetter::getTableInfoImpl<false>(DatabaseID db_id, TableID table_id);
-template TiDB::TableInfoPtr SchemaGetter::getTableInfoImpl<true>(DatabaseID db_id, TableID table_id);
+template std::pair<TiDB::TableInfoPtr, bool> SchemaGetter::getTableInfoImpl<false>(DatabaseID db_id, TableID table_id);
+template std::pair<TiDB::TableInfoPtr, bool> SchemaGetter::getTableInfoImpl<true>(DatabaseID db_id, TableID table_id);
 
 std::tuple<TiDB::DBInfoPtr, TiDB::TableInfoPtr> SchemaGetter::getDatabaseAndTableInfo(
     DatabaseID db_id,

dbms/src/TiDB/Schema/SchemaGetter.h

@@ -168,8 +168,13 @@ struct SchemaGetter
     TiDB::TableInfoPtr getTableInfo(DatabaseID db_id, TableID table_id, bool try_mvcc = true)
     {
         if (try_mvcc)
-            return getTableInfoImpl</*mvcc_get*/ true>(db_id, table_id);
-        return getTableInfoImpl</*mvcc_get*/ false>(db_id, table_id);
+            return getTableInfoImpl</*mvcc_get*/ true>(db_id, table_id).first;
+        return getTableInfoImpl</*mvcc_get*/ false>(db_id, table_id).first;
+    }
+
+    std::pair<TiDB::TableInfoPtr, bool> getTableInfoAndCheckMvcc(DatabaseID db_id, TableID table_id)
+    {
+        return getTableInfoImpl</*mvcc_get*/ true>(db_id, table_id);
     }
 
     std::tuple<TiDB::DBInfoPtr, TiDB::TableInfoPtr> getDatabaseAndTableInfo(DatabaseID db_id, TableID table_id);
@@ -182,7 +187,7 @@ struct SchemaGetter
 
 private:
     template <bool mvcc_get>
-    TiDB::TableInfoPtr getTableInfoImpl(DatabaseID db_id, TableID table_id);
+    std::pair<TiDB::TableInfoPtr, bool> getTableInfoImpl(DatabaseID db_id, TableID table_id);
 };
 
 } // namespace DB

dbms/src/TiDB/Schema/TiDBSchemaSyncer.cpp

@@ -158,6 +158,7 @@ std::tuple<bool, String> TiDBSchemaSyncer<mock_getter, mock_mapper>::trySyncTabl
     Context & context,
     TableID physical_table_id,
     Getter & getter,
+    bool force,
     const char * next_action)
 {
     // Get logical_table_id and database_id by physical_table_id.
@@ -177,7 +178,7 @@ std::tuple<bool, String> TiDBSchemaSyncer<mock_getter, mock_mapper>::trySyncTabl
     // If the table schema apply is failed, then we need to update the table-id-mapping
     // and retry.
     SchemaBuilder<Getter, NameMapper> builder(getter, context, databases, table_id_map);
-    if (!builder.applyTable(database_id, logical_table_id, physical_table_id))
+    if (!builder.applyTable(database_id, logical_table_id, physical_table_id, force))
     {
         String message = fmt::format(
             "Can not apply table schema because the table_id_map is not up-to-date, {}."
@@ -207,7 +208,7 @@ bool TiDBSchemaSyncer<mock_getter, mock_mapper>::syncTableSchema(Context & conte
     /// Note that we don't need a lock at the beginning of syncTableSchema.
     /// The AlterLock for storage will be acquired in `SchemaBuilder::applyTable`.
     auto [need_update_id_mapping, message]
-        = trySyncTableSchema(context, physical_table_id, getter, "try to syncSchemas");
+        = trySyncTableSchema(context, physical_table_id, getter, false, "try to syncSchemas");
     if (!need_update_id_mapping)
     {
         LOG_INFO(log, "Sync table schema end, table_id={}", physical_table_id);
@@ -218,8 +219,11 @@ bool TiDBSchemaSyncer<mock_getter, mock_mapper>::syncTableSchema(Context & conte
     GET_METRIC(tiflash_schema_trigger_count, type_sync_table_schema).Increment();
     // Notice: must use the same getter
     syncSchemasByGetter(context, getter);
+    // Try to sync the table schema with `force==true`. Even the table is tombstone (but not physically
+    // dropped in TiKV), it will sync the table schema to handle snapshot or raft commands that come after
+    // table is dropped.
     std::tie(need_update_id_mapping, message)
-        = trySyncTableSchema(context, physical_table_id, getter, "sync table schema fail");
+        = trySyncTableSchema(context, physical_table_id, getter, true, "sync table schema fail");
     if (likely(!need_update_id_mapping))
     {
         LOG_INFO(log, "Sync table schema end after syncSchemas, table_id={}", physical_table_id);

dbms/src/TiDB/Schema/TiDBSchemaSyncer.h

@@ -105,6 +105,7 @@ class TiDBSchemaSyncer : public SchemaSyncer
         Context & context,
         TableID physical_table_id,
         Getter & getter,
+        bool force,
         const char * next_action);
 
     TiDB::DBInfoPtr getDBInfoByName(const String & database_name) override