ddl: Fix snap after dropping database

dbms/src/TiDB/Schema/SchemaBuilder.cpp

@@ -1100,6 +1100,11 @@ void SchemaBuilder<Getter, NameMapper>::applyCreateStorageInstance(
         table_info->engine_type = tmt_context.getEngineType();
     }
 
+    // We need to create a Storage instance to handle its raft log and snapshot when it
+    // is "dropped" but not physically removed in TiDB. To handle it porperly, we get a
+    // tso from PD to create the table. The tso must be newer than what "DROP TABLE" DDL
+    // is executed. So when the gc-safepoint is larger than tombstone_ts, the table can
+    // be safe to physically drop on TiFlash.
     UInt64 tombstone_ts = 0;
     if (is_tombstone)
     {
@@ -1116,13 +1121,38 @@ void SchemaBuilder<Getter, NameMapper>::applyCreateStorageInstance(
         table_info->id,
         stmt);
 
+    // If "CREATE DATABASE" is executed in TiFlash after user has executed "DROP DATABASE"
+    // in TiDB, then TiFlash may not create the IDatabase instance. Make sure we can access
+    // to the IDatabase when creating IStorage.
+    const auto database_mapped_name = name_mapper.mapDatabaseName(database_id, keyspace_id);
+    if (!context.isDatabaseExist(database_mapped_name))
+    {
+        LOG_WARNING(
+            log,
+            "database instance is not exist (applyCreateStorageInstance), may has been dropped, create a database with "
+            "fake DatabaseInfo for it, database_id={} database_name={}",
+            database_id,
+            database_mapped_name);
+        // The database is dropped in TiKV and we can not fetch it. Generate a fake
+        // DatabaseInfo for it. It is OK because the DatabaseInfo will be updated
+        // when the database is `FLASHBACK`.
+        TiDB::DBInfoPtr database_info = std::make_shared<TiDB::DBInfo>();
+        database_info->id = database_id;
+        database_info->keyspace_id = keyspace_id;
+        database_info->name = database_mapped_name; // use the mapped name because we done known the actual name
+        database_info->charset = "utf8mb4"; // default value
+        database_info->collate = "utf8mb4_bin"; // default value
+        database_info->state = TiDB::StateNone; // special state
+        applyCreateDatabaseByInfo(database_info);
+    }
+
     ParserCreateQuery parser;
     ASTPtr ast = parseQuery(parser, stmt.data(), stmt.data() + stmt.size(), "from syncSchema " + table_info->name, 0);
 
     auto * ast_create_query = typeid_cast<ASTCreateQuery *>(ast.get());
     ast_create_query->attach = true;
     ast_create_query->if_not_exists = true;
-    ast_create_query->database = name_mapper.mapDatabaseName(database_id, keyspace_id);
+    ast_create_query->database = database_mapped_name;
 
     InterpreterCreateQuery interpreter(ast, context);
     interpreter.setInternal(true);

dbms/src/TiDB/Schema/SchemaGetter.cpp

@@ -182,6 +182,7 @@ void AffectedOption::deserialize(Poco::JSON::Object::Ptr json)
 
 void SchemaDiff::deserialize(const String & data)
 {
+    assert(!data.empty()); // should be skipped by upper level logic
     Poco::JSON::Parser parser;
     try
     {
@@ -218,7 +219,7 @@ void SchemaDiff::deserialize(const String & data)
     }
     catch (...)
     {
-        LOG_INFO(Logger::get(), "failed to deserialize {}", data);
+        LOG_WARNING(Logger::get(), "failed to deserialize {}", data);
         throw;
     }
 }

dbms/src/TiDB/Schema/SchemaSyncService.cpp

@@ -243,7 +243,7 @@ bool SchemaSyncService::gc(Timestamp gc_safepoint, KeyspaceID keyspace_id)
                 // it is dropped.
                 storages_to_gc.emplace_back(std::weak_ptr<IManageableStorage>(managed_storage));
                 LOG_INFO(
-                    log,
+                    keyspace_log,
                     "Detect stale table, database_name={} table_name={} database_tombstone={} table_tombstone={} "
                     "safepoint={}",
                     managed_storage->getDatabaseName(),

dbms/src/TiDB/Schema/TiDB.cpp

@@ -1096,7 +1096,8 @@ ColumnID TableInfo::getColumnID(const String & name) const
 
     throw DB::Exception(
         DB::ErrorCodes::LOGICAL_ERROR,
-        "Fail to get column id from TableInfo, name={} available_columns={}",
+        "Fail to get column id from TableInfo, table_id={} name={} available_columns={}",
+        id,
         name,
         available_columns);
 }

dbms/src/TiDB/Schema/TiDB.h

@@ -277,7 +277,7 @@ struct DBInfo
     String name;
     String charset;
     String collate;
-    SchemaState state;
+    SchemaState state = StatePublic;
 
     DBInfo() = default;
     explicit DBInfo(const String & json, KeyspaceID keyspace_id_)
@@ -321,8 +321,8 @@ struct IndexColumnInfo
     void deserialize(Poco::JSON::Object::Ptr json);
 
     String name;
-    Int32 length;
-    Int32 offset;
+    Int32 length = 0;
+    Int32 offset = 0;
 };
 struct IndexInfo
 {
@@ -334,16 +334,16 @@ struct IndexInfo
 
     void deserialize(Poco::JSON::Object::Ptr json);
 
-    Int64 id;
+    Int64 id = -1;
     String idx_name;
     String tbl_name;
     std::vector<IndexColumnInfo> idx_cols;
-    SchemaState state;
-    Int32 index_type;
-    bool is_unique;
-    bool is_primary;
-    bool is_invisible;
-    bool is_global;
+    SchemaState state = StatePublic;
+    Int32 index_type = -1;
+    bool is_unique = false;
+    bool is_primary = false;
+    bool is_invisible = false;
+    bool is_global = false;
 };
 
 struct TableInfo

