KVStore: LargeTxn spill: Store kv pairs of large txns in dedicated CFs

dbms/src/Common/FailPoint.cpp

@@ -111,6 +111,7 @@ namespace DB
     M(force_set_fap_candidate_store_id)                      \
     M(force_not_clean_fap_on_destroy)                        \
     M(delta_tree_create_node_fail)                           \
+    M(force_write_to_large_txn_default)                      \
     M(disable_flush_cache)
 
 #define APPLY_FOR_PAUSEABLE_FAILPOINTS_ONCE(M) \

dbms/src/Debug/MockKVStore/MockUtils.h

@@ -69,11 +69,13 @@ inline metapb::Region createRegionInfo(
 inline RegionMeta createRegionMeta(
     UInt64 id,
     DB::TableID table_id,
+    const HandleID from = 0,
+    const HandleID to = 300,
     std::optional<raft_serverpb::RaftApplyState> apply_state = std::nullopt)
 {
     return RegionMeta(
         /*peer=*/createPeer(31, true),
-        /*region=*/createRegionInfo(id, RecordKVFormat::genKey(table_id, 0), RecordKVFormat::genKey(table_id, 300)),
+        /*region=*/createRegionInfo(id, RecordKVFormat::genKey(table_id, from), RecordKVFormat::genKey(table_id, to)),
         /*apply_state_=*/apply_state.value_or(initialApplyState()));
 }
 

dbms/src/Debug/dbgKVStore/dbgFuncMockRaftCommand.cpp

@@ -15,7 +15,7 @@
 #include <Common/typeid_cast.h>
 #include <Debug/DBGInvoker.h>
 #include <Debug/MockTiDB.h>
-#include <Debug/MockTiKV.h>
+#include <Debug/dbgKVStore/MockTiKV.h>
 #include <Debug/dbgKVStore/dbgFuncMockRaftCommand.h>
 #include <Debug/dbgTools.h>
 #include <Interpreters/Context.h>

dbms/src/Debug/dbgKVStore/dbgFuncMockRaftSnapshot.cpp

@@ -18,7 +18,7 @@
 #include <Common/typeid_cast.h>
 #include <Debug/MockKVStore/MockSSTReader.h>
 #include <Debug/MockTiDB.h>
-#include <Debug/MockTiKV.h>
+#include <Debug/dbgKVStore/MockTiKV.h>
 #include <Debug/dbgKVStore/dbgFuncMockRaftCommand.h>
 #include <Debug/dbgKVStore/dbgFuncRegion.h>
 #include <Debug/dbgTools.h>

dbms/src/Debug/dbgKVStore/dbgFuncRegion.cpp

@@ -15,7 +15,7 @@
 #include <Common/FmtUtils.h>
 #include <Common/typeid_cast.h>
 #include <Debug/MockTiDB.h>
-#include <Debug/MockTiKV.h>
+#include <Debug/dbgKVStore/MockTiKV.h>
 #include <Debug/dbgKVStore/dbgFuncRegion.h>
 #include <Debug/dbgKVStore/dbgKVStore.h>
 #include <Debug/dbgTools.h>

dbms/src/Debug/dbgKVStore/dbgKVStore.cpp

@@ -18,6 +18,11 @@
 namespace DB::RegionBench
 {
 
+RegionSerdeOpts & DebugKVStore::mutRegionSerdeOpts()
+{
+    return kvstore.region_persister->region_serde_opts;
+}
+
 void DebugKVStore::mockRemoveRegion(DB::RegionID region_id, RegionTable & region_table)
 {
     auto task_lock = kvstore.genTaskLock();

dbms/src/Debug/dbgKVStore/dbgKVStore.h

@@ -28,6 +28,7 @@ struct DebugKVStore
     KVStore * operator->() { return &kvstore; }
     KVStore * operator->() const { return &kvstore; }
 
+    RegionSerdeOpts & mutRegionSerdeOpts();
     template <typename RegionPtrWrap>
     void onSnapshot(const RegionPtrWrap & r, RegionPtr old_region, UInt64 old_region_index, TMTContext & tmt)
     {

dbms/src/Debug/dbgNaturalDag.cpp

@@ -13,8 +13,8 @@
 // limitations under the License.
 
 #include <Debug/MockTiDB.h>
-#include <Debug/MockTiKV.h>
 #include <Debug/dbgFuncCoprocessor.h>
+#include <Debug/dbgKVStore/MockTiKV.h>
 #include <Debug/dbgNaturalDag.h>
 #include <Debug/dbgTools.h>
 #include <Flash/Coprocessor/DAGContext.h>

dbms/src/Debug/dbgTools.cpp

@@ -14,7 +14,7 @@
 
 #include <Common/typeid_cast.h>
 #include <Debug/MockTiDB.h>
-#include <Debug/MockTiKV.h>
+#include <Debug/dbgKVStore/MockTiKV.h>
 #include <Debug/dbgKVStore/dbgKVStore.h>
 #include <Debug/dbgTools.h>
 #include <Interpreters/Context.h>
@@ -560,10 +560,12 @@ void concurrentBatchInsert(
     Context & context)
 {
     TMTContext & tmt = context.getTMTContext();
+    auto & orig_kvs = tmt.getKVStore();
+    auto debug_kvs = DebugKVStore(*orig_kvs);
 
     RegionID curr_max_region_id(InvalidRegionID);
     HandleID curr_max_handle_id = 0;
-    tmt.getKVStore()->traverseRegions([&](const RegionID region_id, const RegionPtr & region) {
+    debug_kvs->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);
         const auto range = region->getRange();

dbms/src/Storages/KVStore/KVStore.h

@@ -147,6 +147,7 @@ class KVStore final : private boost::noncopyable
     FileUsageStatistics getFileUsageStatistics() const;
     // Proxy will validate and refit the config items from the toml file.
     const ProxyConfigSummary & getProxyConfigSummay() const { return proxy_config_summary; }
+    const RegionPersister & getRegionPersister() const { return *region_persister; }
 
 public: // Region Management
     void restore(PathPool & path_pool, const TiFlashRaftProxyHelper *);

dbms/src/Storages/KVStore/MultiRaft/Disagg/CheckpointIngestInfo.cpp

@@ -140,7 +140,10 @@ FastAddPeerProto::CheckpointIngestInfoPersisted CheckpointIngestInfo::serializeM
         // Although the region is the first peer of this region in this store, we can't write it to formal KVStore for now.
         // Otherwise it could be uploaded and then overwritten.
         WriteBufferFromOwnString wb;
-        RegionPersister::computeRegionWriteBuffer(*region, wb);
+        RegionPersister::computeRegionWriteBuffer(
+            *region,
+            wb,
+            tmt.getKVStore()->getRegionPersister().getRegionSerdeOpts());
         ingest_info_persisted.set_region_info(wb.releaseStr());
     }
     ingest_info_persisted.set_remote_store_id(remote_store_id);

dbms/src/Storages/KVStore/MultiRaft/RegionCFDataBase.cpp

@@ -16,6 +16,7 @@
 #include <Storages/KVStore/MultiRaft/RegionCFDataTrait.h>
 #include <Storages/KVStore/MultiRaft/RegionData.h>
 #include <Storages/KVStore/MultiRaft/RegionRangeKeys.h>
+#include <Storages/KVStore/MultiRaft/Spill/LargeTxnDefaultCf.h>
 
 namespace DB
 {
@@ -39,6 +40,21 @@ const TiKVValue & RegionCFDataBase<Trait>::getTiKVValue(const Value & val)
     return *getTiKVValuePtr<Value>(val);
 }
 
+RegionDataRes insertWithTs(
+    RegionCFDataBase<RegionDefaultCFDataTrait> & default_cf,
+    TiKVKey && key,
+    TiKVValue && value,
+    Timestamp ts,
+    DupCheck mode)
+{
+    const auto & raw_key = RecordKVFormat::decodeTiKVKey(key);
+    auto kv_pair = RegionDefaultCFDataTrait::genKVPairWithTs(std::move(key), ts, raw_key, std::move(value));
+    if (!kv_pair)
+        return 0;
+
+    return default_cf.doInsert(std::move(*kv_pair), mode);
+}
+
 template <typename Trait>
 RegionDataRes RegionCFDataBase<Trait>::insert(TiKVKey && key, TiKVValue && value, DupCheck mode)
 {
@@ -47,7 +63,7 @@ RegionDataRes RegionCFDataBase<Trait>::insert(TiKVKey && key, TiKVValue && value
     if (!kv_pair)
         return 0;
 
-    return insert(std::move(*kv_pair), mode);
+    return doInsert(std::move(*kv_pair), mode);
 }
 
 template <>
@@ -70,7 +86,7 @@ RegionDataRes RegionCFDataBase<RegionLockCFDataTrait>::insert(TiKVKey && key, Ti
 }
 
 template <typename Trait>
-RegionDataRes RegionCFDataBase<Trait>::insert(std::pair<Key, Value> && kv_pair, DupCheck mode)
+RegionDataRes RegionCFDataBase<Trait>::doInsert(std::pair<Key, Value> && kv_pair, DupCheck mode)
 {
     auto & map = data;
     TiKVValue prev_value;
@@ -265,15 +281,12 @@ size_t RegionCFDataBase<Trait>::splitInto(const RegionRange & range, RegionCFDat
 {
     const auto & [start_key, end_key] = range;
     size_t size_changed = 0;
-
     {
         auto & ori_map = data;
         auto & tar_map = new_region_data.data;
-
         for (auto it = ori_map.begin(); it != ori_map.end();)
         {
             const auto & key = getTiKVKey(it->second);
-
             if (start_key.compare(key) <= 0 && end_key.compare(key) > 0)
             {
                 size_changed += calcTiKVKeyValueSize(it->second);
@@ -336,4 +349,5 @@ typename RegionCFDataBase<Trait>::Data & RegionCFDataBase<Trait>::getDataMut()
 template struct RegionCFDataBase<RegionWriteCFDataTrait>;
 template struct RegionCFDataBase<RegionDefaultCFDataTrait>;
 template struct RegionCFDataBase<RegionLockCFDataTrait>;
+template struct RegionCFDataBase<LargeDefaultCFDataTrait>;
 } // namespace DB

dbms/src/Storages/KVStore/MultiRaft/RegionCFDataBase.h

@@ -21,8 +21,9 @@
 
 namespace DB
 {
-
+struct RegionDefaultCFDataTrait;
 struct TiKVRangeKey;
+
 using RegionRange = RegionRangeKeys::RegionRange;
 using RegionDataRes = int64_t;
 
@@ -43,7 +44,6 @@ struct RegionCFDataBase
     using Status = bool;
 
     static const TiKVKey & getTiKVKey(const Value & val);
-
     static const TiKVValue & getTiKVValue(const Value & val);
 
     RegionDataRes insert(TiKVKey && key, TiKVValue && value, DupCheck mode = DupCheck::Deny);
@@ -52,7 +52,7 @@ struct RegionCFDataBase
 
     static size_t calcTiKVKeyValueSize(const TiKVKey & key, const TiKVValue & value);
 
-    size_t remove(const Key & key, bool quiet = false);
+    size_t remove(const Key & key, bool quiet);
 
     static bool cmp(const Map & a, const Map & b);
 
@@ -68,19 +68,26 @@ struct RegionCFDataBase
     size_t mergeFrom(const RegionCFDataBase & ori_region_data);
 
     size_t serialize(WriteBuffer & buf) const;
-
     static size_t deserialize(ReadBuffer & buf, RegionCFDataBase & new_region_data);
 
     const Data & getData() const;
 
     Data & getDataMut();
 
+    RegionDataRes doInsert(std::pair<Key, Value> && kv_pair, DupCheck mode = DupCheck::Deny);
+
 private:
     static bool shouldIgnoreRemove(const Value & value);
-    RegionDataRes insert(std::pair<Key, Value> && kv_pair, DupCheck mode = DupCheck::Deny);
 
 private:
     Data data;
 };
 
+RegionDataRes insertWithTs(
+    RegionCFDataBase<RegionDefaultCFDataTrait> & default_cf,
+    TiKVKey && key,
+    TiKVValue && value,
+    Timestamp ts,
+    DupCheck mode);
+
 } // namespace DB

dbms/src/Storages/KVStore/MultiRaft/RegionCFDataTrait.h

@@ -62,7 +62,6 @@ struct RegionWriteCFDataTrait
     static UInt8 getWriteType(const Value & value) { return std::get<2>(value).write_type; }
 };
 
-
 struct RegionDefaultCFDataTrait
 {
     using Key = std::pair<RawTiDBPK, Timestamp>;
@@ -80,6 +79,20 @@ struct RegionDefaultCFDataTrait
                 std::make_shared<const TiKVValue>(std::move(value))));
     }
 
+    static std::optional<Map::value_type> genKVPairWithTs(
+        TiKVKey && key,
+        Timestamp ts,
+        const DecodedTiKVKey & raw_key,
+        TiKVValue && value)
+    {
+        RawTiDBPK tidb_pk = RecordKVFormat::getRawTiDBPK(raw_key);
+        return Map::value_type(
+            Key(std::move(tidb_pk), ts),
+            Value(
+                std::make_shared<const TiKVKey>(std::move(key)),
+                std::make_shared<const TiKVValue>(std::move(value))));
+    }
+
     static std::shared_ptr<const TiKVValue> getTiKVValue(const Map::const_iterator & it)
     {
         return std::get<1>(it->second);