diff --git a/dbms/pch-stl.h b/dbms/pch-stl.h
index 6037372b333..01b3123650d 100644
--- a/dbms/pch-stl.h
+++ b/dbms/pch-stl.h
@@ -18,14 +18,17 @@
 #include <cassert>
 #include <chrono>
 #include <cstdint>
+#include <fstream>
 #include <functional>
 #include <future>
 #include <map>
 #include <memory>
 #include <mutex>
 #include <ostream>
+#include <queue>
+#include <random>
 #include <shared_mutex>
 #include <string>
 #include <unordered_map>
 #include <unordered_set>
-#include <vector>
+#include <vector>
\ No newline at end of file
diff --git a/dbms/src/Interpreters/Context.h b/dbms/src/Interpreters/Context.h
index 63aefcbece9..5d5c39263c6 100644
--- a/dbms/src/Interpreters/Context.h
+++ b/dbms/src/Interpreters/Context.h
@@ -14,7 +14,6 @@
 
 #pragma once
 
-#include <Core/NamesAndTypes.h>
 #include <Core/Types.h>
 #include <IO/CompressionSettings.h>
 #include <Interpreters/ClientInfo.h>
diff --git a/dbms/src/Storages/Transaction/KVStore.cpp b/dbms/src/Storages/Transaction/KVStore.cpp
index 2b82220cbc4..318a04c6ed9 100644
--- a/dbms/src/Storages/Transaction/KVStore.cpp
+++ b/dbms/src/Storages/Transaction/KVStore.cpp
@@ -25,6 +25,7 @@
 #include <Storages/Transaction/ReadIndexWorker.h>
 #include <Storages/Transaction/Region.h>
 #include <Storages/Transaction/RegionExecutionResult.h>
+#include <Storages/Transaction/RegionPersister.h>
 #include <Storages/Transaction/RegionTable.h>
 #include <Storages/Transaction/TMTContext.h>
 #include <common/likely.h>
@@ -38,15 +39,15 @@ extern const int TABLE_IS_DROPPED;
 } // namespace ErrorCodes
 
 KVStore::KVStore(Context & context, TiDB::SnapshotApplyMethod snapshot_apply_method_)
-    : region_persister(context, region_manager)
+    : region_persister(std::make_unique<RegionPersister>(context, region_manager))
     , raft_cmd_res(std::make_unique<RaftCommandResult>())
     , snapshot_apply_method(snapshot_apply_method_)
     , log(&Poco::Logger::get("KVStore"))
+    , region_compact_log_period(120)
+    , region_compact_log_min_rows(40 * 1024)
+    , region_compact_log_min_bytes(32 * 1024 * 1024)
 {
     // default config about compact-log: period 120s, rows 40k, bytes 32MB.
-    REGION_COMPACT_LOG_PERIOD = 120;
-    REGION_COMPACT_LOG_MIN_ROWS = 40 * 1024;
-    REGION_COMPACT_LOG_MIN_BYTES = 32 * 1024 * 1024;
 }
 
 void KVStore::restore(const TiFlashRaftProxyHelper * proxy_helper)
@@ -55,7 +56,7 @@ void KVStore::restore(const TiFlashRaftProxyHelper * proxy_helper)
     auto manage_lock = genRegionWriteLock(task_lock);
 
     this->proxy_helper = proxy_helper;
-    manage_lock.regions = region_persister.restore(proxy_helper);
+    manage_lock.regions = region_persister->restore(proxy_helper);
 
     LOG_FMT_INFO(log, "Restored {} regions", manage_lock.regions.size());
 
@@ -166,7 +167,7 @@ void KVStore::tryPersist(RegionID region_id)
     if (region)
     {
         LOG_FMT_INFO(log, "Try to persist {}", region->toString(false));
-        region_persister.persist(*region);
+        region_persister->persist(*region);
         LOG_FMT_INFO(log, "After persisted {}, cache {} bytes", region->toString(false), region->dataSize());
     }
 }
