diff --git a/dbms/src/Debug/MockSchemaGetter.h b/dbms/src/Debug/MockSchemaGetter.h
index 0208ce49c5c..7b4fae8a2e7 100644
--- a/dbms/src/Debug/MockSchemaGetter.h
+++ b/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));
diff --git a/dbms/src/TiDB/Schema/SchemaBuilder.cpp b/dbms/src/TiDB/Schema/SchemaBuilder.cpp
index abf4f2e0997..816ca718c68 100644
--- a/dbms/src/TiDB/Schema/SchemaBuilder.cpp
+++ b/dbms/src/TiDB/Schema/SchemaBuilder.cpp
@@ -43,6 +43,7 @@
 #include <TiDB/Schema/SchemaNameMapper.h>
 #include <TiDB/Schema/TiDB.h>
 #include <common/logger_useful.h>
+#include <fmt/format.h>
 
 #include <boost/algorithm/string/join.hpp>
 #include <magic_enum.hpp>
@@ -66,12 +67,16 @@ bool isReservedDatabase(Context & context, const String & database_name)
 template <typename Getter, typename NameMapper>
 void SchemaBuilder<Getter, NameMapper>::applyCreateTable(DatabaseID database_id, TableID table_id)
 {
-    auto table_info = getter.getTableInfo(database_id, table_id);
-    if (table_info == nullptr) // the database maybe dropped
+    TableInfoPtr table_info;
+    bool get_by_mvcc = false;
+    std::tie(table_info, get_by_mvcc) = getter.getTableInfoAndCheckMvcc(database_id, table_id);
+    if (table_info == nullptr)
     {
         LOG_INFO(
             log,
-            "table is not exist in TiKV, may have been dropped, applyCreateTable is ignored, table_id={}",
+            "table is not exist in TiKV, may have been dropped, applyCreateTable is ignored, database_id={} "
+            "table_id={}",
+            database_id,
             table_id);
         return;
     }
@@ -85,17 +90,32 @@ void SchemaBuilder<Getter, NameMapper>::applyCreateTable(DatabaseID database_id,
         return;
     }
 
-    // If table is partition table, we will create the logical table here.
-    // Because we get the table_info, so we can ensure new_db_info will not be nullptr.
-    auto new_db_info = getter.getDatabase(database_id);
-    applyCreateStorageInstance(new_db_info, table_info);
+    // If table is partition table, we will create the Storage instance for the logical table
+    // here (and store the table info to local).
+    // Because `applyPartitionDiffOnLogicalTable` need the logical table for comparing
+    // the latest partitioning and the local partitioning in table info to apply the changes.
+    auto db_info = getter.getDatabase(database_id);
+    if (unlikely(db_info == nullptr))
+    {
+        // the database has been dropped
+        LOG_INFO(
+            log,
+            "database is not exist in TiKV, may have been dropped, applyCreateTable is ignored, database_id={} "
+            "table_id={}",
+            database_id,
+            table_id);
+        return;
+    }
+    applyCreateStorageInstance(db_info, table_info, get_by_mvcc);
 
     // Register the partition_id -> logical_table_id mapping
     for (const auto & part_def : table_info->partition.definitions)
     {
         LOG_DEBUG(
             log,
-            "register table to table_id_map for partition table, logical_table_id={} physical_table_id={}",
+            "register table to table_id_map for partition table, database_id={} logical_table_id={} "
+            "physical_table_id={}",
+            database_id,
             table_id,
             part_def.id);
         table_id_map.emplacePartitionTableID(part_def.id, table_id);
@@ -174,7 +194,7 @@ void SchemaBuilder<Getter, NameMapper>::applyExchangeTablePartition(const Schema
 
             auto [new_db_info, new_table_info]
                 = getter.getDatabaseAndTableInfo(partition_database_id, partition_logical_table_id);
-            if (new_table_info == nullptr)
+            if (unlikely(new_db_info == nullptr || new_table_info == nullptr))
             {
                 LOG_INFO(
                     log,
@@ -206,7 +226,7 @@ void SchemaBuilder<Getter, NameMapper>::applyExchangeTablePartition(const Schema
 
             auto [new_db_info, new_table_info]
                 = getter.getDatabaseAndTableInfo(non_partition_database_id, partition_physical_table_id);
-            if (new_table_info == nullptr)
+            if (unlikely(new_db_info == nullptr || new_table_info == nullptr))
             {
                 LOG_INFO(
                     log,
@@ -234,6 +254,7 @@ void SchemaBuilder<Getter, NameMapper>::applyExchangeTablePartition(const Schema
 template <typename Getter, typename NameMapper>
 void SchemaBuilder<Getter, NameMapper>::applyDiff(const SchemaDiff & diff)
 {
+    LOG_TRACE(log, "applyDiff accept type={}", magic_enum::enum_name(diff.type));
     switch (diff.type)
     {
     case SchemaActionType::CreateSchema:
@@ -340,7 +361,7 @@ void SchemaBuilder<Getter, NameMapper>::applyDiff(const SchemaDiff & diff)
         {
             // >= SchemaActionType::MaxRecognizedType
             // log down the Int8 value directly
-            LOG_ERROR(log, "Unsupported change type: {}, diff_version={}", static_cast<Int8>(diff.type), diff.version);
+            LOG_ERROR(log, "Unsupported change type: {}, diff_version={}", fmt::underlying(diff.type), diff.version);
         }
 
         break;
@@ -352,16 +373,16 @@ template <typename Getter, typename NameMapper>
 void SchemaBuilder<Getter, NameMapper>::applySetTiFlashReplica(DatabaseID database_id, TableID table_id)
 {
     auto [db_info, table_info] = getter.getDatabaseAndTableInfo(database_id, table_id);
-    if (unlikely(table_info == nullptr))
+    if (unlikely(db_info == nullptr || table_info == nullptr))
     {
         LOG_WARNING(log, "table is not exist in TiKV, applySetTiFlashReplica is ignored, table_id={}", table_id);
         return;
     }
 
+    auto & tmt_context = context.getTMTContext();
     if (table_info->replica_info.count == 0)
     {
         // if set 0, drop table in TiFlash
-        auto & tmt_context = context.getTMTContext();
         auto storage = tmt_context.getStorages().get(keyspace_id, table_info->id);
         if (unlikely(storage == nullptr))
         {
@@ -373,75 +394,107 @@ void SchemaBuilder<Getter, NameMapper>::applySetTiFlashReplica(DatabaseID databa
         }
 
         applyDropTable(db_info->id, table_id);
+        return;
     }
-    else
+
+    assert(table_info->replica_info.count != 0);
+    // Replica info is set to non-zero, create the storage if not exists.
+    auto storage = tmt_context.getStorages().get(keyspace_id, table_info->id);
+    if (storage == nullptr)
     {
-        // if set not 0, we first check whether the storage exists, and then check the replica_count and available
-        auto & tmt_context = context.getTMTContext();
-        auto storage = tmt_context.getStorages().get(keyspace_id, table_info->id);
-        if (storage != nullptr)
+        if (!table_id_map.tableIDInDatabaseIdMap(table_id))
         {
-            if (storage->getTombstone() == 0)
-            {
-                auto managed_storage = std::dynamic_pointer_cast<IManageableStorage>(storage);
-                auto storage_replica_info = managed_storage->getTableInfo().replica_info;
-                if (storage_replica_info.count == table_info->replica_info.count
-                    && storage_replica_info.available == table_info->replica_info.available)
-                {
-                    return;
-                }
-                else
-                {
-                    if (table_info->isLogicalPartitionTable())
-                    {
-                        for (const auto & part_def : table_info->partition.definitions)
-                        {
-                            auto new_part_table_info = table_info->producePartitionTableInfo(part_def.id, name_mapper);
-                            auto part_storage = tmt_context.getStorages().get(keyspace_id, new_part_table_info->id);
-                            if (part_storage != nullptr)
-                            {
-                                auto alter_lock = part_storage->lockForAlter(getThreadNameAndID());
-                                part_storage->alterSchemaChange(
-                                    alter_lock,
-                                    *new_part_table_info,
-                                    name_mapper.mapDatabaseName(db_info->id, keyspace_id),
-                                    name_mapper.mapTableName(*new_part_table_info),
-                                    context);
-                            }
-                            else
-                                table_id_map.emplacePartitionTableID(part_def.id, table_id);
-                        }
-                    }
-                    auto alter_lock = storage->lockForAlter(getThreadNameAndID());
-                    storage->alterSchemaChange(
-                        alter_lock,
-                        *table_info,
-                        name_mapper.mapDatabaseName(db_info->id, keyspace_id),
-                        name_mapper.mapTableName(*table_info),
-                        context);
-                }
-                return;
-            }
-            else
-            {
-                applyRecoverTable(db_info->id, table_id);
-            }
+            applyCreateTable(db_info->id, table_id);
         }
-        else
+        return;
+    }
+
+    // Recover the table if tombstoned
+    if (storage->isTombstone())
+    {
+        applyRecoverLogicalTable(db_info, table_info);
+        return;
+    }
+
+    auto local_logical_storage_table_info = storage->getTableInfo(); // copy
+    // Check whether replica_count and available changed
+    if (const auto & local_logical_storage_replica_info = local_logical_storage_table_info.replica_info;
+        local_logical_storage_replica_info.count == table_info->replica_info.count
+        && local_logical_storage_replica_info.available == table_info->replica_info.available)
+    {
+        return; // nothing changed
+    }
+
+    size_t old_replica_count = 0;
+    size_t new_replica_count = 0;
+    if (table_info->isLogicalPartitionTable())
+    {
+        for (const auto & part_def : table_info->partition.definitions)
         {
-            if (!table_id_map.tableIDInDatabaseIdMap(table_id))
+            auto new_part_table_info = table_info->producePartitionTableInfo(part_def.id, name_mapper);
+            auto part_storage = tmt_context.getStorages().get(keyspace_id, new_part_table_info->id);
+            if (part_storage == nullptr)
             {
-                applyCreateTable(db_info->id, table_id);
+                table_id_map.emplacePartitionTableID(part_def.id, table_id);
+                continue;
             }
+            {
+                auto alter_lock = part_storage->lockForAlter(getThreadNameAndID());
+                auto local_table_info = part_storage->getTableInfo(); // copy
+                old_replica_count = local_table_info.replica_info.count;
+                new_replica_count = new_part_table_info->replica_info.count;
+                // Only update the replica info, do not change other fields. Or it may
+                // lead to other DDL is unexpectedly ignored.
+                local_table_info.replica_info = new_part_table_info->replica_info;
+                part_storage->alterSchemaChange(
+                    alter_lock,
+                    local_table_info,
+                    name_mapper.mapDatabaseName(db_info->id, keyspace_id),
+                    name_mapper.mapTableName(local_table_info),
+                    context);
+            }
+            LOG_INFO(
+                log,
+                "Updating replica info, replica count old={} new={} available={}"
+                " physical_table_id={} logical_table_id={}",
+                old_replica_count,
+                new_replica_count,
+                table_info->replica_info.available,
+                part_def.id,
+                table_id);
         }
     }
+
+    {
+        auto alter_lock = storage->lockForAlter(getThreadNameAndID());
+        old_replica_count = local_logical_storage_table_info.replica_info.count;
+        new_replica_count = table_info->replica_info.count;
+        // Only update the replica info, do not change other fields. Or it may
+        // lead to other DDL is unexpectedly ignored.
+        local_logical_storage_table_info.replica_info = table_info->replica_info;
+        storage->alterSchemaChange(
+            alter_lock,
+            local_logical_storage_table_info,
+            name_mapper.mapDatabaseName(db_info->id, keyspace_id),
+            name_mapper.mapTableName(local_logical_storage_table_info),
+            context);
+    }
+    LOG_INFO(
+        log,
+        "Updating replica info, replica count old={} new={} available={}"
+        " physical_table_id={} logical_table_id={}",
+        old_replica_count,
+        new_replica_count,
+        table_info->replica_info.available,
+        table_id,
+        table_id);
 }
 
 template <typename Getter, typename NameMapper>
 void SchemaBuilder<Getter, NameMapper>::applyPartitionDiff(DatabaseID database_id, TableID table_id)
 {
     auto [db_info, table_info] = getter.getDatabaseAndTableInfo(database_id, table_id);
-    if (table_info == nullptr)
+    if (unlikely(db_info == nullptr || table_info == nullptr))
     {
         LOG_ERROR(log, "table is not exist in TiKV, applyPartitionDiff is ignored, table_id={}", table_id);
         return;
@@ -468,11 +521,11 @@ void SchemaBuilder<Getter, NameMapper>::applyPartitionDiff(DatabaseID database_i
         return;
     }
 
-    applyPartitionDiff(db_info, table_info, storage);
+    applyPartitionDiffOnLogicalTable(db_info, table_info, storage);
 }
 
 template <typename Getter, typename NameMapper>
-void SchemaBuilder<Getter, NameMapper>::applyPartitionDiff(
+void SchemaBuilder<Getter, NameMapper>::applyPartitionDiffOnLogicalTable(
     const TiDB::DBInfoPtr & db_info,
     const TableInfoPtr & table_info,
     const ManageableStoragePtr & storage)
@@ -567,7 +620,7 @@ template <typename Getter, typename NameMapper>
 void SchemaBuilder<Getter, NameMapper>::applyRenameTable(DatabaseID database_id, TableID table_id)
 {
     auto [new_db_info, new_table_info] = getter.getDatabaseAndTableInfo(database_id, table_id);
-    if (new_table_info == nullptr)
+    if (unlikely(new_db_info == nullptr || new_table_info == nullptr))
     {
         LOG_ERROR(log, "table is not exist in TiKV, applyRenameTable is ignored, table_id={}", table_id);
         return;
@@ -679,7 +732,7 @@ template <typename Getter, typename NameMapper>
 void SchemaBuilder<Getter, NameMapper>::applyRecoverTable(DatabaseID database_id, TiDB::TableID table_id)
 {
     auto [db_info, table_info] = getter.getDatabaseAndTableInfo(database_id, table_id);
-    if (table_info == nullptr)
+    if (unlikely(db_info == nullptr || table_info == nullptr))
     {
         // this table is dropped.
         LOG_INFO(
@@ -689,23 +742,37 @@ void SchemaBuilder<Getter, NameMapper>::applyRecoverTable(DatabaseID database_id
         return;
     }
 
+    applyRecoverLogicalTable(db_info, table_info);
+}
+
+template <typename Getter, typename NameMapper>
+void SchemaBuilder<Getter, NameMapper>::applyRecoverLogicalTable(
+    const TiDB::DBInfoPtr & db_info,
+    const TiDB::TableInfoPtr & table_info)
+{
+    assert(db_info != nullptr);
+    assert(table_info != nullptr);
     if (table_info->isLogicalPartitionTable())
     {
         for (const auto & part_def : table_info->partition.definitions)
         {
-            auto new_table_info = table_info->producePartitionTableInfo(part_def.id, name_mapper);
-            applyRecoverPhysicalTable(db_info, new_table_info);
+            auto part_table_info = table_info->producePartitionTableInfo(part_def.id, name_mapper);
+            tryRecoverPhysicalTable(db_info, part_table_info);
         }
     }
 
-    applyRecoverPhysicalTable(db_info, table_info);
+    tryRecoverPhysicalTable(db_info, table_info);
 }
 
+// Return true - the Storage instance exists and is recovered (or not tombstone)
+//        false - the Storage instance does not exist
 template <typename Getter, typename NameMapper>
-void SchemaBuilder<Getter, NameMapper>::applyRecoverPhysicalTable(
+bool SchemaBuilder<Getter, NameMapper>::tryRecoverPhysicalTable(
     const TiDB::DBInfoPtr & db_info,
     const TiDB::TableInfoPtr & table_info)
 {
+    assert(db_info != nullptr);
+    assert(table_info != nullptr);
     auto & tmt_context = context.getTMTContext();
     auto storage = tmt_context.getStorages().get(keyspace_id, table_info->id);
     if (storage == nullptr)
@@ -714,48 +781,47 @@ void SchemaBuilder<Getter, NameMapper>::applyRecoverPhysicalTable(
             log,
             "Storage instance does not exist, tryRecoverPhysicalTable is ignored, table_id={}",
             table_info->id);
+        return false;
     }
-    else
-    {
-        if (!storage->isTombstone())
-        {
-            LOG_INFO(
-                log,
-                "Trying to recover table {} but it is not marked as tombstone, skip, database_id={} table_id={}",
-                name_mapper.debugCanonicalName(*db_info, *table_info),
-                db_info->id,
-                table_info->id);
-            return;
-        }
 
+    if (!storage->isTombstone())
+    {
         LOG_INFO(
             log,
-            "Create table {} by recover begin, database_id={} table_id={}",
-            name_mapper.debugCanonicalName(*db_info, *table_info),
-            db_info->id,
-            table_info->id);
-        AlterCommands commands;
-        {
-            AlterCommand command;
-            command.type = AlterCommand::RECOVER;
-            commands.emplace_back(std::move(command));
-        }
-        auto alter_lock = storage->lockForAlter(getThreadNameAndID());
-        storage->updateTombstone(
-            alter_lock,
-            commands,
-            name_mapper.mapDatabaseName(*db_info),
-            *table_info,
-            name_mapper,
-            context);
-        LOG_INFO(
-            log,
-            "Create table {} by recover end, database_id={} table_id={}",
+            "Trying to recover table {} but it is not marked as tombstone, skip, database_id={} table_id={}",
             name_mapper.debugCanonicalName(*db_info, *table_info),
             db_info->id,
             table_info->id);
-        return;
+        return true;
+    }
+
+    LOG_INFO(
+        log,
+        "Create table {} by recover begin, database_id={} table_id={}",
+        name_mapper.debugCanonicalName(*db_info, *table_info),
+        db_info->id,
+        table_info->id);
+    AlterCommands commands;
+    {
+        AlterCommand command;
+        command.type = AlterCommand::RECOVER;
+        commands.emplace_back(std::move(command));
     }
+    auto alter_lock = storage->lockForAlter(getThreadNameAndID());
+    storage->updateTombstone(
+        alter_lock,
+        commands,
+        name_mapper.mapDatabaseName(*db_info),
+        *table_info,
+        name_mapper,
+        context);
+    LOG_INFO(
+        log,
+        "Create table {} by recover end, database_id={} table_id={}",
+        name_mapper.debugCanonicalName(*db_info, *table_info),
+        db_info->id,
+        table_info->id);
+    return true;
 }
 
 static ASTPtr parseCreateStatement(const String & statement)
@@ -805,12 +871,12 @@ String createDatabaseStmt(Context & context, const DBInfo & db_info, const Schem
 template <typename Getter, typename NameMapper>
 bool SchemaBuilder<Getter, NameMapper>::applyCreateSchema(DatabaseID schema_id)
 {
-    auto db = getter.getDatabase(schema_id);
-    if (db == nullptr)
+    auto db_info = getter.getDatabase(schema_id);
+    if (unlikely(db_info == nullptr))
     {
         return false;
     }
-    applyCreateSchema(db);
+    applyCreateSchema(db_info);
     return true;
 }
 
@@ -916,6 +982,7 @@ String createTableStmt(
     const DBInfo & db_info,
     const TableInfo & table_info,
     const SchemaNameMapper & name_mapper,
+    const UInt64 tombstone,
     const LoggerPtr & log)
 {
     LOG_DEBUG(log, "Analyzing table info : {}", table_info.serialize());
@@ -949,13 +1016,14 @@ String createTableStmt(
         }
         writeString("), '", stmt_buf);
         writeEscapedString(table_info.serialize(), stmt_buf);
-        writeString("')", stmt_buf);
+        writeString(fmt::format("', {})", tombstone), stmt_buf);
     }
     else
     {
         throw TiFlashException(
-            fmt::format("Unknown engine type : {}", static_cast<int32_t>(table_info.engine_type)),
-            Errors::DDL::Internal);
+            Errors::DDL::Internal,
+            "Unknown engine type : {}",
+            fmt::underlying(table_info.engine_type));
     }
 
     return stmt;
@@ -964,8 +1032,12 @@ String createTableStmt(
 template <typename Getter, typename NameMapper>
 void SchemaBuilder<Getter, NameMapper>::applyCreateStorageInstance(
     const TiDB::DBInfoPtr & db_info,
-    const TableInfoPtr & table_info)
+    const TableInfoPtr & table_info,
+    bool is_tombstone)
 {
+    assert(db_info != nullptr);
+    assert(table_info != nullptr);
+
     GET_METRIC(tiflash_schema_internal_ddl_count, type_create_table).Increment();
     LOG_INFO(
         log,
@@ -974,53 +1046,12 @@ void SchemaBuilder<Getter, NameMapper>::applyCreateStorageInstance(
         db_info->id,
         table_info->id);
 
-    /// Check if this is a RECOVER table.
+    /// Try to recover the existing storage instance
+    if (tryRecoverPhysicalTable(db_info, table_info))
     {
-        auto & tmt_context = context.getTMTContext();
-        if (auto * storage = tmt_context.getStorages().get(keyspace_id, table_info->id).get(); storage)
-        {
-            if (!storage->isTombstone())
-            {
-                LOG_DEBUG(
-                    log,
-                    "Trying to create table {}, but it already exists and is not marked as tombstone, database_id={} "
-                    "table_id={}",
-                    name_mapper.debugCanonicalName(*db_info, *table_info),
-                    db_info->id,
-                    table_info->id);
-                return;
-            }
-
-            LOG_DEBUG(
-                log,
-                "Recovering table {} with database_id={}, table_id={}",
-                name_mapper.debugCanonicalName(*db_info, *table_info),
-                db_info->id,
-                table_info->id);
-            AlterCommands commands;
-            {
-                AlterCommand command;
-                command.type = AlterCommand::RECOVER;
-                commands.emplace_back(std::move(command));
-            }
-            auto alter_lock = storage->lockForAlter(getThreadNameAndID());
-            storage->updateTombstone(
-                alter_lock,
-                commands,
-                name_mapper.mapDatabaseName(*db_info),
-                *table_info,
-                name_mapper,
-                context);
-            LOG_INFO(
-                log,
-                "Created table {}, database_id={} table_id={}",
-                name_mapper.debugCanonicalName(*db_info, *table_info),
-                db_info->id,
-                table_info->id);
-            return;
-        }
+        return;
     }
-
+    // Else the storage instance does not exist, create it.
     /// Normal CREATE table.
     if (table_info->engine_type == StorageEngine::UNSPECIFIED)
     {
@@ -1028,7 +1059,13 @@ void SchemaBuilder<Getter, NameMapper>::applyCreateStorageInstance(
         table_info->engine_type = tmt_context.getEngineType();
     }
 
-    String stmt = createTableStmt(*db_info, *table_info, name_mapper, log);
+    UInt64 tombstone_ts = 0;
+    if (is_tombstone)
+    {
+        tombstone_ts = context.getTMTContext().getPDClient()->getTS();
+    }
+
+    String stmt = createTableStmt(*db_info, *table_info, name_mapper, tombstone_ts, log);
 
     LOG_INFO(
         log,
@@ -1148,9 +1185,7 @@ void SchemaBuilder<Getter, NameMapper>::syncAllSchema()
     LOG_INFO(log, "Sync all schemas begin");
 
     /// Create all databases.
-    std::vector<DBInfoPtr> all_schemas = getter.listDBs();
-
-    std::unordered_set<String> created_db_set;
+    std::vector<DBInfoPtr> all_db_info = getter.listDBs();
 
     //We can't use too large default_num_threads, otherwise, the lock grabbing time will be too much.
     size_t default_num_threads = std::max(4UL, std::thread::hardware_concurrency());
@@ -1159,65 +1194,74 @@ void SchemaBuilder<Getter, NameMapper>::syncAllSchema()
     auto sync_all_schema_wait_group = sync_all_schema_thread_pool.waitGroup();
 
     std::mutex created_db_set_mutex;
-    for (const auto & db : all_schemas)
+    std::unordered_set<String> created_db_set;
+    for (const auto & db_info : all_db_info)
     {
-        auto task = [this, db, &created_db_set, &created_db_set_mutex] {
+        auto task = [this, db_info, &created_db_set, &created_db_set_mutex] {
             do
             {
-                if (databases.exists(db->id))
+                if (databases.exists(db_info->id))
                 {
                     break;
                 }
-                applyCreateSchema(db);
+                applyCreateSchema(db_info);
                 {
                     std::unique_lock<std::mutex> created_db_set_lock(created_db_set_mutex);
-                    created_db_set.emplace(name_mapper.mapDatabaseName(*db));
+                    created_db_set.emplace(name_mapper.mapDatabaseName(*db_info));
                 }
 
                 LOG_INFO(
                     log,
                     "Database {} created during sync all schemas, database_id={}",
-                    name_mapper.debugDatabaseName(*db),
-                    db->id);
+                    name_mapper.debugDatabaseName(*db_info),
+                    db_info->id);
             } while (false); // Ensure database existing
 
-            std::vector<TableInfoPtr> tables = getter.listTables(db->id);
-            for (auto & table : tables)
+            std::vector<TableInfoPtr> tables = getter.listTables(db_info->id);
+            for (auto & table_info : tables)
             {
                 LOG_INFO(
                     log,
                     "Table {} syncing during sync all schemas, database_id={} table_id={}",
-                    name_mapper.debugCanonicalName(*db, *table),
-                    db->id,
-                    table->id);
+                    name_mapper.debugCanonicalName(*db_info, *table_info),
+                    db_info->id,
+                    table_info->id);
 
                 /// Ignore view and sequence.
-                if (table->is_view || table->is_sequence)
+                if (table_info->is_view || table_info->is_sequence)
                 {
                     LOG_INFO(
                         log,
                         "Table {} is a view or sequence, ignoring. database_id={} table_id={}",
-                        name_mapper.debugCanonicalName(*db, *table),
-                        db->id,
-                        table->id);
+                        name_mapper.debugCanonicalName(*db_info, *table_info),
+                        db_info->id,
+                        table_info->id);
                     continue;
                 }
 
-                table_id_map.emplaceTableID(table->id, db->id);
-                LOG_DEBUG(log, "register table to table_id_map, database_id={} table_id={}", db->id, table->id);
+                table_id_map.emplaceTableID(table_info->id, db_info->id);
+                LOG_DEBUG(
+                    log,
+                    "register table to table_id_map, database_id={} table_id={}",
+                    db_info->id,
+                    table_info->id);
 
-                applyCreateStorageInstance(db, table);
-                if (table->isLogicalPartitionTable())
+                // `SchemaGetter::listTables` only return non-tombstone tables.
+                // So `syncAllSchema` will not create tombstone tables. But if there are new rows/new snapshot
+                // sent to TiFlash, TiFlash can create the instance by `applyTable` with force==true in the
+                // related process.
+                applyCreateStorageInstance(db_info, table_info, false);
+                if (table_info->isLogicalPartitionTable())
                 {
-                    for (const auto & part_def : table->partition.definitions)
+                    for (const auto & part_def : table_info->partition.definitions)
                     {
                         LOG_DEBUG(
                             log,
                             "register table to table_id_map for partition table, logical_table_id={} "
                             "physical_table_id={}",
-                            table->id,
+                            table_info->id,
                             part_def.id);
-                        table_id_map.emplacePartitionTableID(part_def.id, table->id);
+                        table_id_map.emplacePartitionTableID(part_def.id, table_info->id);
                     }
                 }
             }
@@ -1247,7 +1291,7 @@ void SchemaBuilder<Getter, NameMapper>::syncAllSchema()
         }
     }
 
-    /// Drop all unmapped dbs.
+    /// Drop all unmapped databases
     const auto & dbs = context.getDatabases();
     for (auto it = dbs.begin(); it != dbs.end(); it++)
     {
@@ -1266,21 +1310,53 @@ void SchemaBuilder<Getter, NameMapper>::syncAllSchema()
     LOG_INFO(log, "Sync all schemas end");
 }
 
+/**
+ * Update the schema of given `physical_table_id`.
+ * This function ensure only the lock of `physical_table_id` is involved.
+ *
+ * Param `database_id`, `logical_table_id` is to key to fetch the latest table info. If
+ * something wrong when generating the table info of `physical_table_id`, it means the
+ * TableID mapping is not up-to-date. This function will return false and the caller
+ * should update the TableID mapping then retry.
+ * If the caller ensure the TableID mapping is up-to-date, then it should call with
+ * `force == true`
+ */
 template <typename Getter, typename NameMapper>
 bool SchemaBuilder<Getter, NameMapper>::applyTable(
     DatabaseID database_id,
     TableID logical_table_id,
-    TableID physical_table_id)
+    TableID physical_table_id,
+    bool force)
 {
-    // Here we get table info without mvcc. If the table has been renamed to another
-    // database, it will return false and the caller should update the table_id_map
-    // then retry.
-    auto table_info = getter.getTableInfo(database_id, logical_table_id, /*try_mvcc*/ false);
+    // When `force==false`, we get table info without mvcc. So we can detect that whether
+    // the table has been renamed to another database or dropped.
+    // If the table has been renamed to another database, it is dangerous to use the
+    // old table info from the old database because some new columns may have been
+    // added to the new table.
+    // For the reason above, if we can not get table info without mvcc, this function
+    // will return false and the caller should update the table_id_map then retry.
+    //
+    // When `force==true`, the caller ensure the TableID mapping is up-to-date, so we
+    // need to get table info with mvcc. It can return the table info even if a table is
+    // dropped but not physically removed by TiDB/TiKV gc_safepoint.
+    // It is need for TiFlash correctly decoding the data and get ready for `RECOVER TABLE`
+    // and `RECOVER DATABASE`.
+    TableInfoPtr table_info;
+    bool get_by_mvcc = false;
+    if (!force)
+    {
+        table_info = getter.getTableInfo(database_id, logical_table_id, /*try_mvcc*/ false);
+    }
+    else
+    {
+        std::tie(table_info, get_by_mvcc) = getter.getTableInfoAndCheckMvcc(database_id, logical_table_id);
+    }
     if (table_info == nullptr)
     {
         LOG_WARNING(
             log,
-            "table is not exist in TiKV, applyTable need retry, database_id={} logical_table_id={}",
+            "table is not exist in TiKV, applyTable need retry, get_by_mvcc={} database_id={} logical_table_id={}",
+            get_by_mvcc,
             database_id,
             logical_table_id);
         return false;
@@ -1334,7 +1410,8 @@ bool SchemaBuilder<Getter, NameMapper>::applyTable(
         }
 
         // Create the instance with the latest table info
-        applyCreateStorageInstance(db_info, table_info);
+        // If the table info is get by mvcc, it means the table is actually in "dropped" status
+        applyCreateStorageInstance(db_info, table_info, get_by_mvcc);
         return true;
     }
 
diff --git a/dbms/src/TiDB/Schema/SchemaBuilder.h b/dbms/src/TiDB/Schema/SchemaBuilder.h
index e963d36f4a0..53207d8f320 100644
--- a/dbms/src/TiDB/Schema/SchemaBuilder.h
+++ b/dbms/src/TiDB/Schema/SchemaBuilder.h
@@ -55,9 +55,13 @@ struct SchemaBuilder
 
     void syncAllSchema();
 
+    /**
+      * Drop all schema of a given keyspace.
+      * When a keyspace is removed, drop all its databases and tables.
+      */
     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,19 +73,22 @@ 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);
 
     void applyPartitionDiff(DatabaseID database_id, TableID table_id);
-    void applyPartitionDiff(
+    void applyPartitionDiffOnLogicalTable(
         const TiDB::DBInfoPtr & db_info,
         const TiDB::TableInfoPtr & table_info,
         const ManageableStoragePtr & storage);
diff --git a/dbms/src/TiDB/Schema/SchemaGetter.cpp b/dbms/src/TiDB/Schema/SchemaGetter.cpp
index 7b843e83f81..1197fa3f043 100644
--- a/dbms/src/TiDB/Schema/SchemaGetter.cpp
+++ b/dbms/src/TiDB/Schema/SchemaGetter.cpp
@@ -283,25 +283,24 @@ 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(
@@ -309,14 +308,14 @@ TiDB::TableInfoPtr SchemaGetter::getTableInfoImpl(DatabaseID db_id, TableID tabl
                 "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,
diff --git a/dbms/src/TiDB/Schema/SchemaGetter.h b/dbms/src/TiDB/Schema/SchemaGetter.h
index 40270302378..fe5c1efe803 100644
--- a/dbms/src/TiDB/Schema/SchemaGetter.h
+++ b/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
diff --git a/dbms/src/TiDB/Schema/SchemaSyncer.h b/dbms/src/TiDB/Schema/SchemaSyncer.h
index 5eccfe7428d..135048b745b 100644
--- a/dbms/src/TiDB/Schema/SchemaSyncer.h
+++ b/dbms/src/TiDB/Schema/SchemaSyncer.h
@@ -54,6 +54,10 @@ class SchemaSyncer
 
     virtual void removeTableID(TableID table_id) = 0;
 
+    /**
+      * Drop all schema of a given keyspace.
+      * When a keyspace is removed, drop all its databases and tables.
+      */
     virtual void dropAllSchema(Context & context) = 0;
 };
 
diff --git a/dbms/src/TiDB/Schema/TiDBSchemaSyncer.cpp b/dbms/src/TiDB/Schema/TiDBSchemaSyncer.cpp
index 55ca5a4fdac..f54b50d1676 100644
--- a/dbms/src/TiDB/Schema/TiDBSchemaSyncer.cpp
+++ b/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);
diff --git a/dbms/src/TiDB/Schema/TiDBSchemaSyncer.h b/dbms/src/TiDB/Schema/TiDBSchemaSyncer.h
index 379822f611b..370a0d4e665 100644
--- a/dbms/src/TiDB/Schema/TiDBSchemaSyncer.h
+++ b/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
@@ -112,6 +113,10 @@ class TiDBSchemaSyncer : public SchemaSyncer
         return databases.getDBInfoByName(database_name);
     }
 
+    /**
+      * Drop all schema of a given keyspace.
+      * When a keyspace is removed, drop all its databases and tables.
+      */
     void dropAllSchema(Context & context) override;
 
     // clear all states.
diff --git a/dbms/src/TiDB/tests/gtest_table_info.cpp b/dbms/src/TiDB/tests/gtest_table_info.cpp
index 9da7a9318b1..7d75cfc9d34 100644
--- a/dbms/src/TiDB/tests/gtest_table_info.cpp
+++ b/dbms/src/TiDB/tests/gtest_table_info.cpp
@@ -36,6 +36,7 @@ String createTableStmt(
     const DBInfo & db_info,
     const TableInfo & table_info,
     const SchemaNameMapper & name_mapper,
+    const UInt64 tombstone,
     const LoggerPtr & log);
 
 namespace tests
@@ -131,6 +132,7 @@ CATCH
 struct StmtCase
 {
     TableID table_or_partition_id;
+    UInt64 tombstone;
     String db_info_json;
     String table_info_json;
     String create_stmt_dm;
@@ -149,7 +151,7 @@ struct StmtCase
         // generate create statement with db_info and table_info
         auto verify_stmt = [&](TiDB::StorageEngine engine_type) {
             table_info.engine_type = engine_type;
-            String stmt = createTableStmt(db_info, table_info, MockSchemaNameMapper(), Logger::get());
+            String stmt = createTableStmt(db_info, table_info, MockSchemaNameMapper(), tombstone, Logger::get());
             EXPECT_EQ(stmt, create_stmt_dm) << "Table info create statement mismatch:\n" + stmt + "\n" + create_stmt_dm;
 
             json1 = extractTableInfoFromCreateStatement(stmt, table_info.name);
@@ -168,7 +170,7 @@ struct StmtCase
         ASTPtr ast = parseQuery(parser, stmt.data(), stmt.data() + stmt.size(), "from verifyTableInfo " + tbl_name, 0);
         ASTCreateQuery & ast_create_query = typeid_cast<ASTCreateQuery &>(*ast);
         auto & ast_arguments = typeid_cast<ASTExpressionList &>(*(ast_create_query.storage->engine->arguments));
-        ASTLiteral & ast_literal = typeid_cast<ASTLiteral &>(*(ast_arguments.children.back()));
+        ASTLiteral & ast_literal = typeid_cast<ASTLiteral &>(*(ast_arguments.children[1]));
         return safeGet<String>(ast_literal.value);
     }
 };
@@ -179,45 +181,59 @@ try
     auto cases = {
         StmtCase{
             1145, //
+            0,
             R"json({"id":1939,"db_name":{"O":"customer","L":"customer"},"charset":"utf8mb4","collate":"utf8mb4_bin","state":5})json", //
             R"json({"id":1145,"name":{"O":"customerdebt","L":"customerdebt"},"cols":[{"id":1,"name":{"O":"id","L":"id"},"offset":0,"origin_default":null,"default":null,"default_bit":null,"type":{"Tp":8,"Flag":515,"Flen":20,"Decimal":0},"state":5,"comment":"i\"d"}],"state":5,"pk_is_handle":true,"schema_version":-1,"comment":"负债信息","partition":null})json", //
-            R"stmt(CREATE TABLE `db_1939`.`t_1145`(`id` Int64) Engine = DeltaMerge((`id`), '{"cols":[{"comment":"i\\"d","default":null,"default_bit":null,"id":1,"name":{"L":"id","O":"id"},"offset":0,"origin_default":null,"state":5,"type":{"Charset":null,"Collate":null,"Decimal":0,"Elems":null,"Flag":515,"Flen":20,"Tp":8}}],"comment":"\\u8D1F\\u503A\\u4FE1\\u606F","id":1145,"index_info":[],"is_common_handle":false,"keyspace_id":4294967295,"name":{"L":"customerdebt","O":"customerdebt"},"partition":null,"pk_is_handle":true,"schema_version":-1,"state":5,"tiflash_replica":{"Count":0},"update_timestamp":0}'))stmt", //
+            R"stmt(CREATE TABLE `db_1939`.`t_1145`(`id` Int64) Engine = DeltaMerge((`id`), '{"cols":[{"comment":"i\\"d","default":null,"default_bit":null,"id":1,"name":{"L":"id","O":"id"},"offset":0,"origin_default":null,"state":5,"type":{"Charset":null,"Collate":null,"Decimal":0,"Elems":null,"Flag":515,"Flen":20,"Tp":8}}],"comment":"\\u8D1F\\u503A\\u4FE1\\u606F","id":1145,"index_info":[],"is_common_handle":false,"keyspace_id":4294967295,"name":{"L":"customerdebt","O":"customerdebt"},"partition":null,"pk_is_handle":true,"schema_version":-1,"state":5,"tiflash_replica":{"Count":0},"update_timestamp":0}', 0))stmt", //
         },
         StmtCase{
             2049, //
+            0,
             R"json({"id":1939,"db_name":{"O":"customer","L":"customer"},"charset":"utf8mb4","collate":"utf8mb4_bin","state":5})json", //
             R"json({"id":2049,"name":{"O":"customerdebt","L":"customerdebt"},"cols":[{"id":1,"name":{"O":"id","L":"id"},"offset":0,"origin_default":null,"default":null,"default_bit":null,"type":{"Tp":8,"Flag":515,"Flen":20,"Decimal":0},"state":5,"comment":"i\"d"}],"state":5,"pk_is_handle":true,"schema_version":-1,"comment":"负债信息","update_timestamp":404545295996944390,"partition":null})json", //
-            R"stmt(CREATE TABLE `db_1939`.`t_2049`(`id` Int64) Engine = DeltaMerge((`id`), '{"cols":[{"comment":"i\\"d","default":null,"default_bit":null,"id":1,"name":{"L":"id","O":"id"},"offset":0,"origin_default":null,"state":5,"type":{"Charset":null,"Collate":null,"Decimal":0,"Elems":null,"Flag":515,"Flen":20,"Tp":8}}],"comment":"\\u8D1F\\u503A\\u4FE1\\u606F","id":2049,"index_info":[],"is_common_handle":false,"keyspace_id":4294967295,"name":{"L":"customerdebt","O":"customerdebt"},"partition":null,"pk_is_handle":true,"schema_version":-1,"state":5,"tiflash_replica":{"Count":0},"update_timestamp":404545295996944390}'))stmt", //
+            R"stmt(CREATE TABLE `db_1939`.`t_2049`(`id` Int64) Engine = DeltaMerge((`id`), '{"cols":[{"comment":"i\\"d","default":null,"default_bit":null,"id":1,"name":{"L":"id","O":"id"},"offset":0,"origin_default":null,"state":5,"type":{"Charset":null,"Collate":null,"Decimal":0,"Elems":null,"Flag":515,"Flen":20,"Tp":8}}],"comment":"\\u8D1F\\u503A\\u4FE1\\u606F","id":2049,"index_info":[],"is_common_handle":false,"keyspace_id":4294967295,"name":{"L":"customerdebt","O":"customerdebt"},"partition":null,"pk_is_handle":true,"schema_version":-1,"state":5,"tiflash_replica":{"Count":0},"update_timestamp":404545295996944390}', 0))stmt", //
         },
         StmtCase{
             31, //
+            0,
             R"json({"id":1,"db_name":{"O":"db1","L":"db1"},"charset":"utf8mb4","collate":"utf8mb4_bin","state":5})json", //
             R"json({"id":31,"name":{"O":"simple_t","L":"simple_t"},"charset":"","collate":"","cols":[{"id":1,"name":{"O":"i","L":"i"},"offset":0,"origin_default":null,"default":null,"default_bit":null,"generated_expr_string":"","generated_stored":false,"dependences":null,"type":{"Tp":3,"Flag":0,"Flen":11,"Decimal":0,"Charset":"binary","Collate":"binary","Elems":null},"state":5,"comment":""}],"index_info":null,"fk_info":null,"state":5,"pk_is_handle":false,"schema_version":-1,"comment":"","auto_inc_id":0,"max_col_id":1,"max_idx_id":0,"update_timestamp":404545295996944390,"ShardRowIDBits":0,"partition":null})json", //
-            R"stmt(CREATE TABLE `db_1`.`t_31`(`i` Nullable(Int32), `_tidb_rowid` Int64) Engine = DeltaMerge((`_tidb_rowid`), '{"cols":[{"comment":"","default":null,"default_bit":null,"id":1,"name":{"L":"i","O":"i"},"offset":0,"origin_default":null,"state":5,"type":{"Charset":"binary","Collate":"binary","Decimal":0,"Elems":null,"Flag":0,"Flen":11,"Tp":3}}],"comment":"","id":31,"index_info":[],"is_common_handle":false,"keyspace_id":4294967295,"name":{"L":"simple_t","O":"simple_t"},"partition":null,"pk_is_handle":false,"schema_version":-1,"state":5,"tiflash_replica":{"Count":0},"update_timestamp":404545295996944390}'))stmt", //
+            R"stmt(CREATE TABLE `db_1`.`t_31`(`i` Nullable(Int32), `_tidb_rowid` Int64) Engine = DeltaMerge((`_tidb_rowid`), '{"cols":[{"comment":"","default":null,"default_bit":null,"id":1,"name":{"L":"i","O":"i"},"offset":0,"origin_default":null,"state":5,"type":{"Charset":"binary","Collate":"binary","Decimal":0,"Elems":null,"Flag":0,"Flen":11,"Tp":3}}],"comment":"","id":31,"index_info":[],"is_common_handle":false,"keyspace_id":4294967295,"name":{"L":"simple_t","O":"simple_t"},"partition":null,"pk_is_handle":false,"schema_version":-1,"state":5,"tiflash_replica":{"Count":0},"update_timestamp":404545295996944390}', 0))stmt", //
         },
         StmtCase{
             33, //
+            0,
             R"json({"id":2,"db_name":{"O":"db2","L":"db2"},"charset":"utf8mb4","collate":"utf8mb4_bin","state":5})json", //
             R"json({"id":33,"name":{"O":"pk_t","L":"pk_t"},"charset":"","collate":"","cols":[{"id":1,"name":{"O":"i","L":"i"},"offset":0,"origin_default":null,"default":null,"default_bit":null,"generated_expr_string":"","generated_stored":false,"dependences":null,"type":{"Tp":3,"Flag":3,"Flen":11,"Decimal":0,"Charset":"binary","Collate":"binary","Elems":null},"state":5,"comment":""}],"index_info":null,"fk_info":null,"state":5,"pk_is_handle":true,"schema_version":-1,"comment":"","auto_inc_id":0,"max_col_id":1,"max_idx_id":0,"update_timestamp":404545312978108418,"ShardRowIDBits":0,"partition":null})json", //
-            R"stmt(CREATE TABLE `db_2`.`t_33`(`i` Int32) Engine = DeltaMerge((`i`), '{"cols":[{"comment":"","default":null,"default_bit":null,"id":1,"name":{"L":"i","O":"i"},"offset":0,"origin_default":null,"state":5,"type":{"Charset":"binary","Collate":"binary","Decimal":0,"Elems":null,"Flag":3,"Flen":11,"Tp":3}}],"comment":"","id":33,"index_info":[],"is_common_handle":false,"keyspace_id":4294967295,"name":{"L":"pk_t","O":"pk_t"},"partition":null,"pk_is_handle":true,"schema_version":-1,"state":5,"tiflash_replica":{"Count":0},"update_timestamp":404545312978108418}'))stmt", //
+            R"stmt(CREATE TABLE `db_2`.`t_33`(`i` Int32) Engine = DeltaMerge((`i`), '{"cols":[{"comment":"","default":null,"default_bit":null,"id":1,"name":{"L":"i","O":"i"},"offset":0,"origin_default":null,"state":5,"type":{"Charset":"binary","Collate":"binary","Decimal":0,"Elems":null,"Flag":3,"Flen":11,"Tp":3}}],"comment":"","id":33,"index_info":[],"is_common_handle":false,"keyspace_id":4294967295,"name":{"L":"pk_t","O":"pk_t"},"partition":null,"pk_is_handle":true,"schema_version":-1,"state":5,"tiflash_replica":{"Count":0},"update_timestamp":404545312978108418}', 0))stmt", //
         },
         StmtCase{
             35, //
+            0,
             R"json({"id":1,"db_name":{"O":"db1","L":"db1"},"charset":"utf8mb4","collate":"utf8mb4_bin","state":5})json", //
             R"json({"id":35,"name":{"O":"not_null_t","L":"not_null_t"},"charset":"","collate":"","cols":[{"id":1,"name":{"O":"i","L":"i"},"offset":0,"origin_default":null,"default":null,"default_bit":null,"generated_expr_string":"","generated_stored":false,"dependences":null,"type":{"Tp":3,"Flag":4097,"Flen":11,"Decimal":0,"Charset":"binary","Collate":"binary","Elems":null},"state":5,"comment":""}],"index_info":null,"fk_info":null,"state":5,"pk_is_handle":false,"schema_version":-1,"comment":"","auto_inc_id":0,"max_col_id":1,"max_idx_id":0,"update_timestamp":404545324922961926,"ShardRowIDBits":0,"partition":null})json", //
-            R"stmt(CREATE TABLE `db_1`.`t_35`(`i` Int32, `_tidb_rowid` Int64) Engine = DeltaMerge((`_tidb_rowid`), '{"cols":[{"comment":"","default":null,"default_bit":null,"id":1,"name":{"L":"i","O":"i"},"offset":0,"origin_default":null,"state":5,"type":{"Charset":"binary","Collate":"binary","Decimal":0,"Elems":null,"Flag":4097,"Flen":11,"Tp":3}}],"comment":"","id":35,"index_info":[],"is_common_handle":false,"keyspace_id":4294967295,"name":{"L":"not_null_t","O":"not_null_t"},"partition":null,"pk_is_handle":false,"schema_version":-1,"state":5,"tiflash_replica":{"Count":0},"update_timestamp":404545324922961926}'))stmt", //
+            R"stmt(CREATE TABLE `db_1`.`t_35`(`i` Int32, `_tidb_rowid` Int64) Engine = DeltaMerge((`_tidb_rowid`), '{"cols":[{"comment":"","default":null,"default_bit":null,"id":1,"name":{"L":"i","O":"i"},"offset":0,"origin_default":null,"state":5,"type":{"Charset":"binary","Collate":"binary","Decimal":0,"Elems":null,"Flag":4097,"Flen":11,"Tp":3}}],"comment":"","id":35,"index_info":[],"is_common_handle":false,"keyspace_id":4294967295,"name":{"L":"not_null_t","O":"not_null_t"},"partition":null,"pk_is_handle":false,"schema_version":-1,"state":5,"tiflash_replica":{"Count":0},"update_timestamp":404545324922961926}', 0))stmt", //
         },
         StmtCase{
             37, //
+            0,
             R"json({"id":2,"db_name":{"O":"db2","L":"db2"},"charset":"utf8mb4","collate":"utf8mb4_bin","state":5})json",
             R"json({"id":37,"name":{"O":"mytable","L":"mytable"},"charset":"","collate":"","cols":[{"id":1,"name":{"O":"mycol","L":"mycol"},"offset":0,"origin_default":null,"default":null,"default_bit":null,"generated_expr_string":"","generated_stored":false,"dependences":null,"type":{"Tp":15,"Flag":4099,"Flen":256,"Decimal":0,"Charset":"utf8","Collate":"utf8_bin","Elems":null},"state":5,"comment":""}],"index_info":[{"id":1,"idx_name":{"O":"PRIMARY","L":"primary"},"tbl_name":{"O":"","L":""},"idx_cols":[{"name":{"O":"mycol","L":"mycol"},"offset":0,"length":-1}],"is_unique":true,"is_primary":true,"state":5,"comment":"","index_type":1}],"fk_info":null,"state":5,"pk_is_handle":true,"schema_version":-1,"comment":"","auto_inc_id":0,"max_col_id":1,"max_idx_id":1,"update_timestamp":404566455285710853,"ShardRowIDBits":0,"partition":null})json", //
-            R"stmt(CREATE TABLE `db_2`.`t_37`(`mycol` String) Engine = DeltaMerge((`mycol`), '{"cols":[{"comment":"","default":null,"default_bit":null,"id":1,"name":{"L":"mycol","O":"mycol"},"offset":0,"origin_default":null,"state":5,"type":{"Charset":"utf8","Collate":"utf8_bin","Decimal":0,"Elems":null,"Flag":4099,"Flen":256,"Tp":15}}],"comment":"","id":37,"index_info":[],"is_common_handle":false,"keyspace_id":4294967295,"name":{"L":"mytable","O":"mytable"},"partition":null,"pk_is_handle":true,"schema_version":-1,"state":5,"tiflash_replica":{"Count":0},"update_timestamp":404566455285710853}'))stmt", //
+            R"stmt(CREATE TABLE `db_2`.`t_37`(`mycol` String) Engine = DeltaMerge((`mycol`), '{"cols":[{"comment":"","default":null,"default_bit":null,"id":1,"name":{"L":"mycol","O":"mycol"},"offset":0,"origin_default":null,"state":5,"type":{"Charset":"utf8","Collate":"utf8_bin","Decimal":0,"Elems":null,"Flag":4099,"Flen":256,"Tp":15}}],"comment":"","id":37,"index_info":[],"is_common_handle":false,"keyspace_id":4294967295,"name":{"L":"mytable","O":"mytable"},"partition":null,"pk_is_handle":true,"schema_version":-1,"state":5,"tiflash_replica":{"Count":0},"update_timestamp":404566455285710853}', 0))stmt", //
         },
         StmtCase{
             32, //
+            0,
             R"json({"id":1,"db_name":{"O":"test","L":"test"},"charset":"utf8mb4","collate":"utf8mb4_bin","state":5})json", //
             R"json({"id":31,"name":{"O":"range_part_t","L":"range_part_t"},"charset":"utf8mb4","collate":"utf8mb4_bin","cols":[{"id":1,"name":{"O":"i","L":"i"},"offset":0,"origin_default":null,"default":null,"default_bit":null,"generated_expr_string":"","generated_stored":false,"dependences":null,"type":{"Tp":3,"Flag":0,"Flen":11,"Decimal":0,"Charset":"binary","Collate":"binary","Elems":null},"state":5,"comment":"","version":0}],"index_info":null,"fk_info":null,"state":5,"pk_is_handle":false,"schema_version":-1,"comment":"","auto_inc_id":0,"max_col_id":1,"max_idx_id":0,"update_timestamp":407445773801488390,"ShardRowIDBits":0,"partition":{"type":1,"expr":"`i`","columns":null,"enable":true,"definitions":[{"id":32,"name":{"O":"p0","L":"p0"},"less_than":["0"]},{"id":33,"name":{"O":"p1","L":"p1"},"less_than":["100"]}],"num":0},"compression":"","version":1})json", //
-            R"stmt(CREATE TABLE `db_1`.`t_32`(`i` Nullable(Int32), `_tidb_rowid` Int64) Engine = DeltaMerge((`_tidb_rowid`), '{"belonging_table_id":31,"cols":[{"comment":"","default":null,"default_bit":null,"id":1,"name":{"L":"i","O":"i"},"offset":0,"origin_default":null,"state":5,"type":{"Charset":"binary","Collate":"binary","Decimal":0,"Elems":null,"Flag":0,"Flen":11,"Tp":3}}],"comment":"","id":32,"index_info":[],"is_common_handle":false,"is_partition_sub_table":true,"keyspace_id":4294967295,"name":{"L":"range_part_t_32","O":"range_part_t_32"},"partition":null,"pk_is_handle":false,"schema_version":-1,"state":5,"tiflash_replica":{"Count":0},"update_timestamp":407445773801488390}'))stmt", //
+            R"stmt(CREATE TABLE `db_1`.`t_32`(`i` Nullable(Int32), `_tidb_rowid` Int64) Engine = DeltaMerge((`_tidb_rowid`), '{"belonging_table_id":31,"cols":[{"comment":"","default":null,"default_bit":null,"id":1,"name":{"L":"i","O":"i"},"offset":0,"origin_default":null,"state":5,"type":{"Charset":"binary","Collate":"binary","Decimal":0,"Elems":null,"Flag":0,"Flen":11,"Tp":3}}],"comment":"","id":32,"index_info":[],"is_common_handle":false,"is_partition_sub_table":true,"keyspace_id":4294967295,"name":{"L":"range_part_t_32","O":"range_part_t_32"},"partition":null,"pk_is_handle":false,"schema_version":-1,"state":5,"tiflash_replica":{"Count":0},"update_timestamp":407445773801488390}', 0))stmt", //
+        },
+        StmtCase{
+            32, //
+            1700815239,
+            R"json({"id":1,"db_name":{"O":"test","L":"test"},"charset":"utf8mb4","collate":"utf8mb4_bin","state":5})json", //
+            R"json({"id":31,"name":{"O":"range_part_t","L":"range_part_t"},"charset":"utf8mb4","collate":"utf8mb4_bin","cols":[{"id":1,"name":{"O":"i","L":"i"},"offset":0,"origin_default":null,"default":null,"default_bit":null,"generated_expr_string":"","generated_stored":false,"dependences":null,"type":{"Tp":3,"Flag":0,"Flen":11,"Decimal":0,"Charset":"binary","Collate":"binary","Elems":null},"state":5,"comment":"","version":0}],"index_info":null,"fk_info":null,"state":5,"pk_is_handle":false,"schema_version":-1,"comment":"","auto_inc_id":0,"max_col_id":1,"max_idx_id":0,"update_timestamp":407445773801488390,"ShardRowIDBits":0,"partition":{"type":1,"expr":"`i`","columns":null,"enable":true,"definitions":[{"id":32,"name":{"O":"p0","L":"p0"},"less_than":["0"]},{"id":33,"name":{"O":"p1","L":"p1"},"less_than":["100"]}],"num":0},"compression":"","version":1})json", //
+            R"stmt(CREATE TABLE `db_1`.`t_32`(`i` Nullable(Int32), `_tidb_rowid` Int64) Engine = DeltaMerge((`_tidb_rowid`), '{"belonging_table_id":31,"cols":[{"comment":"","default":null,"default_bit":null,"id":1,"name":{"L":"i","O":"i"},"offset":0,"origin_default":null,"state":5,"type":{"Charset":"binary","Collate":"binary","Decimal":0,"Elems":null,"Flag":0,"Flen":11,"Tp":3}}],"comment":"","id":32,"index_info":[],"is_common_handle":false,"is_partition_sub_table":true,"keyspace_id":4294967295,"name":{"L":"range_part_t_32","O":"range_part_t_32"},"partition":null,"pk_is_handle":false,"schema_version":-1,"state":5,"tiflash_replica":{"Count":0},"update_timestamp":407445773801488390}', 1700815239))stmt", //
         },
     };
 
