diff --git a/dbms/src/Debug/DBGInvoker.cpp b/dbms/src/Debug/DBGInvoker.cpp
index ce80ba28ab1..230a9aff95b 100644
--- a/dbms/src/Debug/DBGInvoker.cpp
+++ b/dbms/src/Debug/DBGInvoker.cpp
@@ -40,6 +40,8 @@ DBGInvoker::DBGInvoker()
     regFunc("add_column_to_tidb_table", MockTiDBTable::dbgFuncAddColumnToTiDBTable);
     regFunc("drop_column_from_tidb_table", MockTiDBTable::dbgFuncDropColumnFromTiDBTable);
     regFunc("modify_column_in_tidb_table", MockTiDBTable::dbgFuncModifyColumnInTiDBTable);
+    regFunc("rename_tidb_table", MockTiDBTable::dbgFuncRenameTiDBTable);
+    regFunc("truncate_tidb_table", MockTiDBTable::dbgFuncTruncateTiDBTable);
 
     regFunc("set_flush_threshold", dbgFuncSetFlushThreshold);
 
diff --git a/dbms/src/Debug/MockTiDB.cpp b/dbms/src/Debug/MockTiDB.cpp
index 6f2d4928840..e6692ab59c1 100644
--- a/dbms/src/Debug/MockTiDB.cpp
+++ b/dbms/src/Debug/MockTiDB.cpp
@@ -221,6 +221,36 @@ void MockTiDB::modifyColumnInTable(const String & database_name, const String &
     it->tp = column_info.tp;
 }
 
+void MockTiDB::renameTable(const String & database_name, const String & table_name, const String & new_table_name)
+{
+    std::lock_guard lock(tables_mutex);
+
+    TablePtr table = getTableByNameInternal(database_name, table_name);
+    String qualified_name = database_name + "." + table_name;
+    String new_qualified_name = database_name + "." + new_table_name;
+
+    TableInfo new_table_info = table->table_info;
+    new_table_info.name = new_table_name;
+    auto new_table = std::make_shared<Table>(database_name, new_table_name, std::move(new_table_info));
+
+    tables_by_id[new_table->table_info.id] = new_table;
+    tables_by_name.erase(qualified_name);
+    tables_by_name.emplace(new_qualified_name, new_table);
+}
+
+void MockTiDB::truncateTable(const String & database_name, const String & table_name)
+{
+    std::lock_guard lock(tables_mutex);
+
+    TablePtr table = getTableByNameInternal(database_name, table_name);
+
+    TableID old_table_id = table->table_info.id;
+    table->table_info.id += 1000; // Just big enough is OK.
+
+    tables_by_id.erase(old_table_id);
+    tables_by_id.emplace(table->id(), table);
+}
+
 TablePtr MockTiDB::getTableByName(const String & database_name, const String & table_name)
 {
     std::lock_guard lock(tables_mutex);
diff --git a/dbms/src/Debug/MockTiDB.h b/dbms/src/Debug/MockTiDB.h
index 5adc0dce5eb..f60c0fe8b59 100644
--- a/dbms/src/Debug/MockTiDB.h
+++ b/dbms/src/Debug/MockTiDB.h
@@ -74,6 +74,10 @@ class MockTiDB : public ext::singleton<MockTiDB>
 
     void modifyColumnInTable(const String & database_name, const String & table_name, const NameAndTypePair & column);
 
+    void renameTable(const String & database_name, const String & table_name, const String & new_table_name);
+
+    void truncateTable(const String & database_name, const String & table_name);
+
     TablePtr getTableByName(const String & database_name, const String & table_name);
 
     void traverseTables(std::function<void(TablePtr)> f);
diff --git a/dbms/src/Debug/dbgFuncMockTiDBTable.cpp b/dbms/src/Debug/dbgFuncMockTiDBTable.cpp
index efd17c4dfef..2db846688b9 100644
--- a/dbms/src/Debug/dbgFuncMockTiDBTable.cpp
+++ b/dbms/src/Debug/dbgFuncMockTiDBTable.cpp
@@ -230,4 +230,35 @@ void MockTiDBTable::dbgFuncModifyColumnInTiDBTable(DB::Context & context, const
     output(ss.str());
 }
 
+void MockTiDBTable::dbgFuncRenameTiDBTable(Context & /*context*/, const ASTs & args, DBGInvoker::Printer output)
+{
+    if (args.size() != 3)
+        throw Exception("Args not matched, should be: database-name, table-name, new-table-name", ErrorCodes::BAD_ARGUMENTS);
+
+    const String & database_name = typeid_cast<const ASTIdentifier &>(*args[0]).name;
+    const String & table_name = typeid_cast<const ASTIdentifier &>(*args[1]).name;
+    const String & new_table_name = typeid_cast<const ASTIdentifier &>(*args[2]).name;
+
+    MockTiDB::instance().renameTable(database_name, table_name, new_table_name);
+
+    std::stringstream ss;
+    ss << "renamed table " << database_name << "." << table_name << " to " << database_name << "." << new_table_name;
+    output(ss.str());
+}
+
+void MockTiDBTable::dbgFuncTruncateTiDBTable(Context & /*context*/, const ASTs & args, DBGInvoker::Printer output)
+{
+    if (args.size() != 2)
+        throw Exception("Args not matched, should be: database-name, table-name", ErrorCodes::BAD_ARGUMENTS);
+
+    const String & database_name = typeid_cast<const ASTIdentifier &>(*args[0]).name;
+    const String & table_name = typeid_cast<const ASTIdentifier &>(*args[1]).name;
+
+    MockTiDB::instance().truncateTable(database_name, table_name);
+
+    std::stringstream ss;
+    ss << "truncated table " << database_name << "." << table_name;
+    output(ss.str());
+}
+
 } // namespace DB
