diff --git a/dbms/src/Debug/dbgFuncRegion.cpp b/dbms/src/Debug/dbgFuncRegion.cpp
index 0feb125da7f..395cb8b6771 100644
--- a/dbms/src/Debug/dbgFuncRegion.cpp
+++ b/dbms/src/Debug/dbgFuncRegion.cpp
@@ -220,7 +220,7 @@ std::string getRegionKeyString(const TiKVRange::Handle s, const TiKVKey & k)
     {
         if (s.type != TiKVHandle::HandleIDType::NORMAL)
         {
-            auto raw_key = k.empty() ? "" : RecordKVFormat::decodeTiKVKey(k);
+            auto raw_key = k.empty() ? DecodedTiKVKey() : RecordKVFormat::decodeTiKVKey(k);
             bool is_record = RecordKVFormat::isRecord(raw_key);
             std::stringstream ss;
             if (is_record)
@@ -285,6 +285,7 @@ void dbgFuncDumpAllRegion(Context & context, TableID table_id, bool ignore_none,
             ss << " [none], ";
         else
             ss << " ranges: [" << range.first.toString() << ", " << range.second.toString() << "), ";
+        ss << "state: " << raft_serverpb::PeerState_Name(region->peerState());
         if (auto s = region->dataInfo(); s.size() > 2)
             ss << ", " << s;
         output(ss.str());
diff --git a/dbms/src/Debug/dbgTools.cpp b/dbms/src/Debug/dbgTools.cpp
index 5e40faabd9a..e4a0f2ac5ee 100644
--- a/dbms/src/Debug/dbgTools.cpp
+++ b/dbms/src/Debug/dbgTools.cpp
@@ -139,7 +139,8 @@ void insert(const TiDB::TableInfo & table_info, RegionID region_id, HandleID han
     RegionPtr region = tmt.getKVStore()->getRegion(region_id);
 
     // Using the region meta's table ID rather than table_info's, as this could be a partition table so that the table ID should be partition ID.
-    TableID table_id = RecordKVFormat::getTableId(region->getRange().first);
+    const auto range = region->getRange();
+    TableID table_id = RecordKVFormat::getTableId(range->rawKeys().first);
 
     TiKVKey key = RecordKVFormat::genKey(table_id, handle_id);
     TiKVValue value = RecordKVFormat::EncodeRow(table_info, fields);
@@ -303,8 +304,8 @@ void concurrentBatchInsert(const TiDB::TableInfo & table_info, Int64 concurrent_
     HandleID curr_max_handle_id = 0;
     tmt.getKVStore()->traverseRegions([&](const RegionID region_id, const RegionPtr & region) {
         curr_max_region_id = (curr_max_region_id == InvalidRegionID) ? region_id : std::max<RegionID>(curr_max_region_id, region_id);
-        auto range = region->getRange();
-        curr_max_handle_id = std::max(RecordKVFormat::getHandle(range.second), curr_max_handle_id);
+        const auto range = region->getRange();
+        curr_max_handle_id = std::max(RecordKVFormat::getHandle(range->rawKeys().second), curr_max_handle_id);
     });
 
     Int64 key_num_each_region = flush_num * batch_num;
@@ -349,9 +350,8 @@ Int64 concurrentRangeOperate(
     Int64 tol = 0;
     for (auto region : regions)
     {
-        auto [start_key, end_key] = region->getRange();
-        auto ss = TiKVRange::getRangeHandle<true>(start_key, table_info.id);
-        auto ee = TiKVRange::getRangeHandle<false>(end_key, table_info.id);
+        const auto range = region->getRange();
+        const auto & [ss, ee] = range->getHandleRangeByTable(table_info.id);
         TiKVRange::Handle handle_begin = std::max<TiKVRange::Handle>(ss, start_handle);
         TiKVRange::Handle handle_end = std::min<TiKVRange::Handle>(ee, end_handle);
         if (handle_end <= handle_begin)
diff --git a/dbms/src/Storages/Transaction/PartitionStreams.cpp b/dbms/src/Storages/Transaction/PartitionStreams.cpp
index 2c36ec331b3..c5abb2b189b 100644
--- a/dbms/src/Storages/Transaction/PartitionStreams.cpp
+++ b/dbms/src/Storages/Transaction/PartitionStreams.cpp
@@ -143,7 +143,7 @@ std::tuple<Block, RegionTable::RegionReadStatus> RegionTable::readBlockByRegion(
             if (version != region_version || conf_ver != conf_version)
                 return {Block(), VERSION_ERROR};
 
-            handle_range = TiKVRange::getHandleRangeByTable(key_range, table_info.id);
+            handle_range = key_range->getHandleRangeByTable(table_info.id);
         }
 
         /// Deal with locks.
diff --git a/dbms/src/Storages/Transaction/Region.cpp b/dbms/src/Storages/Transaction/Region.cpp
index 2e2e6e929ce..78a0ed24a6f 100644
--- a/dbms/src/Storages/Transaction/Region.cpp
+++ b/dbms/src/Storages/Transaction/Region.cpp
@@ -100,7 +100,8 @@ RegionPtr Region::splitInto(RegionMeta meta)
     else
         new_region = std::make_shared<Region>(std::move(meta));
 
-    data.splitInto(new_region->getRange(), new_region->data);
+    const auto range = new_region->getRange();
+    data.splitInto(range->keys(), new_region->data);
 
     return new_region;
 }
@@ -468,7 +469,7 @@ std::string Region::toString(bool dump_status) const { return meta.toString(dump
 
 enginepb::CommandResponse Region::toCommandResponse() const { return meta.toCommandResponse(); }
 
-RegionRange Region::getRange() const { return meta.getRange(); }
+ImutRegionRangePtr Region::getRange() const { return meta.getRange(); }
 
 UInt64 Region::learnerRead()
 {
@@ -494,10 +495,7 @@ UInt64 Region::version() const { return meta.version(); }
 
 UInt64 Region::confVer() const { return meta.confVer(); }
 
-HandleRange<HandleID> Region::getHandleRangeByTable(TableID table_id) const
-{
-    return TiKVRange::getHandleRangeByTable(getRange(), table_id);
-}
+HandleRange<HandleID> Region::getHandleRangeByTable(TableID table_id) const { return getRange()->getHandleRangeByTable(table_id); }
 
 void Region::assignRegion(Region && new_region)
 {
@@ -595,7 +593,8 @@ RegionRaftCommandDelegate & Region::makeRaftCommandDelegate(const KVStoreTaskLoc
 
 void Region::compareAndCompleteSnapshot(const Timestamp safe_point, const Region & source_region)
 {
-    const auto & [start_key, end_key] = getRange();
+    const auto range = getRange();
+    const auto & [start_key, end_key] = range->keys();
     std::unordered_map<TableID, HandleMap> handle_maps;
     {
         std::shared_lock<std::shared_mutex> source_lock(source_region.mutex);
@@ -645,7 +644,7 @@ void Region::doDeleteRange(const std::string & cf, const TiKVKey & start_key, co
     return data.deleteRange(type, start_key, end_key);
 }
 
-std::tuple<RegionVersion, RegionVersion, RegionRange> Region::dumpVersionRange() const { return meta.dumpVersionRange(); }
+std::tuple<RegionVersion, RegionVersion, ImutRegionRangePtr> Region::dumpVersionRange() const { return meta.dumpVersionRange(); }
 
 void Region::tryPreDecodeTiKVValue()
 {
diff --git a/dbms/src/Storages/Transaction/Region.h b/dbms/src/Storages/Transaction/Region.h
index eb5d8eea1f9..3c20f95a421 100644
--- a/dbms/src/Storages/Transaction/Region.h
+++ b/dbms/src/Storages/Transaction/Region.h
@@ -112,7 +112,7 @@ class Region : public std::enable_shared_from_this<Region>
     static RegionPtr deserialize(ReadBuffer & buf, const RegionClientCreateFunc * region_client_create = nullptr);
 
     RegionID id() const;
-    RegionRange getRange() const;
+    ImutRegionRangePtr getRange() const;
 
     enginepb::CommandResponse toCommandResponse() const;
     std::string toString(bool dump_status = true) const;
@@ -150,7 +150,7 @@ class Region : public std::enable_shared_from_this<Region>
     RegionVersion confVer() const;
 
     /// version, conf_version, range
-    std::tuple<RegionVersion, RegionVersion, RegionRange> dumpVersionRange() const;
+    std::tuple<RegionVersion, RegionVersion, ImutRegionRangePtr> dumpVersionRange() const;
 
     HandleRange<HandleID> getHandleRangeByTable(TableID table_id) const;
 
diff --git a/dbms/src/Storages/Transaction/RegionMeta.cpp b/dbms/src/Storages/Transaction/RegionMeta.cpp
index f5cc3d5f330..13bf2d4854d 100644
--- a/dbms/src/Storages/Transaction/RegionMeta.cpp
+++ b/dbms/src/Storages/Transaction/RegionMeta.cpp
@@ -16,7 +16,7 @@ std::tuple<size_t, UInt64> RegionMeta::serialize(WriteBuffer & buf) const
     size += writeBinary2(peer, buf);
     size += writeBinary2(apply_state, buf);
     size += writeBinary2(applied_term, buf);
-    size += writeBinary2(region_state, buf);
+    size += writeBinary2(region_state.getBase(), buf);
     return {size, apply_state.applied_index()};
 }
 
@@ -54,7 +54,7 @@ pingcap::kv::RegionVerID RegionMeta::getRegionVerID() const
     std::lock_guard<std::mutex> lock(mutex);
 
     return pingcap::kv::RegionVerID{
-        regionId(), region_state.region().region_epoch().conf_ver(), region_state.region().region_epoch().version()};
+        regionId(), region_state.getRegion().region_epoch().conf_ver(), region_state.getRegion().region_epoch().version()};
 }
 
 raft_serverpb::RaftApplyState RegionMeta::getApplyState() const
@@ -68,7 +68,7 @@ void RegionMeta::doSetRegion(const metapb::Region & region)
     if (regionId() != region.id())
         throw Exception("[RegionMeta::doSetRegion] region id is not equal, should not happen", ErrorCodes::LOGICAL_ERROR);
 
-    *region_state.mutable_region() = region;
+    region_state.setRegion(region);
 }
 
 void RegionMeta::setApplied(UInt64 index, UInt64 term)
@@ -117,10 +117,10 @@ RegionMeta::RegionMeta(RegionMeta && rhs) : region_id(rhs.regionId())
     region_state = std::move(rhs.region_state);
 }
 
-RegionRange RegionMeta::getRange() const
+ImutRegionRangePtr RegionMeta::getRange() const
 {
     std::lock_guard<std::mutex> lock(mutex);
-    return {TiKVKey::copyFrom(region_state.region().start_key()), TiKVKey::copyFrom(region_state.region().end_key())};
+    return region_state.getRange();
 }
 
 std::string RegionMeta::toString(bool dump_status) const
@@ -145,13 +145,13 @@ std::string RegionMeta::toString(bool dump_status) const
 raft_serverpb::PeerState RegionMeta::peerState() const
 {
     std::lock_guard<std::mutex> lock(mutex);
-    return region_state.state();
+    return region_state.getState();
 }
 
 void RegionMeta::setPeerState(const raft_serverpb::PeerState peer_state_)
 {
     std::lock_guard<std::mutex> lock(mutex);
-    region_state.set_state(peer_state_);
+    region_state.setState(peer_state_);
 }
 
 void RegionMeta::waitIndex(UInt64 index) const
@@ -168,19 +168,19 @@ bool RegionMeta::checkIndex(UInt64 index) const
 
 bool RegionMeta::doCheckIndex(UInt64 index) const
 {
-    return region_state.state() == raft_serverpb::PeerState::Tombstone || apply_state.applied_index() >= index;
+    return region_state.getState() == raft_serverpb::PeerState::Tombstone || apply_state.applied_index() >= index;
 }
 
 UInt64 RegionMeta::version() const
 {
     std::lock_guard<std::mutex> lock(mutex);
-    return region_state.region().region_epoch().version();
+    return region_state.getRegion().region_epoch().version();
 }
 
 UInt64 RegionMeta::confVer() const
 {
     std::lock_guard<std::mutex> lock(mutex);
-    return region_state.region().region_epoch().conf_ver();
+    return region_state.getRegion().region_epoch().conf_ver();
 }
 
 void RegionMeta::assignRegionMeta(RegionMeta && rhs)
@@ -199,38 +199,39 @@ void RegionMeta::assignRegionMeta(RegionMeta && rhs)
 void MetaRaftCommandDelegate::execChangePeer(
     const raft_cmdpb::AdminRequest & request, const raft_cmdpb::AdminResponse & response, UInt64 index, UInt64 term)
 {
+    std::lock_guard<std::mutex> lock(mutex);
+
     const auto & change_peer_request = request.change_peer();
     const auto & new_region = response.change_peer().region();
 
+    bool pending_remove = false;
     switch (change_peer_request.change_type())
     {
         case eraftpb::ConfChangeType::AddNode:
         case eraftpb::ConfChangeType::AddLearnerNode:
         {
-            std::lock_guard<std::mutex> lock(mutex);
-
             // change the peers of region, add conf_ver.
             doSetRegion(new_region);
-            doSetApplied(index, term);
-            return;
+            break;
         }
         case eraftpb::ConfChangeType::RemoveNode:
         {
-            const auto & peer = change_peer_request.peer();
-
-            std::lock_guard<std::mutex> lock(mutex);
+            if (peer.id() == change_peer_request.peer().id())
+                pending_remove = true;
 
             doSetRegion(new_region);
-
-            if (this->peer.id() == peer.id())
-                region_state.set_state(raft_serverpb::PeerState::Tombstone);
-
-            doSetApplied(index, term);
-            return;
+            break;
         }
         default:
             throw Exception("[RegionMeta::execChangePeer] unsupported cmd", ErrorCodes::LOGICAL_ERROR);
     }
+
+    if (pending_remove)
+        region_state.setState(raft_serverpb::PeerState::Tombstone);
+    else
+        region_state.setState(raft_serverpb::PeerState::Normal);
+    region_state.clearMergeState();
+    doSetApplied(index, term);
 }
 
 void MetaRaftCommandDelegate::execCompactLog(
@@ -250,10 +251,10 @@ bool RegionMeta::isPeerRemoved() const
 {
     std::lock_guard<std::mutex> lock(mutex);
 
-    if (region_state.state() == raft_serverpb::PeerState::Tombstone)
+    if (region_state.getState() == raft_serverpb::PeerState::Tombstone)
         return true;
 
-    for (const auto & region_peer : region_state.region().peers())
+    for (const auto & region_peer : region_state.getRegion().peers())
     {
         if (region_peer.id() == peer.id())
             return false;
@@ -270,11 +271,10 @@ bool operator==(const RegionMeta & meta1, const RegionMeta & meta2)
         && meta1.region_state == meta2.region_state;
 }
 
-std::tuple<RegionVersion, RegionVersion, RegionRange> RegionMeta::dumpVersionRange() const
+std::tuple<RegionVersion, RegionVersion, ImutRegionRangePtr> RegionMeta::dumpVersionRange() const
 {
     std::lock_guard<std::mutex> lock(mutex);
-    return {region_state.region().region_epoch().version(), region_state.region().region_epoch().conf_ver(),
-        std::make_pair(TiKVKey::copyFrom(region_state.region().start_key()), TiKVKey::copyFrom(region_state.region().end_key()))};
+    return {region_state.getRegion().region_epoch().version(), region_state.getRegion().region_epoch().conf_ver(), region_state.getRange()};
 }
 
 MetaRaftCommandDelegate & RegionMeta::makeRaftCommandDelegate()
@@ -283,4 +283,27 @@ MetaRaftCommandDelegate & RegionMeta::makeRaftCommandDelegate()
     return static_cast<MetaRaftCommandDelegate &>(*this);
 }
 
+const metapb::Peer & MetaRaftCommandDelegate::getPeer() const { return peer; }
+const raft_serverpb::RaftApplyState & MetaRaftCommandDelegate::applyState() const { return apply_state; }
+const UInt64 & MetaRaftCommandDelegate::appliedTerm() const { return applied_term; }
+const RegionState & MetaRaftCommandDelegate::regionState() const { return region_state; }
+
+RegionMeta::RegionMeta(metapb::Peer peer_, raft_serverpb::RaftApplyState apply_state_, const UInt64 applied_term_,
+    raft_serverpb::RegionLocalState region_state_)
+    : peer(std::move(peer_)),
+      apply_state(std::move(apply_state_)),
+      applied_term(applied_term_),
+      region_state(std::move(region_state_)),
+      region_id(region_state.getRegion().id())
+{}
+
+RegionMeta::RegionMeta(metapb::Peer peer_, metapb::Region region, raft_serverpb::RaftApplyState apply_state_)
+    : peer(std::move(peer_)),
+      apply_state(std::move(apply_state_)),
+      applied_term(apply_state.truncated_state().term()),
+      region_id(region.id())
+{
+    region_state.setRegion(std::move(region));
+}
+
 } // namespace DB
diff --git a/dbms/src/Storages/Transaction/RegionMeta.h b/dbms/src/Storages/Transaction/RegionMeta.h
index 0370d0949b0..6fc48a76025 100644
--- a/dbms/src/Storages/Transaction/RegionMeta.h
+++ b/dbms/src/Storages/Transaction/RegionMeta.h
@@ -2,13 +2,7 @@
 
 #include <condition_variable>
 
-#include <Storages/Transaction/TiKVKeyValue.h>
-
-#pragma GCC diagnostic push
-#pragma GCC diagnostic ignored "-Wunused-parameter"
-#include <kvproto/metapb.pb.h>
-#include <kvproto/raft_serverpb.pb.h>
-#pragma GCC diagnostic pop
+#include <Storages/Transaction/RegionState.h>
 
 namespace pingcap::kv
 {
@@ -18,29 +12,15 @@ struct RegionVerID;
 namespace DB
 {
 
-using RegionRange = std::pair<TiKVKey, TiKVKey>;
 class MetaRaftCommandDelegate;
 
 class RegionMeta
 {
 public:
     RegionMeta(metapb::Peer peer_, raft_serverpb::RaftApplyState apply_state_, const UInt64 applied_term_,
-        raft_serverpb::RegionLocalState region_state_)
-        : peer(std::move(peer_)),
-          apply_state(std::move(apply_state_)),
-          applied_term(applied_term_),
-          region_state(std::move(region_state_)),
-          region_id(region_state.region().id())
-    {}
-
-    RegionMeta(metapb::Peer peer_, metapb::Region region, raft_serverpb::RaftApplyState apply_state_)
-        : peer(std::move(peer_)),
-          apply_state(std::move(apply_state_)),
-          applied_term(apply_state.truncated_state().term()),
-          region_id(region.id())
-    {
-        *region_state.mutable_region() = std::move(region);
-    }
+        raft_serverpb::RegionLocalState region_state_);
+
+    RegionMeta(metapb::Peer peer_, metapb::Region region, raft_serverpb::RaftApplyState apply_state_);
 
     RegionMeta(RegionMeta && meta);
 
@@ -51,7 +31,7 @@ class RegionMeta
     UInt64 appliedIndex() const;
     UInt64 appliedTerm() const;
 
-    RegionRange getRange() const;
+    ImutRegionRangePtr getRange() const;
 
     metapb::Peer getPeer() const;
     pingcap::kv::RegionVerID getRegionVerID() const;
@@ -85,7 +65,7 @@ class RegionMeta
 
     bool isPeerRemoved() const;
 
-    std::tuple<RegionVersion, RegionVersion, RegionRange> dumpVersionRange() const;
+    std::tuple<RegionVersion, RegionVersion, ImutRegionRangePtr> dumpVersionRange() const;
     MetaRaftCommandDelegate & makeRaftCommandDelegate();
 
 private:
@@ -103,7 +83,7 @@ class RegionMeta
     raft_serverpb::RaftApplyState apply_state;
     UInt64 applied_term;
 
-    raft_serverpb::RegionLocalState region_state;
+    RegionState region_state;
 
     mutable std::mutex mutex;
     mutable std::condition_variable cv;
@@ -132,6 +112,11 @@ class MetaRaftCommandDelegate : public RegionMeta, private boost::noncopyable
 
     MetaRaftCommandDelegate() = delete;
 
+    const metapb::Peer & getPeer() const;
+    const raft_serverpb::RaftApplyState & applyState() const;
+    const UInt64 & appliedTerm() const;
+    const RegionState & regionState() const;
+
     void execChangePeer(const raft_cmdpb::AdminRequest & request, const raft_cmdpb::AdminResponse & response, UInt64 index, UInt64 term);
     void execCompactLog(const raft_cmdpb::AdminRequest & request, const raft_cmdpb::AdminResponse & response, UInt64 index, UInt64 term);
 };
diff --git a/dbms/src/Storages/Transaction/RegionRangeKeys.h b/dbms/src/Storages/Transaction/RegionRangeKeys.h
new file mode 100644
index 00000000000..bb54bc4e3d2
--- /dev/null
+++ b/dbms/src/Storages/Transaction/RegionRangeKeys.h
@@ -0,0 +1,22 @@
+#pragma once
+
+#include <Storages/Transaction/TiKVHandle.h>
+#include <Storages/Transaction/TiKVKeyValue.h>
+
+namespace DB
+{
+
+class RegionRangeKeys : boost::noncopyable
+{
+public:
+    const std::pair<TiKVKey, TiKVKey> & keys() const;
+    const std::pair<DecodedTiKVKey, DecodedTiKVKey> & rawKeys() const;
+    HandleRange<HandleID> getHandleRangeByTable(const TableID table_id) const;
+    explicit RegionRangeKeys(TiKVKey start_key, TiKVKey end_key);
+
+private:
+    std::pair<TiKVKey, TiKVKey> ori;
+    std::pair<DecodedTiKVKey, DecodedTiKVKey> raw;
+};
+
+} // namespace DB
diff --git a/dbms/src/Storages/Transaction/RegionState.cpp b/dbms/src/Storages/Transaction/RegionState.cpp
new file mode 100644
index 00000000000..b19b10b48da
--- /dev/null
+++ b/dbms/src/Storages/Transaction/RegionState.cpp
@@ -0,0 +1,71 @@
+#include <Storages/Transaction/RegionState.h>
+#include <Storages/Transaction/TiKVRange.h>
+
+namespace DB
+{
+
+RegionState::RegionState(RegionState && region_state) : Base(std::move(region_state)), region_range(region_state.region_range) {}
+RegionState::RegionState(Base && region_state) : Base(std::move(region_state)) { updateRegionRange(); }
+RegionState & RegionState::operator=(RegionState && from)
+{
+    if (&from == this)
+        return *this;
+
+    (Base &)* this = (Base &&) from;
+    region_range = std::move(from.region_range);
+    return *this;
+}
+
+void RegionState::setRegion(metapb::Region region)
+{
+    getMutRegion() = std::move(region);
+    updateRegionRange();
+}
+
+void RegionState::setVersion(const UInt64 version) { getMutRegion().mutable_region_epoch()->set_version(version); }
+
+void RegionState::setConfVersion(const UInt64 version) { getMutRegion().mutable_region_epoch()->set_conf_ver(version); }
+
+void RegionState::setStartKey(std::string key)
+{
+    getMutRegion().set_start_key(std::move(key));
+    updateRegionRange();
+}
+
+void RegionState::setEndKey(std::string key)
+{
+    getMutRegion().set_end_key(std::move(key));
+    updateRegionRange();
+}
+
+ImutRegionRangePtr RegionState::getRange() const { return region_range; }
+
+void RegionState::updateRegionRange()
+{
+    region_range = std::make_shared<RegionRangeKeys>(TiKVKey::copyFrom(getRegion().start_key()), TiKVKey::copyFrom(getRegion().end_key()));
+}
+
+metapb::Region & RegionState::getMutRegion() { return *mutable_region(); }
+const metapb::Region & RegionState::getRegion() const { return region(); }
+raft_serverpb::PeerState RegionState::getState() const { return state(); }
+void RegionState::setState(raft_serverpb::PeerState value) { set_state(value); }
+void RegionState::clearMergeState() { clear_merge_state(); }
+bool RegionState::operator==(const RegionState & region_state) const { return getBase() == region_state.getBase(); }
+const RegionState::Base & RegionState::getBase() const { return *this; }
+
+RegionRangeKeys::RegionRangeKeys(TiKVKey start_key, TiKVKey end_key)
+    : ori(std::move(start_key), std::move(end_key)),
+      raw(ori.first.empty() ? DecodedTiKVKey() : RecordKVFormat::decodeTiKVKey(ori.first),
+          ori.second.empty() ? DecodedTiKVKey() : RecordKVFormat::decodeTiKVKey(ori.second))
+{}
+
+const std::pair<TiKVKey, TiKVKey> & RegionRangeKeys::keys() const { return ori; }
+
+const std::pair<DecodedTiKVKey, DecodedTiKVKey> & RegionRangeKeys::rawKeys() const { return raw; }
+
+HandleRange<HandleID> RegionRangeKeys::getHandleRangeByTable(const TableID table_id) const
+{
+    return TiKVRange::getHandleRangeByTable(rawKeys().first, rawKeys().second, table_id);
+}
+
+} // namespace DB
diff --git a/dbms/src/Storages/Transaction/RegionState.h b/dbms/src/Storages/Transaction/RegionState.h
new file mode 100644
index 00000000000..6c78ec18c52
--- /dev/null
+++ b/dbms/src/Storages/Transaction/RegionState.h
@@ -0,0 +1,47 @@
+#pragma once
+
+#include <Storages/Transaction/RegionRangeKeys.h>
+
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wunused-parameter"
+#include <kvproto/metapb.pb.h>
+#include <kvproto/raft_serverpb.pb.h>
+#pragma GCC diagnostic pop
+
+namespace DB
+{
+
+using ImutRegionRangePtr = std::shared_ptr<const RegionRangeKeys>;
+
+class RegionState : private raft_serverpb::RegionLocalState, private boost::noncopyable
+{
+public:
+    using Base = raft_serverpb::RegionLocalState;
+
+    RegionState() = default;
+    explicit RegionState(RegionState && region_state);
+    explicit RegionState(Base && region_state);
+    RegionState & operator=(RegionState && from);
+
+    void setRegion(metapb::Region region);
+    void setVersion(const UInt64 version);
+    void setConfVersion(const UInt64 version);
+    void setStartKey(std::string key);
+    void setEndKey(std::string key);
+    ImutRegionRangePtr getRange() const;
+    const metapb::Region & getRegion() const;
+    raft_serverpb::PeerState getState() const;
+    void setState(raft_serverpb::PeerState value);
+    void clearMergeState();
+    bool operator==(const RegionState & region_state) const;
+    const Base & getBase() const;
+
+private:
+    void updateRegionRange();
+    metapb::Region & getMutRegion();
+
+private:
+    ImutRegionRangePtr region_range = nullptr;
+};
+
+} // namespace DB
diff --git a/dbms/src/Storages/Transaction/RegionTable.cpp b/dbms/src/Storages/Transaction/RegionTable.cpp
index bcb36472aa1..0545c34206f 100644
--- a/dbms/src/Storages/Transaction/RegionTable.cpp
+++ b/dbms/src/Storages/Transaction/RegionTable.cpp
@@ -43,15 +43,14 @@ RegionTable::Table & RegionTable::getOrCreateTable(const TableID table_id)
 RegionTable::InternalRegion & RegionTable::insertRegion(Table & table, const Region & region)
 {
     const auto range = region.getRange();
-    return insertRegion(table, range.first, range.second, region.id());
+    return insertRegion(table, *range, region.id());
 }
 
-RegionTable::InternalRegion & RegionTable::insertRegion(Table & table, const TiKVKey & start, const TiKVKey & end, const RegionID region_id)
+RegionTable::InternalRegion & RegionTable::insertRegion(Table & table, const RegionRangeKeys & region_range_keys, const RegionID region_id)
 {
     auto & table_regions = table.regions;
     // Insert table mapping.
-    auto [it, ok]
-        = table_regions.emplace(region_id, InternalRegion(region_id, TiKVRange::getHandleRangeByTable(start, end, table.table_id)));
+    auto [it, ok] = table_regions.emplace(region_id, InternalRegion(region_id, region_range_keys.getHandleRangeByTable(table.table_id)));
     if (!ok)
         throw Exception(
             "[RegionTable::insertRegion] insert duplicate internal region " + DB::toString(region_id), ErrorCodes::LOGICAL_ERROR);
@@ -95,8 +94,7 @@ void RegionTable::doShrinkRegionRange(const Region & region)
     {
         auto table_id = *region_table_it;
 
-        const auto handle_range = TiKVRange::getHandleRangeByTable(range, table_id);
-
+        const auto handle_range = range->getHandleRangeByTable(table_id);
         auto table_it = tables.find(table_id);
         if (table_it == tables.end())
             throw Exception("Table " + DB::toString(table_id) + " not found in table map", ErrorCodes::LOGICAL_ERROR);
diff --git a/dbms/src/Storages/Transaction/RegionTable.h b/dbms/src/Storages/Transaction/RegionTable.h
index 36ab104ca66..05f197908bf 100644
--- a/dbms/src/Storages/Transaction/RegionTable.h
+++ b/dbms/src/Storages/Transaction/RegionTable.h
@@ -30,6 +30,7 @@ class Block;
 // for debug
 struct MockTiDBTable;
 using RegionMap = std::unordered_map<RegionID, RegionPtr>;
+class RegionRangeKeys;
 
 class RegionTable : private boost::noncopyable
 {
@@ -135,7 +136,7 @@ class RegionTable : private boost::noncopyable
 
     InternalRegion & insertRegion(Table & table, const Region & region);
     InternalRegion & getOrInsertRegion(TableID table_id, const Region & region);
-    InternalRegion & insertRegion(Table & table, const TiKVKey & start, const TiKVKey & end, const RegionID region_id);
+    InternalRegion & insertRegion(Table & table, const RegionRangeKeys & region_range_keys, const RegionID region_id);
 
     bool shouldFlush(const InternalRegion & region) const;
 
diff --git a/dbms/src/Storages/Transaction/TiKVKeyValue.h b/dbms/src/Storages/Transaction/TiKVKeyValue.h
index 353c3f9afe2..a0c67123735 100644
--- a/dbms/src/Storages/Transaction/TiKVKeyValue.h
+++ b/dbms/src/Storages/Transaction/TiKVKeyValue.h
@@ -79,8 +79,7 @@ struct DecodedTiKVKey : std::string
 {
     using Base = std::string;
     DecodedTiKVKey(Base && str_) : Base(std::move(str_)) {}
-    DecodedTiKVKey() = delete;
-    DecodedTiKVKey(const char * str) : Base(str) {}
+    DecodedTiKVKey() = default;
 };
 
 static_assert(sizeof(DecodedTiKVKey) == sizeof(std::string));
diff --git a/dbms/src/Storages/Transaction/TiKVRange.h b/dbms/src/Storages/Transaction/TiKVRange.h
index 4ecf2fc9e3a..e31eec406e3 100644
--- a/dbms/src/Storages/Transaction/TiKVRange.h
+++ b/dbms/src/Storages/Transaction/TiKVRange.h
@@ -80,11 +80,6 @@ inline HandleRange<HandleID> getHandleRangeByTable(const KeyType & start_key, co
     return {start_handle, end_handle};
 }
 
-inline HandleRange<HandleID> getHandleRangeByTable(const std::pair<TiKVKey, TiKVKey> & range, TableID table_id)
-{
-    return getHandleRangeByTable(range.first, range.second, table_id);
-}
-
 } // namespace TiKVRange
 
 } // namespace DB
diff --git a/dbms/src/Storages/Transaction/applySnapshot.cpp b/dbms/src/Storages/Transaction/applySnapshot.cpp
index add7e60fa02..d029e6b7395 100644
--- a/dbms/src/Storages/Transaction/applySnapshot.cpp
+++ b/dbms/src/Storages/Transaction/applySnapshot.cpp
@@ -115,10 +115,7 @@ void applySnapshot(const KVStorePtr & kvstore, RequestReader read, Context * con
             auto & key = *it->mutable_key();
             auto & value = *it->mutable_value();
 
-            auto & tikv_key = static_cast<TiKVKey &>(key);
-            auto & tikv_value = static_cast<TiKVValue &>(value);
-
-            new_region->insert(data.cf(), std::move(tikv_key), std::move(tikv_value));
+            new_region->insert(data.cf(), TiKVKey(std::move(key)), TiKVValue(std::move(value)));
         }
     }