diff --git a/tests/fullstack-test/fault-inject/create-database.test b/tests/fullstack-test2/ddl/alter_create_database_crash.test
similarity index 94%
rename from tests/fullstack-test/fault-inject/create-database.test
rename to tests/fullstack-test2/ddl/alter_create_database_crash.test
index fbfd89ae7dd..eda7b3def66 100644
--- a/tests/fullstack-test/fault-inject/create-database.test
+++ b/tests/fullstack-test2/ddl/alter_create_database_crash.test
@@ -28,8 +28,7 @@ mysql> alter table db_test.t set tiflash replica 1 location labels 'rack', 'host
 
 func> wait_table db_test t
 
-mysql> insert into db_test.t values (1, 1)
-mysql> insert into db_test.t values (1, 2)
+mysql> insert into db_test.t values (1, 1), (1, 2);
 
 mysql> set session tidb_isolation_read_engines='tiflash'; select * from db_test.t;
 +---+---+
diff --git a/tests/fullstack-test2/ddl/alter_drop_database.test b/tests/fullstack-test2/ddl/alter_drop_database.test
index e4b283bce43..838d46afbcf 100644
--- a/tests/fullstack-test2/ddl/alter_drop_database.test
+++ b/tests/fullstack-test2/ddl/alter_drop_database.test
@@ -45,5 +45,7 @@ mysql> drop database d1;
 # make write cmd take effect
 >> DBGInvoke __disable_fail_point(pause_before_apply_raft_cmd)
 