diff --git a/dbms/src/Debug/dbgFuncMockTiDBTable.h b/dbms/src/Debug/dbgFuncMockTiDBTable.h
index 27414b5582d..92b29a7589a 100644
--- a/dbms/src/Debug/dbgFuncMockTiDBTable.h
+++ b/dbms/src/Debug/dbgFuncMockTiDBTable.h
@@ -48,6 +48,16 @@ struct MockTiDBTable
     // Usage:
     //   ./storages-client.sh "DBGInvoke modify_column_in_tidb_table(database_name, table_name, 'col type')"
     static void dbgFuncModifyColumnInTiDBTable(Context & context, const ASTs & args, DBGInvoker::Printer output);
+
+    // Rename a TiDB table.
+    // Usage:
+    //   ./storages-client.sh "DBGInvoke rename_tidb_table(database_name, table_name, new_table)"
+    static void dbgFuncRenameTiDBTable(Context & context, const ASTs & args, DBGInvoker::Printer output);
+
+    // Truncate a TiDB table.
+    // Usage:
+    //   ./storages-client.sh "DBGInvoke truncate_tidb_table(database_name, table_name)"
+    static void dbgFuncTruncateTiDBTable(Context & context, const ASTs & args, DBGInvoker::Printer output);
 };
 
 } // namespace DB
