diff --git a/dbms/src/Databases/test/gtest_database.cpp b/dbms/src/Databases/test/gtest_database.cpp
index b469a571570..95aa75520f2 100644
--- a/dbms/src/Databases/test/gtest_database.cpp
+++ b/dbms/src/Databases/test/gtest_database.cpp
@@ -97,16 +97,10 @@ class DatabaseTiFlashTest : public ::testing::Test
     static void recreateMetadataPath()
     {
         String path = TiFlashTestEnv::getContext().getPath();
-
         auto p = path + "/metadata/";
-        if (Poco::File file(p); file.exists())
-            file.remove(true);
-        Poco::File{p}.createDirectories();
-
+        TiFlashTestEnv::tryRemovePath(p, /*recreate=*/true);
         p = path + "/data/";
-        if (Poco::File file(p); file.exists())
-            file.remove(true);
-        Poco::File{p}.createDirectories();
+        TiFlashTestEnv::tryRemovePath(p, /*recreate=*/true);
     }
 
 protected:
diff --git a/dbms/src/Debug/MockSchemaGetter.h b/dbms/src/Debug/MockSchemaGetter.h
index 11c5d97f036..677455e895a 100644
--- a/dbms/src/Debug/MockSchemaGetter.h
+++ b/dbms/src/Debug/MockSchemaGetter.h
@@ -23,30 +23,33 @@ namespace DB
 {
 struct MockSchemaGetter
 {
-    TiDB::DBInfoPtr getDatabase(DatabaseID db_id) { return MockTiDB::instance().getDBInfoByID(db_id); }
+    static TiDB::DBInfoPtr getDatabase(DatabaseID db_id) { return MockTiDB::instance().getDBInfoByID(db_id); }
 
-    Int64 getVersion() { return MockTiDB::instance().getVersion(); }
+    static Int64 getVersion() { return MockTiDB::instance().getVersion(); }
 
-    std::optional<SchemaDiff> getSchemaDiff(Int64 version)
+    static std::optional<SchemaDiff> getSchemaDiff(Int64 version)
     {
         return MockTiDB::instance().getSchemaDiff(version);
     }
 
-    bool checkSchemaDiffExists(Int64 version)
+    static bool checkSchemaDiffExists(Int64 version)
     {
         return MockTiDB::instance().checkSchemaDiffExists(version);
     }
 
-    TiDB::TableInfoPtr getTableInfo(DatabaseID, TableID table_id) { return MockTiDB::instance().getTableInfoByID(table_id); }
+    static TiDB::TableInfoPtr getTableInfo(DatabaseID, TableID table_id)
+    {
+        return MockTiDB::instance().getTableInfoByID(table_id);
+    }
 
-    std::vector<TiDB::DBInfoPtr> listDBs()
+    static std::vector<TiDB::DBInfoPtr> listDBs()
     {
         std::vector<TiDB::DBInfoPtr> res;
         const auto & databases = MockTiDB::instance().getDatabases();
-        for (auto it = databases.begin(); it != databases.end(); it++)
+        for (const auto & database : databases)
         {
-            auto db_id = it->second;
-            auto db_name = it->first;
+            auto db_id = database.second;
+            auto db_name = database.first;
             TiDB::DBInfoPtr db_ptr = std::make_shared<TiDB::DBInfo>(TiDB::DBInfo());
             db_ptr->id = db_id;
             db_ptr->name = db_name;
@@ -55,15 +58,15 @@ struct MockSchemaGetter
         return res;
     }
 
-    std::vector<TiDB::TableInfoPtr> listTables(Int64 db_id)
+    static std::vector<TiDB::TableInfoPtr> listTables(Int64 db_id)
     {
         auto tables_by_id = MockTiDB::instance().getTables();
         std::vector<TiDB::TableInfoPtr> res;
-        for (auto it = tables_by_id.begin(); it != tables_by_id.end(); it++)
+        for (auto & it : tables_by_id)
         {
-            if (it->second->dbID() == db_id)
+            if (it.second->dbID() == db_id)
             {
-                res.push_back(std::make_shared<TiDB::TableInfo>(TiDB::TableInfo(it->second->table_info)));
+                res.push_back(std::make_shared<TiDB::TableInfo>(TiDB::TableInfo(it.second->table_info)));
             }
         }
         return res;
diff --git a/dbms/src/Debug/MockTiDB.cpp b/dbms/src/Debug/MockTiDB.cpp
index 99d9625461b..61c359ec298 100644
--- a/dbms/src/Debug/MockTiDB.cpp
+++ b/dbms/src/Debug/MockTiDB.cpp
@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#include <Common/Exception.h>
 #include <DataTypes/DataTypeDecimal.h>
 #include <DataTypes/DataTypeEnum.h>
 #include <DataTypes/DataTypeMyDate.h>
@@ -31,6 +32,8 @@
 #include <Storages/Transaction/TiDB.h>
 #include <Storages/Transaction/TypeMapping.h>
 
+#include <mutex>
+
 namespace DB
 {
 namespace ErrorCodes
@@ -350,27 +353,40 @@ Field getDefaultValue(const ASTPtr & default_value_ast)
     return Field();
 }
 
-void MockTiDB::newPartition(const String & database_name, const String & table_name, TableID partition_id, Timestamp tso, bool is_add_part)
+TableID MockTiDB::newPartition(TableID belong_logical_table, const String & partition_name, Timestamp tso, bool is_add_part)
 {
     std::lock_guard lock(tables_mutex);
 
-    TablePtr table = getTableByNameInternal(database_name, table_name);
-    TableInfo & table_info = table->table_info;
+    TablePtr logical_table = getTableByID(belong_logical_table);
+    TableID partition_id = table_id_allocator++; // allocate automatically
 
-    const auto & part_def = find_if(
-        table_info.partition.definitions.begin(),
-        table_info.partition.definitions.end(),
-        [&partition_id](PartitionDefinition & part_def) { return part_def.id == partition_id; });
-    if (part_def != table_info.partition.definitions.end())
-        throw Exception("Mock TiDB table " + database_name + "." + table_name + " already has partition " + std::to_string(partition_id),
-                        ErrorCodes::LOGICAL_ERROR);
+    return newPartitionImpl(logical_table, partition_id, partition_name, tso, is_add_part);
+}
+
+TableID MockTiDB::newPartition(const String & database_name, const String & table_name, TableID partition_id, Timestamp tso, bool is_add_part)
+{
+    std::lock_guard lock(tables_mutex);
+
+    TablePtr logical_table = getTableByNameInternal(database_name, table_name);
+    return newPartitionImpl(logical_table, partition_id, toString(partition_id), tso, is_add_part);
+}
+
+TableID MockTiDB::newPartitionImpl(const TablePtr & logical_table, TableID partition_id, const String & partition_name, Timestamp tso, bool is_add_part)
+{
+    TableInfo & table_info = logical_table->table_info;
+    RUNTIME_CHECK_MSG(!logical_table->existPartitionID(partition_id),
+                      "Mock TiDB table {}.{} already has partition {}, table_info={}",
+                      logical_table->database_name,
+                      logical_table->table_name,
+                      partition_id,
+                      table_info.serialize());
 
     table_info.is_partition_table = true;
     table_info.partition.enable = true;
     table_info.partition.num++;
     PartitionDefinition partition_def;
     partition_def.id = partition_id;
-    partition_def.name = std::to_string(partition_id);
+    partition_def.name = partition_name;
     table_info.partition.definitions.emplace_back(partition_def);
     table_info.update_timestamp = tso;
 
@@ -380,11 +396,12 @@ void MockTiDB::newPartition(const String & database_name, const String & table_n
 
         SchemaDiff diff;
         diff.type = SchemaActionType::AddTablePartition;
-        diff.schema_id = table->database_id;
-        diff.table_id = table->id();
+        diff.schema_id = logical_table->database_id;
+        diff.table_id = logical_table->id();
         diff.version = version;
         version_diff[version] = diff;
     }
+    return partition_id;
 }
 
 void MockTiDB::dropPartition(const String & database_name, const String & table_name, TableID partition_id)
@@ -631,6 +648,13 @@ TablePtr MockTiDB::getTableByNameInternal(const String & database_name, const St
     return it->second;
 }
 
+TablePtr MockTiDB::getTableByID(TableID table_id)
+{
+    if (auto it = tables_by_id.find(table_id); it != tables_by_id.end())
+        return it->second;
+    throw Exception(fmt::format("Mock TiDB table does not exists, table_id={}", table_id), ErrorCodes::UNKNOWN_TABLE);
+}
+
 TiDB::TableInfoPtr MockTiDB::getTableInfoByID(TableID table_id)
 {
     auto it = tables_by_id.find(table_id);
diff --git a/dbms/src/Debug/MockTiDB.h b/dbms/src/Debug/MockTiDB.h
index 261e547b13a..1f0df73031e 100644
--- a/dbms/src/Debug/MockTiDB.h
+++ b/dbms/src/Debug/MockTiDB.h
@@ -37,23 +37,32 @@ class MockTiDB : public ext::Singleton<MockTiDB>
     public:
         Table(const String & database_name, DatabaseID database_id, const String & table_name, TiDB::TableInfo && table_info);
 
-        TableID id() { return table_info.id; }
-        DatabaseID dbID() { return database_id; }
+        TableID id() const { return table_info.id; }
+        DatabaseID dbID() const { return database_id; }
 
         ColumnID allocColumnID() { return ++col_id; }
 
-        bool isPartitionTable() { return table_info.is_partition_table; }
+        bool isPartitionTable() const { return table_info.is_partition_table; }
 
-        std::vector<TableID> getPartitionIDs()
+        std::vector<TableID> getPartitionIDs() const
         {
             std::vector<TableID> partition_ids;
             std::for_each(
                 table_info.partition.definitions.begin(),
                 table_info.partition.definitions.end(),
-                [&](const TiDB::PartitionDefinition & part_def) { partition_ids.emplace_back(part_def.id); });
+                [&](const auto & part_def) { partition_ids.emplace_back(part_def.id); });
             return partition_ids;
         }
 
+        bool existPartitionID(TableID part_id) const
+        {
+            const auto & part_def = find_if(
+                table_info.partition.definitions.begin(),
+                table_info.partition.definitions.end(),
+                [&part_id](const auto & part_def) { return part_def.id == part_id; });
+            return part_def != table_info.partition.definitions.end();
+        }
+
         TiDB::TableInfo table_info;
 
     private:
@@ -89,7 +98,8 @@ class MockTiDB : public ext::Singleton<MockTiDB>
 
     DatabaseID newDataBase(const String & database_name);
 
-    void newPartition(const String & database_name, const String & table_name, TableID partition_id, Timestamp tso, bool);
+    TableID newPartition(const String & database_name, const String & table_name, TableID partition_id, Timestamp tso, bool);
+    TableID newPartition(TableID belong_logical_table, const String & partition_name, Timestamp tso, bool);
 
     void dropPartition(const String & database_name, const String & table_name, TableID partition_id);
 
@@ -135,13 +145,15 @@ class MockTiDB : public ext::Singleton<MockTiDB>
 
     std::unordered_map<TableID, TablePtr> getTables() { return tables_by_id; }
 
-    Int64 getVersion() { return version; }
+    Int64 getVersion() const { return version; }
 
     TableID newTableID() { return table_id_allocator++; }
 
 private:
+    TableID newPartitionImpl(const TablePtr & logical_table, TableID partition_id, const String & partition_name, Timestamp tso, bool is_add_part);
     TablePtr dropTableInternal(Context & context, const String & database_name, const String & table_name, bool drop_regions);
     TablePtr getTableByNameInternal(const String & database_name, const String & table_name);
+    TablePtr getTableByID(TableID table_id);
 
 private:
     std::mutex tables_mutex;
diff --git a/dbms/src/Server/RaftConfigParser.h b/dbms/src/Server/RaftConfigParser.h
index c42304289e2..659a42d76fb 100644
--- a/dbms/src/Server/RaftConfigParser.h
+++ b/dbms/src/Server/RaftConfigParser.h
@@ -37,8 +37,13 @@ struct TiFlashRaftConfig
     std::unordered_set<std::string> ignore_databases{"system"};
     // Actually it is "flash.service_addr"
     std::string flash_server_addr;
+
+    // Use PageStorage V1 for kvstore or not.
+    // TODO: remove this config
     bool enable_compatible_mode = true;
 
+    bool for_unit_test = false;
+
     static constexpr TiDB::StorageEngine DEFAULT_ENGINE = TiDB::StorageEngine::DT;
     TiDB::StorageEngine engine = DEFAULT_ENGINE;
     TiDB::SnapshotApplyMethod snapshot_apply_method = TiDB::SnapshotApplyMethod::DTFile_Directory;
diff --git a/dbms/src/Storages/PathPool.cpp b/dbms/src/Storages/PathPool.cpp
index 465874564f1..1295962e585 100644
--- a/dbms/src/Storages/PathPool.cpp
+++ b/dbms/src/Storages/PathPool.cpp
@@ -230,8 +230,8 @@ void StoragePathPool::clearPSV2ObsoleteData()
 
 void StoragePathPool::rename(const String & new_database, const String & new_table)
 {
-    RUNTIME_CHECK(!new_database.empty() && !new_table.empty(), new_database, new_table);
-    RUNTIME_CHECK(!path_need_database_name);
+    RUNTIME_CHECK(!new_database.empty() && !new_table.empty(), database, table, new_database, new_table);
+    RUNTIME_CHECK(!path_need_database_name, database, table, new_database, new_table);
 
     // The directories for storing table data is not changed, just rename related names.
     std::lock_guard lock{mutex};
diff --git a/dbms/src/Storages/Transaction/PDTiKVClient.h b/dbms/src/Storages/Transaction/PDTiKVClient.h
index e5801cc7fae..111ae9862ed 100644
--- a/dbms/src/Storages/Transaction/PDTiKVClient.h
+++ b/dbms/src/Storages/Transaction/PDTiKVClient.h
@@ -31,11 +31,6 @@
 
 #include <atomic>
 
-// We define a shared ptr here, because TMTContext / SchemaSyncer / IndexReader all need to
-// `share` the resource of cluster.
-using KVClusterPtr = std::shared_ptr<pingcap::kv::Cluster>;
-
-
 namespace DB
 {
 struct PDClientHelper
diff --git a/dbms/src/Storages/Transaction/TMTContext.cpp b/dbms/src/Storages/Transaction/TMTContext.cpp
index c67e3b7755e..09f63b087ff 100644
--- a/dbms/src/Storages/Transaction/TMTContext.cpp
+++ b/dbms/src/Storages/Transaction/TMTContext.cpp
@@ -27,6 +27,8 @@
 #include <TiDB/Schema/TiDBSchemaSyncer.h>
 #include <pingcap/pd/MockPDClient.h>
 
+#include <memory>
+
 namespace DB
 {
 // default batch-read-index timeout is 10_000ms.
@@ -38,6 +40,28 @@ const int64_t DEFAULT_WAIT_REGION_READY_TIMEOUT_SEC = 20 * 60;
 
 const int64_t DEFAULT_READ_INDEX_WORKER_TICK_MS = 10;
 
+static SchemaSyncerPtr createSchemaSyncer(bool exist_pd_addr, bool for_unit_test, const KVClusterPtr & cluster)
+{
+    if (exist_pd_addr)
+    {
+        // product env
+        // Get DBInfo/TableInfo from TiKV, and create table with names `t_${table_id}`
+        return std::static_pointer_cast<SchemaSyncer>(
+            std::make_shared<TiDBSchemaSyncer</*mock_getter*/ false, /*mock_mapper*/ false>>(cluster));
+    }
+    else if (!for_unit_test)
+    {
+        // mock test
+        // Get DBInfo/TableInfo from MockTiDB, and create table with its display names
+        return std::static_pointer_cast<SchemaSyncer>(
+            std::make_shared<TiDBSchemaSyncer</*mock_getter*/ true, /*mock_mapper*/ true>>(cluster));
+    }
+    // unit test.
+    // Get DBInfo/TableInfo from MockTiDB, but create table with names `t_${table_id}`
+    return std::static_pointer_cast<SchemaSyncer>(
+        std::make_shared<TiDBSchemaSyncer</*mock_getter*/ true, /*mock_mapper*/ false>>(cluster));
+}
+
 TMTContext::TMTContext(Context & context_, const TiFlashRaftConfig & raft_config, const pingcap::ClusterConfig & cluster_config)
     : context(context_)
     , kvstore(std::make_shared<KVStore>(context, raft_config.snapshot_apply_method))
@@ -47,9 +71,7 @@ TMTContext::TMTContext(Context & context_, const TiFlashRaftConfig & raft_config
     , cluster(raft_config.pd_addrs.empty() ? std::make_shared<pingcap::kv::Cluster>()
                                            : std::make_shared<pingcap::kv::Cluster>(raft_config.pd_addrs, cluster_config))
     , ignore_databases(raft_config.ignore_databases)
-    , schema_syncer(raft_config.pd_addrs.empty()
-                        ? std::static_pointer_cast<SchemaSyncer>(std::make_shared<TiDBSchemaSyncer</*mock*/ true>>(cluster))
-                        : std::static_pointer_cast<SchemaSyncer>(std::make_shared<TiDBSchemaSyncer</*mock*/ false>>(cluster)))
+    , schema_syncer(createSchemaSyncer(!raft_config.pd_addrs.empty(), raft_config.for_unit_test, cluster))
     , mpp_task_manager(std::make_shared<MPPTaskManager>(
           std::make_unique<MinTSOScheduler>(
               context.getSettingsRef().task_scheduler_thread_soft_limit,
diff --git a/dbms/src/Storages/Transaction/TMTContext.h b/dbms/src/Storages/Transaction/TMTContext.h
index 41cc3b6557c..62e1d574275 100644
--- a/dbms/src/Storages/Transaction/TMTContext.h
+++ b/dbms/src/Storages/Transaction/TMTContext.h
@@ -44,6 +44,10 @@ using GCManagerPtr = std::shared_ptr<GCManager>;
 
 struct TiFlashRaftConfig;
 
+// We define a shared ptr here, because TMTContext / SchemaSyncer / IndexReader all need to
+// `share` the resource of cluster.
+using KVClusterPtr = std::shared_ptr<pingcap::kv::Cluster>;
+
 class TMTContext : private boost::noncopyable
 {
 public:
diff --git a/dbms/src/TestUtils/TiFlashTestBasic.h b/dbms/src/TestUtils/TiFlashTestBasic.h
index 91c2cc1d061..5de50a71424 100644
--- a/dbms/src/TestUtils/TiFlashTestBasic.h
+++ b/dbms/src/TestUtils/TiFlashTestBasic.h
@@ -73,6 +73,10 @@ namespace tests
         FAIL();                                                                        \
     }
 
+/**
+  * GTest related helper functions
+  */
+
 /// helper functions for comparing DataType
 ::testing::AssertionResult DataTypeCompare(
     const char * lhs_expr,
diff --git a/dbms/src/TestUtils/TiFlashTestEnv.cpp b/dbms/src/TestUtils/TiFlashTestEnv.cpp
index 5ef80f48872..b34abb48b3d 100644
--- a/dbms/src/TestUtils/TiFlashTestEnv.cpp
+++ b/dbms/src/TestUtils/TiFlashTestEnv.cpp
@@ -28,6 +28,42 @@ namespace DB::tests
 {
 std::unique_ptr<Context> TiFlashTestEnv::global_context = nullptr;
 
+String TiFlashTestEnv::getTemporaryPath(const std::string_view test_case, bool get_abs)
+{
+    String path = "./tmp/";
+    if (!test_case.empty())
+        path += std::string(test_case);
+
+    Poco::Path poco_path(path);
+    if (get_abs)
+        return poco_path.absolute().toString();
+    else
+        return poco_path.toString();
+}
+
+void TiFlashTestEnv::tryRemovePath(const std::string & path, bool recreate)
+{
+    try
+    {
+        // drop the data on disk
+        Poco::File p(path);
+        if (p.exists())
+        {
+            p.remove(true);
+        }
+
+        // re-create empty directory for testing
+        if (recreate)
+        {
+            p.createDirectories();
+        }
+    }
+    catch (...)
+    {
+        tryLogCurrentException("gtest", fmt::format("while removing dir `{}`", path));
+    }
+}
+
 void TiFlashTestEnv::initializeGlobalContext(Strings testdata_path, PageStorageRunMode ps_run_mode, uint64_t bg_thread_count)
 {
     // set itself as global context
@@ -79,8 +115,9 @@ void TiFlashTestEnv::initializeGlobalContext(Strings testdata_path, PageStorageR
 
     TiFlashRaftConfig raft_config;
 
-    raft_config.ignore_databases = {"default", "system"};
+    raft_config.ignore_databases = {"system"};
     raft_config.engine = TiDB::StorageEngine::DT;
+    raft_config.for_unit_test = true;
     global_context->createTMTContext(raft_config, pingcap::ClusterConfig());
 
     global_context->setDeltaIndexManager(1024 * 1024 * 100 /*100MB*/);
@@ -94,7 +131,12 @@ Context TiFlashTestEnv::getContext(const DB::Settings & settings, Strings testda
     Context context = *global_context;
     context.setGlobalContext(*global_context);
     // Load `testdata_path` as path if it is set.
-    const String root_path = testdata_path.empty() ? (DB::toString(getpid()) + "/" + getTemporaryPath()) : testdata_path[0];
+    const String root_path = [&]() {
+        const auto root_path = testdata_path.empty()
+            ? (DB::toString(getpid()) + "/" + getTemporaryPath("", /*get_abs*/ false))
+            : testdata_path[0];
+        return Poco::Path(root_path).absolute().toString();
+    }();
     if (testdata_path.empty())
         testdata_path.push_back(root_path);
     context.setPath(root_path);
diff --git a/dbms/src/TestUtils/TiFlashTestEnv.h b/dbms/src/TestUtils/TiFlashTestEnv.h
index 636ffa06f95..e599016b87f 100644
--- a/dbms/src/TestUtils/TiFlashTestEnv.h
+++ b/dbms/src/TestUtils/TiFlashTestEnv.h
@@ -27,37 +27,9 @@ namespace DB::tests
 class TiFlashTestEnv
 {
 public:
-    static String getTemporaryPath(const std::string_view test_case = "")
-    {
-        String path = "./tmp/";
-        if (!test_case.empty())
-            path += std::string(test_case);
-
-        return Poco::Path(path).absolute().toString();
-    }
-
-    static void tryRemovePath(const std::string & path, bool recreate = false)
-    {
-        try
-        {
-            // drop the data on disk
-            Poco::File p(path);
-            if (p.exists())
-            {
-                p.remove(true);
-            }
+    static String getTemporaryPath(const std::string_view test_case = "", bool get_abs = true);
 
-            // re-create empty directory for testing
-            if (recreate)
-            {
-                p.createDirectories();
-            }
-        }
-        catch (...)
-        {
-            tryLogCurrentException("gtest", fmt::format("while removing dir `{}`", path));
-        }
-    }
+    static void tryRemovePath(const std::string & path, bool recreate = false);
 
     static std::pair<Strings, Strings> getPathPool(const Strings & testdata_path = {})
     {
diff --git a/dbms/src/TiDB/Schema/SchemaBuilder.cpp b/dbms/src/TiDB/Schema/SchemaBuilder.cpp
index 85287bebc0f..9db15685c74 100644
--- a/dbms/src/TiDB/Schema/SchemaBuilder.cpp
+++ b/dbms/src/TiDB/Schema/SchemaBuilder.cpp
@@ -1361,8 +1361,12 @@ void SchemaBuilder<Getter, NameMapper>::syncAllSchema()
     LOG_INFO(log, "Loaded all schemas.");
 }
 
+// product env
 template struct SchemaBuilder<SchemaGetter, SchemaNameMapper>;
+// mock test
 template struct SchemaBuilder<MockSchemaGetter, MockSchemaNameMapper>;
+// unit test
+template struct SchemaBuilder<MockSchemaGetter, SchemaNameMapper>;
 
 // end namespace
 } // namespace DB
diff --git a/dbms/src/TiDB/Schema/TiDBSchemaSyncer.h b/dbms/src/TiDB/Schema/TiDBSchemaSyncer.h
index 880828c1259..2f71f846490 100644
--- a/dbms/src/TiDB/Schema/TiDBSchemaSyncer.h
+++ b/dbms/src/TiDB/Schema/TiDBSchemaSyncer.h
@@ -27,21 +27,22 @@
 
 namespace DB
 {
+using KVClusterPtr = std::shared_ptr<pingcap::kv::Cluster>;
 namespace ErrorCodes
 {
 extern const int FAIL_POINT_ERROR;
 };
 
-template <bool mock_getter>
+template <bool mock_getter, bool mock_mapper>
 struct TiDBSchemaSyncer : public SchemaSyncer
 {
     using Getter = std::conditional_t<mock_getter, MockSchemaGetter, SchemaGetter>;
 
-    using NameMapper = std::conditional_t<mock_getter, MockSchemaNameMapper, SchemaNameMapper>;
+    using NameMapper = std::conditional_t<mock_mapper, MockSchemaNameMapper, SchemaNameMapper>;
 
     KVClusterPtr cluster;
 
-    const Int64 maxNumberOfDiffs = 100;
+    static constexpr Int64 maxNumberOfDiffs = 100;
 
     Int64 cur_version;
 
@@ -51,8 +52,8 @@ struct TiDBSchemaSyncer : public SchemaSyncer
 
     Poco::Logger * log;
 
-    TiDBSchemaSyncer(KVClusterPtr cluster_)
-        : cluster(cluster_)
+    explicit TiDBSchemaSyncer(KVClusterPtr cluster_)
+        : cluster(std::move(cluster_))
         , cur_version(0)
         , log(&Poco::Logger::get("SchemaSyncer"))
     {}
@@ -73,6 +74,9 @@ struct TiDBSchemaSyncer : public SchemaSyncer
     }
 
     // just for test
+    // It clear all synced database info and reset the `cur_version` to 0.
+    // All info will fetch from the `getter` again the next time
+    // `syncSchemas` is call.
     void reset() override
     {
         std::lock_guard lock(schema_mutex);
diff --git a/dbms/src/TiDB/Schema/tests/gtest_schema_sync.cpp b/dbms/src/TiDB/Schema/tests/gtest_schema_sync.cpp
new file mode 100644
index 00000000000..8acf2c22e62
--- /dev/null
+++ b/dbms/src/TiDB/Schema/tests/gtest_schema_sync.cpp
@@ -0,0 +1,335 @@
+// Copyright 2022 PingCAP, Ltd.
+//
+// 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.
+
+#include <Common/Exception.h>
+#include <Common/TiFlashMetrics.h>
+#include <Databases/IDatabase.h>
+#include <Debug/MockTiDB.h>
+#include <Interpreters/InterpreterDropQuery.h>
+#include <Parsers/ASTDropQuery.h>
+#include <Parsers/IAST.h>
+#include <Storages/ColumnsDescription.h>
+#include <Storages/IManageableStorage.h>
+#include <Storages/Transaction/RegionBlockReader.h>
+#include <Storages/Transaction/TMTContext.h>
+#include <Storages/Transaction/Types.h>
+#include <Storages/registerStorages.h>
+#include <TestUtils/TiFlashTestBasic.h>
+#include <TestUtils/TiFlashTestEnv.h>
+#include <TiDB/Schema/SchemaSyncService.h>
+#include <common/defines.h>
+
+namespace DB
+{
+namespace FailPoints
+{
+extern const char exception_before_rename_table_old_meta_removed[];
+extern const char force_context_path[];
+} // namespace FailPoints
+namespace tests
+{
+
+class SchemaSyncTest : public ::testing::Test
+{
+public:
+    SchemaSyncTest()
+        : global_ctx(TiFlashTestEnv::getGlobalContext())
+    {}
+
+    static void SetUpTestCase()
+    {
+        try
+        {
+            registerStorages();
+        }
+        catch (DB::Exception &)
+        {
+            // Maybe another test has already registed, ignore exception here.
+        }
+
+        FailPointHelper::enableFailPoint(FailPoints::force_context_path);
+    }
+
+    static void TearDownTestCase()
+    {
+        FailPointHelper::disableFailPoint(FailPoints::force_context_path);
+    }
+
+    void SetUp() override
+    {
+        // disable schema sync timer
+        global_ctx.getSchemaSyncService().reset();
+        recreateMetadataPath();
+    }
+
+    void TearDown() override
+    {
+        for (auto & [db_name, db_id] : MockTiDB::instance().getDatabases())
+        {
+            UNUSED(db_id);
+            MockTiDB::instance().dropDB(global_ctx, db_name, false);
+        }
+        for (auto & db : global_ctx.getDatabases())
+        {
+            mustDropSyncedDatabase(db.first);
+        }
+
+        // restore schema sync timer
+        if (!global_ctx.getSchemaSyncService())
+            global_ctx.initializeSchemaSyncService();
+    }
+
+    // Sync schema info from TiDB/MockTiDB to TiFlash
+    void refreshSchema()
+    {
+        auto & flash_ctx = global_ctx.getTMTContext();
+        auto schema_syncer = flash_ctx.getSchemaSyncer();
+        try
+        {
+            schema_syncer->syncSchemas(global_ctx);
+        }
+        catch (Exception & e)
+        {
+            if (e.code() == ErrorCodes::FAIL_POINT_ERROR)
+            {
+                return;
+            }
+            else
+            {
+                throw;
+            }
+        }
+    }
+
+    // Reset the schema syncer to mock TiFlash shutdown
+    void resetSchemas()
+    {
+        auto & flash_ctx = global_ctx.getTMTContext();
+        flash_ctx.getSchemaSyncer()->reset();
+    }
+
+    // Get the TiFlash synced table
+    ManageableStoragePtr mustGetSyncedTable(TableID table_id)
+    {
+        auto & flash_ctx = global_ctx.getTMTContext();
+        auto & flash_storages = flash_ctx.getStorages();
+        auto tbl = flash_storages.get(table_id);
+        RUNTIME_CHECK_MSG(tbl, "Can not find table in TiFlash instance! table_id={}", table_id);
+        return tbl;
+    }
+
+    // Get the TiFlash synced table
+    // `db_name`, `tbl_name` is the name from the TiDB-server side
+    ManageableStoragePtr mustGetSyncedTableByName(const String & db_name, const String & tbl_name)
+    {
+        auto & flash_ctx = global_ctx.getTMTContext();
+        auto & flash_storages = flash_ctx.getStorages();
+        auto mock_tbl = MockTiDB::instance().getTableByName(db_name, tbl_name);
+        auto tbl = flash_storages.get(mock_tbl->id());
+        RUNTIME_CHECK_MSG(tbl, "Can not find table in TiFlash instance! db_name={}, tbl_name={}", db_name, tbl_name);
+        return tbl;
+    }
+
+    /*
+     * Helper methods work with `db_${database_id}`.`t_${table_id}` in the TiFlash side
+     */
+
+    void mustDropSyncedTable(const String & db_idname, const String & tbl_idname)
+    {
+        auto drop_query = std::make_shared<ASTDropQuery>();
+        drop_query->database = db_idname;
+        drop_query->table = tbl_idname;
+        drop_query->if_exists = false;
+        ASTPtr ast_drop_query = drop_query;
+        InterpreterDropQuery drop_interpreter(ast_drop_query, global_ctx);
+        drop_interpreter.execute();
+    }
+
+    void mustDropSyncedDatabase(const String & db_idname)
+    {
+        auto drop_query = std::make_shared<ASTDropQuery>();
+        drop_query->database = db_idname;
+        drop_query->if_exists = false;
+        InterpreterDropQuery drop_interpreter(drop_query, global_ctx);
+        drop_interpreter.execute();
+    }
+
+private:
+    static void recreateMetadataPath()
+    {
+        String path = TiFlashTestEnv::getContext().getPath();
+        auto p = path + "/metadata/";
+        TiFlashTestEnv::tryRemovePath(p, /*recreate=*/true);
+        p = path + "/data/";
+        TiFlashTestEnv::tryRemovePath(p, /*recreate=*/true);
+    }
+
+protected:
+    Context & global_ctx;
+};
+
+TEST_F(SchemaSyncTest, RenameTables)
+try
+{
+    auto pd_client = global_ctx.getTMTContext().getPDClient();
+
+    const String db_name = "mock_db";
+    MockTiDB::instance().newDataBase(db_name);
+
+    auto cols = ColumnsDescription({
+        {"col1", typeFromString("String")},
+        {"col2", typeFromString("Int64")},
+    });
+    // table_name, cols, pk_name
+    std::vector<std::tuple<String, ColumnsDescription, String>> tables{
+        {"t1", cols, ""},
+        {"t2", cols, ""},
+    };
+    MockTiDB::instance().newTables(db_name, tables, pd_client->getTS(), "dt");
+
+    refreshSchema();
+
+    TableID t1_id = mustGetSyncedTableByName(db_name, "t1")->getTableInfo().id;
+    TableID t2_id = mustGetSyncedTableByName(db_name, "t2")->getTableInfo().id;
+
+    // database_name, table_name, new_table_name
+    std::vector<std::tuple<String, String, String>> table_rename_map{
+        {db_name, "t1", "r1"},
+        {db_name, "t2", "r2"},
+    };
+    MockTiDB::instance().renameTables(table_rename_map);
+
+    refreshSchema();
+
+    ASSERT_EQ(mustGetSyncedTable(t1_id)->getTableInfo().name, "r1");
+    ASSERT_EQ(mustGetSyncedTable(t2_id)->getTableInfo().name, "r2");
+}
+CATCH
+
+TEST_F(SchemaSyncTest, RenamePartitionTable)
+try
+{
+    auto pd_client = global_ctx.getTMTContext().getPDClient();
+
+    const String db_name = "mock_db";
+    const String tbl_name = "mock_part_tbl";
+
+    auto cols = ColumnsDescription({
+        {"col1", typeFromString("String")},
+        {"col2", typeFromString("Int64")},
+    });
+
+    MockTiDB::instance().newDataBase(db_name);
+    auto logical_table_id = MockTiDB::instance().newTable(db_name, tbl_name, cols, pd_client->getTS(), "", "dt");
+    auto part1_id = MockTiDB::instance().newPartition(logical_table_id, "red", pd_client->getTS(), /*is_add_part*/ true);
+    auto part2_id = MockTiDB::instance().newPartition(logical_table_id, "blue", pd_client->getTS(), /*is_add_part*/ true);
+
+    // TODO: write some data
+
+    refreshSchema();
+
+    // check partition table are created
+    // TODO: read from partition table
+    {
+        auto logical_tbl = mustGetSyncedTable(logical_table_id);
+        ASSERT_EQ(logical_tbl->getTableInfo().name, tbl_name);
+        auto part1_tbl = mustGetSyncedTable(part1_id);
+        ASSERT_EQ(part1_tbl->getTableInfo().name, fmt::format("t_{}", part1_id));
+        auto part2_tbl = mustGetSyncedTable(part2_id);
+        ASSERT_EQ(part2_tbl->getTableInfo().name, fmt::format("t_{}", part2_id));
+    }
+
+    const String new_tbl_name = "mock_part_tbl_renamed";
+    MockTiDB::instance().renameTable(db_name, tbl_name, new_tbl_name);
+    refreshSchema();
+
+    // check partition table are renamed
+    // TODO: read from renamed partition table
+    {
+        auto logical_tbl = mustGetSyncedTable(logical_table_id);
+        ASSERT_EQ(logical_tbl->getTableInfo().name, new_tbl_name);
+        auto part1_tbl = mustGetSyncedTable(part1_id);
+        ASSERT_EQ(part1_tbl->getTableInfo().name, fmt::format("t_{}", part1_id));
+        auto part2_tbl = mustGetSyncedTable(part2_id);
+        ASSERT_EQ(part2_tbl->getTableInfo().name, fmt::format("t_{}", part2_id));
+    }
+}
+CATCH
+
+TEST_F(SchemaSyncTest, PartitionTableRestart)
+try
+{
+    auto pd_client = global_ctx.getTMTContext().getPDClient();
+
+    const String db_name = "mock_db";
+    const String tbl_name = "mock_part_tbl";
+
+    auto cols = ColumnsDescription({
+        {"col_1", typeFromString("String")},
+        {"col_2", typeFromString("Int64")},
+    });
+
+    auto db_id = MockTiDB::instance().newDataBase(db_name);
+    auto logical_table_id = MockTiDB::instance().newTable(db_name, tbl_name, cols, pd_client->getTS(), "", "dt");
+    auto part1_id = MockTiDB::instance().newPartition(logical_table_id, "red", pd_client->getTS(), /*is_add_part*/ true);
+    auto part2_id = MockTiDB::instance().newPartition(logical_table_id, "green", pd_client->getTS(), /*is_add_part*/ true);
+    auto part3_id = MockTiDB::instance().newPartition(logical_table_id, "blue", pd_client->getTS(), /*is_add_part*/ true);
+
+    refreshSchema();
+    {
+        mustGetSyncedTable(part1_id);
+        mustGetSyncedTable(part2_id);
+        mustGetSyncedTable(part3_id);
+        mustGetSyncedTable(logical_table_id);
+    }
+
+    // schema syncer guarantees logical table creation at last, so there won't be cases
+    // that logical table exists whereas physical table not.
+    // mock that part3 and logical table is not created before restart
+    {
+        mustDropSyncedTable(fmt::format("db_{}", db_id), fmt::format("t_{}", part3_id));
+        mustDropSyncedTable(fmt::format("db_{}", db_id), fmt::format("t_{}", logical_table_id));
+    }
+
+    resetSchemas();
+
+    // add column
+    MockTiDB::instance().addColumnToTable(db_name, tbl_name, NameAndTypePair{"col_3", typeFromString("Nullable(Int8)")}, Field{});
+    const String new_tbl_name = "mock_part_tbl_1";
+    MockTiDB::instance().renameTable(db_name, tbl_name, new_tbl_name);
+    refreshSchema();
+
+    {
+        auto part1_tbl = mustGetSyncedTable(part1_id);
+        ASSERT_EQ(part1_tbl->getTableInfo().name, fmt::format("t_{}", part1_id));
+        auto part2_tbl = mustGetSyncedTable(part2_id);
+        ASSERT_EQ(part2_tbl->getTableInfo().name, fmt::format("t_{}", part2_id));
+        auto part3_tbl = mustGetSyncedTable(part3_id);
+        ASSERT_EQ(part3_tbl->getTableInfo().name, fmt::format("t_{}", part3_id));
+        auto logical_tbl = mustGetSyncedTable(logical_table_id);
+        ASSERT_EQ(logical_tbl->getTableInfo().name, new_tbl_name);
+    }
+
+
+    // drop one partition
+    resetSchemas();
+    MockTiDB::instance().dropPartition(db_name, new_tbl_name, part1_id);
+    refreshSchema();
+    auto part1_tbl = mustGetSyncedTable(part1_id);
+    ASSERT_EQ(part1_tbl->isTombstone(), true);
+}
+CATCH
+
+} // namespace tests
+} // namespace DB
diff --git a/tests/delta-merge-test/raft/schema/partition_table_restart.test b/tests/delta-merge-test/raft/schema/partition_table_restart.test
deleted file mode 100644
index c7a5e488111..00000000000
--- a/tests/delta-merge-test/raft/schema/partition_table_restart.test
+++ /dev/null
@@ -1,55 +0,0 @@
-# Copyright 2022 PingCAP, Ltd.
-#
-# 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.
-
-=> DBGInvoke __enable_schema_sync_service('false')
-
-=> DBGInvoke __drop_tidb_table(default, test)
-=> DBGInvoke __refresh_schemas()
-=> drop table if exists default.test
-=> drop table if exists default.test_9999
-=> drop table if exists default.test_9998
-=> drop table if exists default.test_9997
-
-=> DBGInvoke __mock_tidb_table(default, test, 'col_1 String, col_2 Int64', '', 'dt')
-=> DBGInvoke __mock_tidb_partition(default, test, 9999)
-=> DBGInvoke __mock_tidb_partition(default, test, 9998)
-=> DBGInvoke __mock_tidb_partition(default, test, 9997)
-=> DBGInvoke __refresh_schemas()
-
-=> drop table default.test_9997
-# schema syncer guarantees logical table creation at last, so there won't be cases that logical table exists whereas physical table not.
-=> drop table default.test
-=> DBGInvoke __reset_schemas()
-
-=> DBGInvoke __add_column_to_tidb_table(default, test, 'col_3 Nullable(Int8)')
-=> DBGInvoke __refresh_schemas()
-=> select col_2 from default.test_9997
-
-=> DBGInvoke __reset_schemas()
-=> DBGInvoke __drop_tidb_partition(default, test, 9997)
-=> DBGInvoke __refresh_schemas()
-=> DBGInvoke is_tombstone(default, test_9997)
-┌─is_tombstone(default, test_9997)─┐
-│ true                              │
-└───────────────────────────────────┘
-=> select * from default.test_9999
-
-=> DBGInvoke __drop_tidb_table(default, test)
-=> DBGInvoke __refresh_schemas()
-=> drop table if exists default.test
-=> drop table if exists default.test_9999
-=> drop table if exists default.test_9998
-=> drop table if exists default.test_9997
-
-=> DBGInvoke __enable_schema_sync_service('true')