+# the `t1` is still mark as tombstone
 >> select tidb_database,tidb_name from system.tables where is_tombstone = 0 and tidb_database = 'd1' and tidb_name='t1';
+#TODO: check the row is written to the storage or not
 
diff --git a/tests/fullstack-test2/ddl/alter_drop_table.test b/tests/fullstack-test2/ddl/alter_drop_table.test
index 93db6be2e49..5fd6329f000 100644
--- a/tests/fullstack-test2/ddl/alter_drop_table.test
+++ b/tests/fullstack-test2/ddl/alter_drop_table.test
@@ -17,16 +17,31 @@
 #    if we drop the table without tiflash storage, it works well 
 #    if we drop the table with tiflash storage, it works well, and check the tombstone in TiFlash
 
+# Clean the tombstone table in the testing env
+>> DBGInvoke __enable_schema_sync_service('true')
+>> DBGInvoke __gc_schemas(18446744073709551615)
+
+## create table and drop without tiflash replica
 mysql> drop table if exists test.t1;
 mysql> create table test.t1(a int primary key, b decimal(5,2) not NULL, c varchar(10), d int default 0);
 mysql> insert into test.t1 values(1, 1.1, 'a', 1);
 mysql> drop table test.t1;
 
+## create empty table -> add tiflash replica -> wait tiflash sync the table schema -> drop table
 mysql> drop table if exists test.t2;
 mysql> create table test.t2(a int primary key, b decimal(5,2) not NULL, c varchar(10), d int default 0);
 mysql> alter table test.t2 set tiflash replica 1; 