@@ -182,7 +183,7 @@ void KVStore::gcRegionPersistedCache(Seconds gc_persist_period)
     if (now < (last_gc_time.load() + gc_persist_period))
         return;
     last_gc_time = now;
-    region_persister.gc();
+    region_persister->gc();
 }
 
 void KVStore::removeRegion(RegionID region_id, bool remove_data, RegionTable & region_table, const KVStoreTaskLock & task_lock, const RegionTaskLock & region_lock)
@@ -203,7 +204,7 @@ void KVStore::removeRegion(RegionID region_id, bool remove_data, RegionTable & r
         }
     }
 
-    region_persister.drop(region_id, region_lock);
+    region_persister->drop(region_id, region_lock);
     LOG_FMT_INFO(log, "Persisted [region {}] deleted", region_id);
 
     region_table.removeRegion(region_id, remove_data, region_lock);
@@ -306,9 +307,9 @@ void KVStore::handleDestroy(UInt64 region_id, TMTContext & tmt, const KVStoreTas
 
 void KVStore::setRegionCompactLogConfig(UInt64 sec, UInt64 rows, UInt64 bytes)
 {
-    REGION_COMPACT_LOG_PERIOD = sec;
-    REGION_COMPACT_LOG_MIN_ROWS = rows;
-    REGION_COMPACT_LOG_MIN_BYTES = bytes;
+    region_compact_log_period = sec;
+    region_compact_log_min_rows = rows;
+    region_compact_log_min_bytes = bytes;
 
     LOG_FMT_INFO(
         log,
@@ -321,7 +322,7 @@ void KVStore::setRegionCompactLogConfig(UInt64 sec, UInt64 rows, UInt64 bytes)
 void KVStore::persistRegion(const Region & region, const RegionTaskLock & region_task_lock, const char * caller)
 {
     LOG_FMT_INFO(log, "Start to persist {}, cache size: {} bytes for `{}`", region.toString(true), region.dataSize(), caller);
-    region_persister.persist(region, region_task_lock);
+    region_persister->persist(region, region_task_lock);
     LOG_FMT_DEBUG(log, "Persist {} done", region.toString(false));
 }
 
@@ -362,8 +363,8 @@ EngineStoreApplyRes KVStore::handleUselessAdminRaftCmd(
 
             LOG_FMT_DEBUG(log, "{} approx mem cache info: rows {}, bytes {}", curr_region.toString(false), rows, size_bytes);
 
-            if (rows >= REGION_COMPACT_LOG_MIN_ROWS.load(std::memory_order_relaxed)
-                || size_bytes >= REGION_COMPACT_LOG_MIN_BYTES.load(std::memory_order_relaxed))
+            if (rows >= region_compact_log_min_rows.load(std::memory_order_relaxed)
+                || size_bytes >= region_compact_log_min_bytes.load(std::memory_order_relaxed))
             {
                 // if rows or bytes more than threshold, flush cache and perist mem data.
                 return true;
@@ -372,7 +373,7 @@ EngineStoreApplyRes KVStore::handleUselessAdminRaftCmd(
             {
                 // if thhere is little data in mem, wait until time interval reached threshold.
                 // use random period so that lots of regions will not be persisted at same time.
-                auto compact_log_period = std::rand() % REGION_COMPACT_LOG_PERIOD.load(std::memory_order_relaxed); // NOLINT
+                auto compact_log_period = std::rand() % region_compact_log_period.load(std::memory_order_relaxed); // NOLINT
                 return !(curr_region.lastCompactLogTime() + Seconds{compact_log_period} > Clock::now());
             }
         }
@@ -765,4 +766,9 @@ KVStore::~KVStore()
     releaseReadIndexWorkers();
 }
 
+FileUsageStatistics KVStore::getFileUsageStatistics() const
+{
+    return region_persister->getFileUsageStatistics();
+}
+
 } // namespace DB
diff --git a/dbms/src/Storages/Transaction/KVStore.h b/dbms/src/Storages/Transaction/KVStore.h
index 8673cae3ff3..bb45e65d18b 100644
--- a/dbms/src/Storages/Transaction/KVStore.h
+++ b/dbms/src/Storages/Transaction/KVStore.h
@@ -16,20 +16,19 @@
 
 #include <Storages/Transaction/RegionDataRead.h>
 #include <Storages/Transaction/RegionManager.h>
-#include <Storages/Transaction/RegionPersister.h>
 #include <Storages/Transaction/StorageEngineType.h>
 
-
 namespace TiDB
 {
 struct TableInfo;
 }
 namespace DB
 {
+class Context;
 namespace RegionBench
 {
 extern void concurrentBatchInsert(const TiDB::TableInfo &, Int64, Int64, Int64, UInt64, UInt64, Context &);
-}
+} // namespace RegionBench
 namespace DM
 {
 enum class FileConvertJobType;
@@ -40,7 +39,6 @@ namespace tests
 class RegionKVStoreTest;
 }
 
-class Context;
 class IAST;
 using ASTPtr = std::shared_ptr<IAST>;
 using ASTs = std::vector<ASTPtr>;
@@ -71,6 +69,8 @@ using RegionPreDecodeBlockDataPtr = std::unique_ptr<RegionPreDecodeBlockData>;
 class ReadIndexWorkerManager;
 using BatchReadIndexRes = std::vector<std::pair<kvrpcpb::ReadIndexResponse, uint64_t>>;
 class ReadIndexStressTest;
+struct FileUsageStatistics;
+class RegionPersister;
 
 /// TODO: brief design document.
 class KVStore final : private boost::noncopyable
@@ -157,10 +157,7 @@ class KVStore final : private boost::noncopyable
 
     ~KVStore();
 
-    FileUsageStatistics getFileUsageStatistics() const
-    {
-        return region_persister.getFileUsageStatistics();
-    }
+    FileUsageStatistics getFileUsageStatistics() const;
 
 private:
     friend class MockTiDB;
@@ -229,7 +226,7 @@ class KVStore final : private boost::noncopyable
 private:
     RegionManager region_manager;
 
-    RegionPersister region_persister;
+    std::unique_ptr<RegionPersister> region_persister;
 
     std::atomic<Timepoint> last_gc_time = Timepoint::min();
 
@@ -242,9 +239,9 @@ class KVStore final : private boost::noncopyable
 
     Poco::Logger * log;
 
-    std::atomic<UInt64> REGION_COMPACT_LOG_PERIOD;
-    std::atomic<UInt64> REGION_COMPACT_LOG_MIN_ROWS;
-    std::atomic<UInt64> REGION_COMPACT_LOG_MIN_BYTES;
+    std::atomic<UInt64> region_compact_log_period;
+    std::atomic<UInt64> region_compact_log_min_rows;
+    std::atomic<UInt64> region_compact_log_min_bytes;
 
     mutable std::mutex bg_gc_region_data_mutex;
     std::list<RegionDataReadInfoList> bg_gc_region_data;
diff --git a/dbms/src/Storages/Transaction/ProxyFFI.cpp b/dbms/src/Storages/Transaction/ProxyFFI.cpp
index cc7d1e10a49..8a40ca9b15e 100644
--- a/dbms/src/Storages/Transaction/ProxyFFI.cpp
+++ b/dbms/src/Storages/Transaction/ProxyFFI.cpp
@@ -24,6 +24,8 @@
 #include <Storages/Transaction/TMTContext.h>
 #include <kvproto/diagnosticspb.pb.h>
 
+#include <ext/scope_guard.h>
+
 #define CHECK_PARSE_PB_BUFF_IMPL(n, a, b, c)                                              \
     do                                                                                    \
     {                                                                                     \