diff --git a/tests/mutable-test/txn_schema/rename_on_read.test b/tests/mutable-test/txn_schema/rename_on_read.test
new file mode 100644
index 00000000000..ab1b1148c0d
--- /dev/null
+++ b/tests/mutable-test/txn_schema/rename_on_read.test
@@ -0,0 +1,25 @@
+=> DBGInvoke __enable_schema_sync_service('false')
+
+=> DBGInvoke __drop_tidb_table(default, test)
+=> drop table if exists default.test
+
+=> DBGInvoke __drop_tidb_table(default, test1)
+=> drop table if exists default.test1
+
+=> DBGInvoke __set_flush_threshold(1000000, 1000000)
+=> DBGInvoke __mock_schema_syncer('true')
+
+=> DBGInvoke __mock_tidb_table(default, test, 'col_1 String')
+=> DBGInvoke __refresh_schemas()
+=> select * from default.test
+=> DBGInvoke __rename_tidb_table(default, test, test1)
+=> select * from default.test
+=> select * from default.test " --schema_version "100
+Received exception from server (version {#WORD}):
+Code: 60. DB::Exception: Received from {#WORD} DB::Exception: Table default.test doesn't exist..
+=> select * from default.test1
+=> select * from default.test1 " --schema_version "100
+
+=> DBGInvoke __drop_tidb_table(default, test1)
+=> drop table if exists default.test1
+=> DBGInvoke __enable_schema_sync_service('true')
diff --git a/tests/mutable-test/txn_schema/rename_on_write.test b/tests/mutable-test/txn_schema/rename_on_write.test
new file mode 100644
index 00000000000..3b031c45f09
--- /dev/null
+++ b/tests/mutable-test/txn_schema/rename_on_write.test
@@ -0,0 +1,30 @@
+=> DBGInvoke __enable_schema_sync_service('false')
+
+=> DBGInvoke __drop_tidb_table(default, test)
+=> drop table if exists default.test
+
+=> DBGInvoke __drop_tidb_table(default, test1)
+=> drop table if exists default.test1
+
+=> DBGInvoke __set_flush_threshold(1000000, 1000000)
+=> DBGInvoke __mock_schema_syncer('true')
+
+=> DBGInvoke __mock_tidb_table(default, test, 'col_1 String')
+=> DBGInvoke __refresh_schemas()
+=> DBGInvoke __put_region(4, 0, 100, default, test)
+=> select col_1 from default.test
+=> DBGInvoke __add_column_to_tidb_table(default, test, 'col_2 Nullable(Int8)')
+=> DBGInvoke __raft_insert_row(default, test, 4, 50, 'test1', 1)
+=> DBGInvoke __rename_tidb_table(default, test, test1)
+=> DBGInvoke __try_flush_region(4)
+=> select * from default.test
+Received exception from server (version {#WORD}):
+Code: 60. DB::Exception: Received from {#WORD} DB::Exception: Table default.test doesn't exist..
+=> select * from default.test1
+┌─col_1─┬─_tidb_rowid─┬─col_2─┐
+│ test1 │          50 │     1 │
+└───────┴─────────────┴───────┘
+
+=> DBGInvoke __drop_tidb_table(default, test1)
+=> drop table if exists default.test1
+=> DBGInvoke __enable_schema_sync_service('true')
diff --git a/tests/mutable-test/txn_schema/truncate_on_read.test b/tests/mutable-test/txn_schema/truncate_on_read.test
new file mode 100644
index 00000000000..9aa469e118b
--- /dev/null
+++ b/tests/mutable-test/txn_schema/truncate_on_read.test
@@ -0,0 +1,26 @@
+=> DBGInvoke __enable_schema_sync_service('false')
+
+=> DBGInvoke __drop_tidb_table(default, test)
+=> drop table if exists default.test
+
+=> DBGInvoke __set_flush_threshold(1000000, 1000000)
+=> DBGInvoke __mock_schema_syncer('true')
+
+=> DBGInvoke __mock_tidb_table(default, test, 'col_1 String')
+=> DBGInvoke __refresh_schemas()
+=> DBGInvoke __put_region(4, 0, 100, default, test)
+=> DBGInvoke __raft_insert_row(default, test, 4, 50, 'test1')
+=> select * from default.test
+┌─col_1─┬─_tidb_rowid─┐
+│ test1 │          50 │
+└───────┴─────────────┘
+=> DBGInvoke __truncate_tidb_table(default, test)
+=> select * from default.test
+┌─col_1─┬─_tidb_rowid─┐
+│ test1 │          50 │
+└───────┴─────────────┘
+=> select * from default.test " --schema_version "100
+
+=> DBGInvoke __drop_tidb_table(default, test)
+=> drop table if exists default.test
+=> DBGInvoke __enable_schema_sync_service('true')
diff --git a/tests/mutable-test/txn_schema/truncate_on_write.test b/tests/mutable-test/txn_schema/truncate_on_write.test
new file mode 100644
index 00000000000..62cb594151b
--- /dev/null
+++ b/tests/mutable-test/txn_schema/truncate_on_write.test
@@ -0,0 +1,25 @@
+=> DBGInvoke __enable_schema_sync_service('false')
+
+=> DBGInvoke __drop_tidb_table(default, test)
+=> drop table if exists default.test
+
+=> DBGInvoke __set_flush_threshold(1000000, 1000000)
+=> DBGInvoke __mock_schema_syncer('true')
+
+=> DBGInvoke __mock_tidb_table(default, test, 'col_1 String')
+=> DBGInvoke __refresh_schemas()
+=> DBGInvoke __put_region(4, 0, 100, default, test)
+=> DBGInvoke __raft_insert_row(default, test, 4, 50, 'test1')
+=> select col_1 from default.test
+┌─col_1─┐
+│ test1 │
+└───────┘
+=> DBGInvoke __add_column_to_tidb_table(default, test, 'col_2 Nullable(Int8)')
+=> DBGInvoke __raft_insert_row(default, test, 4, 51, 'test1', 1)
+=> DBGInvoke __truncate_tidb_table(default, test)
+=> DBGInvoke __try_flush_region(4)
+=> select * from default.test
+
+=> DBGInvoke __drop_tidb_table(default, test)
+=> drop table if exists default.test
+=> DBGInvoke __enable_schema_sync_service('true')