+func> wait_table test t2
+mysql> set session tidb_isolation_read_engines='tiflash';select * from test.t2;
+>> select tidb_database,tidb_name from system.tables where tidb_database = 'test' and tidb_name = 't2' and is_tombstone = 0
+┌─tidb_database─┬─tidb_name─┐
+│ test          │ t2        │
+└───────────────┴───────────┘
 mysql> drop table test.t2;
+=> DBGInvoke __refresh_schemas()
+>> select tidb_database,tidb_name,tidb_table_id from system.tables where tidb_database = 'test' and tidb_name = 't2' and is_tombstone = 0
 
+## create table -> add tiflash replica -> drop table
 mysql> drop table if exists test.t3;
 mysql> create table test.t3(a int primary key, b decimal(5,2) not NULL, c varchar(10), d int default 0);
 mysql> alter table test.t3 set tiflash replica 1; 
@@ -44,66 +59,29 @@ mysql> set session tidb_isolation_read_engines='tiflash';select * from test.t3;
 ┌─tidb_database─┬─tidb_name─┐
 │ test          │ t3        │
 └───────────────┴───────────┘
-
 mysql> drop table test.t3;
-
 => DBGInvoke __refresh_schemas()
-
 >> select tidb_database,tidb_name from system.tables where tidb_database = 'test' and tidb_name = 't3' and is_tombstone = 0
 