tests/fullstack-test2/ddl/flashback/flashback_database.test

@@ -54,6 +54,7 @@ mysql> set session tidb_isolation_read_engines='tiflash'; select * from d1_new.t
 
 # ensure the flashbacked table and database is not mark as tombstone
 >> DBGInvoke __enable_schema_sync_service('true')
+=> DBGInvoke __refresh_schemas()
 >> DBGInvoke __gc_schemas(18446744073709551615)
 
 mysql> set session tidb_isolation_read_engines='tiflash'; select * from d1_new.t3 order by a;
@@ -189,3 +190,50 @@ mysql> set session tidb_isolation_read_engines='tiflash'; select * from d1_new.t
 # cleanup
 mysql> drop database if exists d1;
 mysql> drop database if exists d1_new;
+
+## case 4, "create database" and "region snapshot" comes after "drop database" is executed in tidb
+
+# disable all raft log/snapshot
+>> DBGInvoke __enable_fail_point(pause_before_apply_raft_cmd)
+>> DBGInvoke __enable_fail_point(pause_before_prehandle_snapshot)
+
+mysql> create database d1;
+mysql> create table d1.t1 (a int);
+mysql> insert into d1.t1 values(1),(2),(3);
+mysql> alter table d1.t1 set tiflash replica 1;
+
+mysql> drop database d1;
+
+# "create database" and "snapshot" comes after "drop database"
+>> DBGInvoke __disable_fail_point(pause_before_apply_raft_cmd)
+>> DBGInvoke __disable_fail_point(pause_before_prehandle_snapshot)
+
+# check the row is written to the storage or not
+mysql> flashback database d1 to d1_new
+# wait available after flashback
+func> wait_table d1_new t1
+mysql> set session tidb_isolation_read_engines='tiflash'; select * from d1_new.t1 order by a;
++------+
+| a    |
++------+
+|    1 |
+|    2 |
+|    3 |
++------+
+
+# ensure the flashbacked table and database 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 d1_new.t1 order by a;
++------+
+| a    |
++------+
+|    1 |
+|    2 |
+|    3 |
++------+
+
+# cleanup
+mysql> drop database if exists d1;
+mysql> drop database if exists d1_new;