-
-## drop table arrive tiflash before ddl and insert, and do recover, check the data is not lost
-## because we want to test we actually drop the table, so please not use the same name for this table
-mysql> drop table if exists test.t_drop; 
-mysql> create table test.t_drop(a int, b int);
-mysql> alter table test.t_drop set tiflash replica 1;
-mysql> insert into test.t_drop values(1, 1);
-
-func> wait_table test t_drop
-
-=> DBGInvoke __enable_schema_sync_service('false') 
-=> DBGInvoke __init_fail_point()
-
-mysql> alter table test.t_drop add column c int;
+# create table -> add tiflash replica -> drop table -> raft command is send to tiflash
+>> DBGInvoke __enable_schema_sync_service('false')
+mysql> drop table if exists test.t4;
+mysql> create table test.t4(a int primary key, b decimal(5,2) not NULL, c varchar(10), d int default 0);
+mysql> alter table test.t4 set tiflash replica 1; 
+func> wait_table test t4
 
 >> DBGInvoke __enable_fail_point(pause_before_apply_raft_cmd)
-
-# exactly write until fail point "pause_before_apply_raft_cmd" to be disable
-mysql> insert into test.t_drop values(1,2,3);
-
-mysql> drop table test.t_drop;
-
+mysql> insert into test.t4 values(1, 1.2, 'v', 2);
 => DBGInvoke __refresh_schemas()
-
-# make write cmd take effect
+mysql> drop table test.t4;
 >> DBGInvoke __disable_fail_point(pause_before_apply_raft_cmd)
-
-## wait the insert finish
-SLEEP 3
-
-# check the table is tombstone
->> select tidb_database,tidb_name from system.tables where tidb_database = 'test' and tidb_name = 't_drop' and is_tombstone = 0
-
-mysql> recover table test.t_drop;
-
-mysql> set session tidb_isolation_read_engines='tiflash';select * from test.t_drop;
-+----+-----+------+
-| a  |  b  |  c   |
-+----+-----+------+
-|  1 |   1 | NULL |
-|  1 |   2 |   3  |
-+----+-----+------+
-
-mysql> drop table test.t_drop;
-
 => DBGInvoke __refresh_schemas()
->> select tidb_database,tidb_name from system.tables where tidb_database = 'test' and tidb_name = 't_drop'
-┌─tidb_database─┬─tidb_name─┐
-│ test          │ t_drop    │
-└───────────────┴───────────┘
-
-=> DBGInvoke __enable_schema_sync_service('true') 
-=> DBGInvoke __gc_schemas(9223372036854775807)
-
-# check the table is physically dropped
->> select tidb_database,tidb_name from system.tables where tidb_database = 'test' and tidb_name = 't_drop'
\ No newline at end of file
+# ensure t4 is tombstone
+>> select tidb_database,tidb_name,tidb_table_id from system.tables where tidb_database = 'test' and tidb_name = 't4' and is_tombstone = 0
+>> select count(*) from system.tables where tidb_database = 'test' and tidb_name = 't4' and is_tombstone != 0
+┌─count()─┐
+│       1 │
+└─────────┘
+
+# re-enable
+>> DBGInvoke __enable_schema_sync_service('true')
diff --git a/tests/fullstack-test/fault-inject/drop-table.test b/tests/fullstack-test2/ddl/alter_drop_table_crash.test
similarity index 95%
rename from tests/fullstack-test/fault-inject/drop-table.test
rename to tests/fullstack-test2/ddl/alter_drop_table_crash.test
index feeb1feeaa3..c073ef4dc2f 100644
--- a/tests/fullstack-test/fault-inject/drop-table.test
+++ b/tests/fullstack-test2/ddl/alter_drop_table_crash.test
@@ -43,8 +43,7 @@ mysql> set session tidb_isolation_read_engines='tiflash'; select * from test.t;
 
 
 
-mysql> insert into test.t values (1, 1)
-mysql> insert into test.t values (1, 2)
+mysql> insert into test.t values (1, 1), (1, 2);
 mysql> set session tidb_isolation_read_engines='tiflash'; select * from test.t;
 +---+---+
 | a | b |
@@ -61,8 +60,7 @@ func> wait_table test t
 # After restart, test.t is truncated, it is empty
 mysql> set session tidb_isolation_read_engines='tiflash'; select * from test.t;
 
-mysql> insert into test.t values (1, 1)
-mysql> insert into test.t values (1, 2)
+mysql> insert into test.t values (1, 1), (1, 2);
 mysql> set session tidb_isolation_read_engines='tiflash'; select * from test.t;
 
 +---+---+
diff --git a/tests/fullstack-test2/ddl/flashback/flashback_table.test b/tests/fullstack-test2/ddl/flashback/flashback_table.test
new file mode 100644
index 00000000000..5277f26daca
--- /dev/null
+++ b/tests/fullstack-test2/ddl/flashback/flashback_table.test
@@ -0,0 +1,72 @@
+# Copyright 2023 PingCAP, Inc.
+#
+# Licensed 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.
+
+mysql> drop table if exists test.t
+mysql> drop table if exists test.t2
+
+mysql> create table test.t (c1 int, c2 varchar(64))
+mysql> ALTER TABLE test.t SET TIFLASH REPLICA 1
+
+func> wait_table test t
+
+mysql> insert into test.t values(1, 'abc')
+
+mysql> drop table test.t
+
+SLEEP 10
+
+mysql> flashback table test.t to t2
+mysql> set session tidb_isolation_read_engines='tiflash'; select * from test.t2
++------+------+
+| c1   | c2   |
++------+------+
+|    1 | abc  |
++------+------+
+mysql> set session tidb_isolation_read_engines='tikv'; select * from test.t2
++------+------+
+| c1   | c2   |
++------+------+
+|    1 | abc  |
++------+------+
+
+mysql> drop table if exists test.t
+mysql> drop table if exists test.t2
+
+mysql> create table test.t (c1 int, c2 varchar(64))
+mysql> ALTER TABLE test.t SET TIFLASH REPLICA 1
+
+func> wait_table test t
+
+mysql> insert into test.t values(1, 'abc')
+mysql> truncate table test.t
+
+SLEEP 10
+
+mysql> flashback table test.t to t2
+
+func> wait_table test t2
+mysql> set session tidb_isolation_read_engines='tiflash'; select * from test.t2
++------+------+
+| c1   | c2   |
++------+------+
+|    1 | abc  |
++------+------+
+mysql> set session tidb_isolation_read_engines='tikv'; select * from test.t2
++------+------+
+| c1   | c2   |
++------+------+
+|    1 | abc  |
++------+------+
+
+mysql> drop table if exists test.t2
diff --git a/tests/fullstack-test/fault-inject/recover_table.test b/tests/fullstack-test2/ddl/flashback/recover_table.test
similarity index 58%
rename from tests/fullstack-test/fault-inject/recover_table.test
rename to tests/fullstack-test2/ddl/flashback/recover_table.test
index 6b6700945bd..890f776be02 100644
--- a/tests/fullstack-test/fault-inject/recover_table.test
+++ b/tests/fullstack-test2/ddl/flashback/recover_table.test
@@ -12,17 +12,18 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-mysql> drop table if exists test.t;
+# Clean the tombstone table in the testing env
+>> DBGInvoke __enable_schema_sync_service('true')
+>> DBGInvoke __gc_schemas(18446744073709551615)
 
-### Test case for applying raft cmd for tombstoned table
+### Case 1
+## Test case for applying raft cmd for tombstoned table
+mysql> drop table if exists test.t;
 mysql> create table test.t(id int);
 mysql> alter table test.t set tiflash replica 1;
 
 func> wait_table test t
 
-# Disable flushing.
-
-
 # Insert a record and Read once (not necessary).
 mysql> insert into test.t values (1);
 mysql> set session tidb_isolation_read_engines='tiflash'; select * from test.t;
@@ -60,8 +61,8 @@ mysql> set session tidb_isolation_read_engines='tiflash'; select * from test.t;
 mysql> drop table if exists test.t;
 
 
-#####
-### Test case for applying raft snapshot for tombstoned table
+### Case 2
+## Test case for applying raft snapshot for tombstoned table
 mysql> create table test.t(id int);
 # It is important that TiFlash has synced the table schema
 >> DBGInvoke __refresh_schemas()
@@ -103,3 +104,64 @@ mysql> set session tidb_isolation_read_engines='tiflash'; select * from test.t;
 +------+
 
 mysql> drop table if exists test.t;
+
+
+### Case 3
+## drop table arrive tiflash before ddl and insert, and do recover, check the data is not lost
+mysql> drop table if exists test.t_drop; 
+mysql> create table test.t_drop(a int, b int);
+mysql> alter table test.t_drop set tiflash replica 1;
+mysql> insert into test.t_drop values(1, 1);
+
+func> wait_table test t_drop
+
+=> DBGInvoke __enable_schema_sync_service('false') 
+=> DBGInvoke __init_fail_point()
+
+mysql> alter table test.t_drop add column c int;
+
+>> DBGInvoke __enable_fail_point(pause_before_apply_raft_cmd)
+
+# exactly write until fail point "pause_before_apply_raft_cmd" to be disable
+mysql> insert into test.t_drop values(1,2,3);
+
+mysql> drop table test.t_drop;
+
+=> DBGInvoke __refresh_schemas()
+
+# make write cmd take effect
+>> DBGInvoke __disable_fail_point(pause_before_apply_raft_cmd)
+
+## wait the insert finish
+SLEEP 3
+
+# check the table is tombstone
+>> select tidb_database,tidb_name from system.tables where tidb_database = 'test' and tidb_name = 't_drop' and is_tombstone = 0
+
+mysql> recover table test.t_drop;
+
+# we should be able to read the data on column `c`
+mysql> set session tidb_isolation_read_engines='tiflash';select * from test.t_drop;
++----+-----+------+
+| a  |  b  |  c   |
++----+-----+------+
+|  1 |   1 | NULL |
+|  1 |   2 |   3  |
++----+-----+------+
+
+mysql> drop table test.t_drop;
+
+=> DBGInvoke __refresh_schemas()
+>> select tidb_database,tidb_name from system.tables where tidb_database = 'test' and tidb_name = 't_drop'
+┌─tidb_database─┬─tidb_name─┐
+│ test          │ t_drop    │
+└───────────────┴───────────┘
+
+=> DBGInvoke __enable_schema_sync_service('true') 
+=> DBGInvoke __gc_schemas(9223372036854775807)
+
+# check the table is physically dropped
+>> select tidb_database,tidb_name from system.tables where tidb_database = 'test' and tidb_name = 't_drop'
+
+# re-enable
+>> DBGInvoke __enable_schema_sync_service('true')
diff --git a/tests/fullstack-test2/ddl/alter_exchange_partition.test b/tests/fullstack-test2/ddl/partitions/alter_exchange_partition.test
similarity index 74%
rename from tests/fullstack-test2/ddl/alter_exchange_partition.test
rename to tests/fullstack-test2/ddl/partitions/alter_exchange_partition.test
index 8854c401e7e..6f740ace25c 100644
--- a/tests/fullstack-test2/ddl/alter_exchange_partition.test
+++ b/tests/fullstack-test2/ddl/partitions/alter_exchange_partition.test
@@ -89,7 +89,7 @@ mysql> drop table if exists test.e2;
 mysql> drop table if exists test_new.e2;
 mysql> drop database if exists test_new;
 
-# case 11, create non-partition table and execute exchagne partition immediately
+## case 11, create non-partition table and execute exchagne partition immediately
 mysql> create table test.e(id INT NOT NULL,fname VARCHAR(30),lname VARCHAR(30)) PARTITION BY RANGE (id) ( PARTITION p0 VALUES LESS THAN (50),PARTITION p1 VALUES LESS THAN (100),PARTITION p2 VALUES LESS THAN (150), PARTITION p3 VALUES LESS THAN (MAXVALUE));
 mysql> insert into test.e values (1, 'a', 'b'),(108, 'a', 'b');
 # sync the partition table to tiflash
@@ -99,44 +99,52 @@ mysql> create table test.e2(id int not null,fname varchar(30),lname varchar(30))
 >> DBGInvoke __refresh_schemas()
 mysql> insert into test.e2 values (2, 'a', 'b');
 mysql> set @@tidb_enable_exchange_partition=1; alter table test.e exchange partition p0 with table test.e2
+mysql> alter table test.e add column c1 int;
+mysql> alter table test.e2 add column c2 int;
+mysql> insert into test.e2 values (3, 'a', 'b', 3); insert into test.e values (4, 'a', 'b', 4);
 
 mysql> alter table test.e set tiflash replica 1;
 mysql> alter table test.e2 set tiflash replica 1;
 func> wait_table test e e2
+# ensure tiflash see the column `e.c1` and `e2.c2`
 >> DBGInvoke __refresh_schemas()
 mysql> set session tidb_isolation_read_engines='tiflash'; select * from test.e order by id;
-+-----+-------+-------+
-| id  | fname | lname |
-+-----+-------+-------+
-|   2 | a     | b     |
-| 108 | a     | b     |
-+-----+-------+-------+
++-----+-------+-------+------+
+| id  | fname | lname | c1   |
++-----+-------+-------+------+
+|   2 | a     | b     | NULL |
+|   4 | a     | b     |    4 |
+| 108 | a     | b     | NULL |
++-----+-------+-------+------+
 mysql> set session tidb_isolation_read_engines='tiflash'; select * from test.e2 order by id;
-+-----+-------+-------+
-| id  | fname | lname |
-+-----+-------+-------+
-|   1 | a     | b     |
-+-----+-------+-------+
++----+-------+-------+------+
+| id | fname | lname | c2   |
++----+-------+-------+------+
+|  1 | a     | b     | NULL |
+|  3 | a     | b     |    3 |
++----+-------+-------+------+
 
 # ensure the swap out table is not mark as tombstone
 >> DBGInvoke __enable_schema_sync_service('true')
 >> DBGInvoke __gc_schemas(18446744073709551615)
 >> DBGInvoke __enable_schema_sync_service('false')
 mysql> set session tidb_isolation_read_engines='tiflash'; select * from test.e order by id;
-+-----+-------+-------+
-| id  | fname | lname |
-+-----+-------+-------+
-|   2 | a     | b     |
-| 108 | a     | b     |
-+-----+-------+-------+
++-----+-------+-------+------+
+| id  | fname | lname | c1   |
++-----+-------+-------+------+
+|   2 | a     | b     | NULL |
+|   4 | a     | b     |    4 |
+| 108 | a     | b     | NULL |
++-----+-------+-------+------+
 mysql> set session tidb_isolation_read_engines='tiflash'; select * from test.e2 order by id;
-+-----+-------+-------+
-| id  | fname | lname |
-+-----+-------+-------+
-|   1 | a     | b     |
-+-----+-------+-------+
-
-# case 12, create partition table, non-partition table and execute exchagne partition immediately
++----+-------+-------+------+
+| id | fname | lname | c2   |
++----+-------+-------+------+
+|  1 | a     | b     | NULL |
+|  3 | a     | b     |    3 |
++----+-------+-------+------+
+
+## case 12, create partition table, non-partition table and execute exchagne partition immediately
 mysql> drop table if exists test.e
 mysql> drop table if exists test.e2
 mysql> create table test.e(id INT NOT NULL,fname VARCHAR(30),lname VARCHAR(30)) PARTITION BY RANGE (id) ( PARTITION p0 VALUES LESS THAN (50),PARTITION p1 VALUES LESS THAN (100),PARTITION p2 VALUES LESS THAN (150), PARTITION p3 VALUES LESS THAN (MAXVALUE));
@@ -144,6 +152,9 @@ mysql> insert into test.e values (1, 'a', 'b'),(108, 'a', 'b');
 mysql> create table test.e2(id int not null,fname varchar(30),lname varchar(30));
 mysql> insert into test.e2 values (2, 'a', 'b');
 mysql> set @@tidb_enable_exchange_partition=1; alter table test.e exchange partition p0 with table test.e2
+mysql> alter table test.e add column c1 int;
+mysql> alter table test.e2 add column c2 int;
+mysql> insert into test.e2 values (3, 'a', 'b', 3); insert into test.e values (4, 'a', 'b', 4);
 
 mysql> alter table test.e set tiflash replica 1;
 mysql> alter table test.e2 set tiflash replica 1;
@@ -151,35 +162,39 @@ func> wait_table test e e2
 # tiflash the final result
 >> DBGInvoke __refresh_schemas()
 mysql> set session tidb_isolation_read_engines='tiflash'; select * from test.e order by id;
-+-----+-------+-------+
-| id  | fname | lname |
-+-----+-------+-------+
-|   2 | a     | b     |
-| 108 | a     | b     |
-+-----+-------+-------+
++-----+-------+-------+------+
+| id  | fname | lname | c1   |
++-----+-------+-------+------+
+|   2 | a     | b     | NULL |
+|   4 | a     | b     |    4 |
+| 108 | a     | b     | NULL |
++-----+-------+-------+------+
 mysql> set session tidb_isolation_read_engines='tiflash'; select * from test.e2 order by id;
-+-----+-------+-------+
-| id  | fname | lname |
-+-----+-------+-------+
-|   1 | a     | b     |
-+-----+-------+-------+
++----+-------+-------+------+
+| id | fname | lname | c2   |
++----+-------+-------+------+
+|  1 | a     | b     | NULL |
+|  3 | a     | b     |    3 |
++----+-------+-------+------+
 # ensure the swap out table is not mark as tombstone
 >> DBGInvoke __enable_schema_sync_service('true')
 >> DBGInvoke __gc_schemas(18446744073709551615)
 >> DBGInvoke __enable_schema_sync_service('false')
 mysql> set session tidb_isolation_read_engines='tiflash'; select * from test.e order by id;
-+-----+-------+-------+
-| id  | fname | lname |
-+-----+-------+-------+
-|   2 | a     | b     |
-| 108 | a     | b     |
-+-----+-------+-------+
++-----+-------+-------+------+
+| id  | fname | lname | c1   |
++-----+-------+-------+------+
+|   2 | a     | b     | NULL |
+|   4 | a     | b     |    4 |
+| 108 | a     | b     | NULL |
++-----+-------+-------+------+
 mysql> set session tidb_isolation_read_engines='tiflash'; select * from test.e2 order by id;
-+-----+-------+-------+
-| id  | fname | lname |
-+-----+-------+-------+
-|   1 | a     | b     |
-+-----+-------+-------+
++----+-------+-------+------+
+| id | fname | lname | c2   |
++----+-------+-------+------+
+|  1 | a     | b     | NULL |
+|  3 | a     | b     |    3 |
++----+-------+-------+------+
 
 # cleanup
 mysql> drop table if exists test.e;
diff --git a/tests/fullstack-test2/ddl/alter_partition_by.test b/tests/fullstack-test2/ddl/partitions/alter_partition_by.test
similarity index 100%
rename from tests/fullstack-test2/ddl/alter_partition_by.test
rename to tests/fullstack-test2/ddl/partitions/alter_partition_by.test
diff --git a/tests/fullstack-test2/ddl/alter_partition.test b/tests/fullstack-test2/ddl/partitions/partition_basic.test
similarity index 84%
rename from tests/fullstack-test2/ddl/alter_partition.test
rename to tests/fullstack-test2/ddl/partitions/partition_basic.test
index a211e26b744..fdba3f2f922 100644
--- a/tests/fullstack-test2/ddl/alter_partition.test
+++ b/tests/fullstack-test2/ddl/partitions/partition_basic.test
@@ -12,17 +12,17 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+## case 1
 # basic add / drop / truncate partitions
 mysql> drop table if exists test.t1;
 mysql> create table test.t1(id INT NOT NULL,name VARCHAR(30)) PARTITION BY RANGE (id) ( PARTITION p0 VALUES LESS THAN (50),PARTITION p1 VALUES LESS THAN (100));
 mysql> alter table test.t1 set tiflash replica 1;
 
-mysql> insert into test.t1 values (1, 'abc');
-mysql> insert into test.t1 values (60, 'cba');
+mysql> insert into test.t1 values (1, 'abc'),(60, 'cba');
 
 func> wait_table test t1
 
-mysql> select /*+ read_from_storage(tiflash[t]) */ * from test.t1;
+mysql> set session tidb_isolation_read_engines='tiflash'; select * from test.t1;
 +----+------+
 | id | name |
 +----+------+
@@ -41,7 +41,7 @@ mysql> alter table test.t1 add partition (partition p2 values less than (200));
 
 mysql> insert into test.t1 values (150, 'aaa');
 
-mysql> select /*+ read_from_storage(tiflash[t]) */ * from test.t1;
+mysql> set session tidb_isolation_read_engines='tiflash'; select * from test.t1;
 +----+------+
 | id | name |
 +----+------+
@@ -65,7 +65,7 @@ mysql> alter table test.t1 drop partition p0;
 │ 1/1/                                                │
 └─────────────────────────────────────────────────────┘
 
-mysql> select /*+ read_from_storage(tiflash[t]) */ * from test.t1;
+mysql> set session tidb_isolation_read_engines='tiflash'; select * from test.t1;
 +----+------+
 | id | name |
 +----+------+
@@ -78,8 +78,8 @@ mysql> alter table test.t1 truncate partition p1;
 => DBGInvoke __refresh_schemas()
 
 # In raftstore v1, this new partition will not be created in TiFlash. On the contrary in raftstore v2.
-
-mysql> select /*+ read_from_storage(tiflash[t]) */ * from test.t1;
+func> wait_table test t1
+mysql> set session tidb_isolation_read_engines='tiflash'; select * from test.t1;
 +----+------+
 | id | name |
 +----+------+
@@ -88,14 +88,14 @@ mysql> select /*+ read_from_storage(tiflash[t]) */ * from test.t1;
 
 mysql> drop table test.t1;
 
+## case 2
 ## test before drop / truncate partition, we make alter column and insert data
 
 mysql> drop table if exists test.t2;
 mysql> create table test.t2(id INT NOT NULL,name VARCHAR(30)) PARTITION BY RANGE (id) ( PARTITION p0 VALUES LESS THAN (50),PARTITION p1 VALUES LESS THAN (100));
 mysql> alter table test.t2 set tiflash replica 1;
 
-mysql> insert into test.t2 values (1, 'abc');
-mysql> insert into test.t2 values (60, 'cba');
+mysql> insert into test.t2 values (1, 'abc'),(60, 'cba');
 
 func> wait_table test t2
 
@@ -116,7 +116,7 @@ mysql> alter table test.t2 drop partition p0;
 # make write cmd take effect
 >> DBGInvoke __disable_fail_point(pause_before_apply_raft_cmd)
 
-mysql> select /*+ read_from_storage(tiflash[t]) */ * from test.t2;
+mysql> set session tidb_isolation_read_engines='tiflash'; select * from test.t2;
 +----+------+----+
 | id | name |  c |
 +----+------+----+
@@ -135,9 +135,10 @@ mysql> alter table test.t2 truncate partition p1;
 
 => DBGInvoke __refresh_schemas()
 
-# make write cmd take effect
+# make write cmd take effect, the row is written to the old partition without crash
 >> DBGInvoke __disable_fail_point(pause_before_apply_raft_cmd)
 
-mysql> select /*+ read_from_storage(tiflash[t]) */ * from test.t2;
+func> wait_table test t2
+mysql> set session tidb_isolation_read_engines='tiflash'; select * from test.t2;
 
 mysql> drop table test.t2;
diff --git a/tests/fullstack-test2/ddl/remove_partitioning.test b/tests/fullstack-test2/ddl/partitions/remove_partitioning.test
similarity index 100%
rename from tests/fullstack-test2/ddl/remove_partitioning.test
rename to tests/fullstack-test2/ddl/partitions/remove_partitioning.test
diff --git a/tests/fullstack-test2/ddl/reorganize_partition.test b/tests/fullstack-test2/ddl/partitions/reorganize_partition.test
similarity index 84%
rename from tests/fullstack-test2/ddl/reorganize_partition.test
rename to tests/fullstack-test2/ddl/partitions/reorganize_partition.test
index 9453bc373e3..7c208d843d8 100644
--- a/tests/fullstack-test2/ddl/reorganize_partition.test
+++ b/tests/fullstack-test2/ddl/partitions/reorganize_partition.test
@@ -119,14 +119,14 @@ mysql> set session tidb_isolation_read_engines='tiflash'; select count(*) from t
 mysql> drop table test.t;
 
 
+## case 2
 # do ddl and insert before action reorganize partition
 
 mysql> drop table if exists test.t1
 mysql> create table test.t1(id INT NOT NULL,name VARCHAR(30)) PARTITION BY RANGE (id) ( PARTITION p0 VALUES LESS THAN (50),PARTITION p1 VALUES LESS THAN (100));
 mysql> alter table test.t1 set tiflash replica 1;
 
-mysql> insert into test.t1 values (1, 'abc');
-mysql> insert into test.t1 values (60, 'cba');
+mysql> insert into test.t1 values (1, 'abc'),(60, 'cba');
 
 func> wait_table test t1
 
@@ -140,6 +140,7 @@ mysql> alter table test.t1 add column c int;
 # exactly write until fail point "pause_before_apply_raft_cmd" to be disable
 mysql> insert into test.t1 values(80, 'aaa', 2);
 
+# reorganize partition will split p1 into new partitions
 mysql> alter table test.t1 reorganize partition p1 INTO (partition p1 values less than (70), partition p2 values less than (100));
 
 => DBGInvoke __refresh_schemas()
@@ -161,4 +162,27 @@ mysql> set session tidb_isolation_read_engines='tiflash'; select * from test.t1
 | 80 | aaa  | 2  |
 +----+------+----+
 
-mysql> drop table test.t1;
+# make sure the p0 is not affected
+mysql> set session tidb_isolation_read_engines='tiflash'; select * from test.t1 order by id;
++----+------+------+
+| id | name | c    |
++----+------+------+
+|  1 | abc  | NULL |
+| 60 | cba  | NULL |
+| 80 | aaa  |    2 |
++----+------+------+
+
+# ensure the partitions is not mark as tombstone
+>> DBGInvoke __enable_schema_sync_service('true')
+>> DBGInvoke __gc_schemas(18446744073709551615)
+>> DBGInvoke __enable_schema_sync_service('false')
+mysql> set session tidb_isolation_read_engines='tiflash'; select * from test.t1 order by id;
++----+------+------+
+| id | name | c    |
++----+------+------+
+|  1 | abc  | NULL |
+| 60 | cba  | NULL |
+| 80 | aaa  |    2 |
++----+------+------+
+
+#mysql> drop table test.t1;
diff --git a/tests/fullstack-test/fault-inject/rename-table.test b/tests/fullstack-test2/ddl/rename_table_crash.test
similarity index 100%
rename from tests/fullstack-test/fault-inject/rename-table.test
rename to tests/fullstack-test2/ddl/rename_table_crash.test
diff --git a/tests/run-test.py b/tests/run-test.py
index 06cf55baf98..b4901d1cd79 100644
--- a/tests/run-test.py
+++ b/tests/run-test.py
@@ -21,6 +21,7 @@
 import re
 import sys
 import time
+import datetime
 
 if sys.version_info.major == 2:
     # print('running with py2: {}.{}.{}'.format(sys.version_info.major, sys.version_info.minor, sys.version_info.micro))
@@ -282,7 +283,7 @@ def on_line(self, line, line_number):
             if line.endswith(NO_UNESCAPE_SUFFIX):
                 unescape_flag = False
                 line = line[:-len(NO_UNESCAPE_SUFFIX)]
-            if verbose: print('running', line)
+            if verbose: print('{} running {}'.format(datetime.datetime.now().strftime('%H:%M:%S.%f'), line))
             if self.outputs != None and ((not self.is_mysql and not matched(self.outputs, self.matches, self.fuzz)) or (
                 self.is_mysql and not MySQLCompare.matched(self.outputs, self.matches))):
                 return False
@@ -294,7 +295,7 @@ def on_line(self, line, line_number):
             self.outputs = [x.strip() for x in self.outputs if len(x.strip()) != 0]
             self.matches = []
         elif line.startswith(CURL_TIDB_STATUS_PREFIX):
-            if verbose: print('running', line)
+            if verbose: print('{} running {}'.format(datetime.datetime.now().strftime('%H:%M:%S.%f'), line))
             if self.outputs != None and ((not self.is_mysql and not matched(self.outputs, self.matches, self.fuzz)) or (
                 self.is_mysql and not MySQLCompare.matched(self.outputs, self.matches))):
                 return False
@@ -306,7 +307,7 @@ def on_line(self, line, line_number):
                 return False
             self.matches = []
         elif line.startswith(CMD_PREFIX) or line.startswith(CMD_PREFIX_ALTER):
-            if verbose: print('running', line)
+            if verbose: print('{} running {}'.format(datetime.datetime.now().strftime('%H:%M:%S.%f'), line))
             if self.outputs != None and ((not self.is_mysql and not matched(self.outputs, self.matches, self.fuzz)) or (
                 self.is_mysql and not MySQLCompare.matched(self.outputs, self.matches))):
                 return False
@@ -320,7 +321,7 @@ def on_line(self, line, line_number):
                 self.outputs = [x for x in self.outputs if x.find(ignored_output) < 0]
             self.matches = []
         elif line.startswith(CMD_PREFIX_FUNC):
-            if verbose: print('running', line)
+            if verbose: print('{} running {}'.format(datetime.datetime.now().strftime('%H:%M:%S.%f'), line))
             if self.outputs != None and ((not self.is_mysql and not matched(self.outputs, self.matches, self.fuzz)) or (
                 self.is_mysql and not MySQLCompare.matched(self.outputs, self.matches))):
                 return False