storage: handle IngestSST using segment split

dbms/src/Common/CurrentMetrics.cpp

@@ -49,10 +49,10 @@
     M(DT_SnapshotOfReadRaw)                     \
     M(DT_SnapshotOfSegmentSplit)                \
     M(DT_SnapshotOfSegmentMerge)                \
+    M(DT_SnapshotOfSegmentIngest)               \
     M(DT_SnapshotOfDeltaMerge)                  \
     M(DT_SnapshotOfDeltaCompact)                \
     M(DT_SnapshotOfPlaceIndex)                  \
-    M(DT_SnapshotOfSegmentIngest)               \
     M(IOLimiterPendingBgWriteReq)               \
     M(IOLimiterPendingFgWriteReq)               \
     M(IOLimiterPendingBgReadReq)                \

dbms/src/Common/FailPoint.cpp

@@ -91,6 +91,8 @@ std::unordered_map<String, std::shared_ptr<FailPointChannel>> FailPointHelper::f
     M(force_context_path)                                    \
     M(force_slow_page_storage_snapshot_release)              \
     M(force_change_all_blobs_to_read_only)                   \
+    M(force_ingest_via_delta)                                \
+    M(force_ingest_via_replace)                              \
     M(unblock_query_init_after_write)                        \
     M(exception_in_merged_task_init)                         \
     M(force_fail_in_flush_region_data)

dbms/src/Common/ProfileEvents.cpp

@@ -100,6 +100,8 @@
     M(DMCleanReadRows)                         \
     M(DMSegmentIsEmptyFastPath)                \
     M(DMSegmentIsEmptySlowPath)                \
+    M(DMSegmentIngestDataByReplace)            \
+    M(DMSegmentIngestDataIntoDelta)            \
                                                \
     M(FileFSync)                               \
                                                \

dbms/src/Common/TiFlashMetrics.h

@@ -124,7 +124,7 @@ namespace DB
         F(type_seg_split_bg, {"type", "seg_split_bg"}),                                                                                   \
         F(type_seg_split_fg, {"type", "seg_split_fg"}),                                                                                   \
         F(type_seg_split_ingest, {"type", "seg_split_ingest"}),                                                                           \
-        F(type_seg_merge_bg_gc, {"type", "type_seg_merge_bg_gc"}),                                                                        \
+        F(type_seg_merge_bg_gc, {"type", "seg_merge_bg_gc"}),                                                                        \
         F(type_place_index_update, {"type", "place_index_update"}))                                                                       \
     M(tiflash_storage_subtask_duration_seconds, "Bucketed histogram of storage's sub task duration", Histogram,                           \
         F(type_delta_merge_bg, {{"type", "delta_merge_bg"}}, ExpBuckets{0.001, 2, 20}),                                                   \

dbms/src/Storages/DeltaMerge/Delta/DeltaValueSpace.h

@@ -60,7 +60,7 @@ class DeltaValueSpace
     , private boost::noncopyable
 {
 public:
-    using Lock = std::unique_lock<std::mutex>;
+    using Lock = std::unique_lock<std::recursive_mutex>;
 
 private:
     /// column files in `persisted_file_set` are all persisted in disks and can be restored after restart.
@@ -94,7 +94,8 @@ class DeltaValueSpace
     DeltaIndexPtr delta_index;
 
     // Protects the operations in this instance.
-    mutable std::mutex mutex;
+    // It is a recursive_mutex because the lock may be also used by the parent segment as its update lock.
+    mutable std::recursive_mutex mutex;
 
     LoggerPtr log;
 

dbms/src/Storages/DeltaMerge/Delta/MemTableSet.cpp

@@ -234,6 +234,9 @@ void MemTableSet::appendDeleteRange(const RowKeyRange & delete_range)
 
 void MemTableSet::ingestColumnFiles(const RowKeyRange & range, const ColumnFiles & new_column_files, bool clear_data_in_range)
 {
+    for (const auto & f : new_column_files)
+        RUNTIME_CHECK(f->isBigFile());
+
     // Prepend a DeleteRange to clean data before applying column files
     if (clear_data_in_range)
     {
@@ -242,9 +245,7 @@ void MemTableSet::ingestColumnFiles(const RowKeyRange & range, const ColumnFiles
     }
 
     for (const auto & f : new_column_files)
-    {
         appendColumnFileInner(f);
-    }
 }
 
 ColumnFileSetSnapshotPtr MemTableSet::createSnapshot(const StorageSnapshotPtr & storage_snap, bool disable_sharing)

dbms/src/Storages/DeltaMerge/DeltaMergeStore.h

@@ -442,7 +442,7 @@ class DeltaMergeStore : private boost::noncopyable
     {
         ForegroundWrite,
         Background,
-        IngestBySplit,
+        ForIngest,
     };
 
     /**
@@ -510,29 +510,31 @@ class DeltaMergeStore : private boost::noncopyable
         SegmentSnapshotPtr segment_snap = nullptr);
 
     /**
-     * Discard all data in the segment, and use the specified DMFile as the stable instead.
-     * The specified DMFile is safe to be shared for multiple segments.
+     * Ingest a DMFile into the segment, optionally causing a new segment being created.
      *
-     * Note 1: This function will not enable GC for the new_stable_file for you, in case of you may want to share the same
-     *         stable file for multiple segments. It is your own duty to enable GC later.
-     *
-     * Note 2: You must ensure the specified new_stable_file has been managed by the storage pool, and has been written
-     *         to the PageStorage's data. Otherwise there will be exceptions.
-     *
-     * Note 3: This API is subjected to be changed in future, as it relies on the knowledge that all current data
-     *         in this segment is useless, which is a pretty tough requirement.
+     * Note 1: You must ensure the DMFile is not shared in multiple segments.
+     * Note 2: You must enable the GC for the DMFile by yourself.
+     * Note 3: You must ensure the DMFile has been managed by the storage pool, and has been written
+     *         to the PageStorage's data.
+
+     * @param clear_all_data_in_segment Whether all data in the segment should be discarded.
+     * @returns one of:
+     *          - A new segment: A new segment is created for containing the data
+     *          - The same segment as passed in: Data is ingested into the delta layer of current segment
+     *          - nullptr: when there are errors
      */
-    SegmentPtr segmentDangerouslyReplaceData(
+    SegmentPtr segmentIngestData(
         DMContext & dm_context,
         const SegmentPtr & segment,
-        const DMFilePtr & data_file);
+        const DMFilePtr & data_file,
+        bool clear_all_data_in_segment);
 
     // isSegmentValid should be protected by lock on `read_write_mutex`
-    inline bool isSegmentValid(const std::shared_lock<std::shared_mutex> &, const SegmentPtr & segment)
+    bool isSegmentValid(const std::shared_lock<std::shared_mutex> &, const SegmentPtr & segment)
     {
         return doIsSegmentValid(segment);
     }
-    inline bool isSegmentValid(const std::unique_lock<std::shared_mutex> &, const SegmentPtr & segment)
+    bool isSegmentValid(const std::unique_lock<std::shared_mutex> &, const SegmentPtr & segment)
     {
         return doIsSegmentValid(segment);
     }
@@ -559,7 +561,8 @@ class DeltaMergeStore : private boost::noncopyable
         DMContext & dm_context,
         const SegmentPtr & segment,
         const RowKeyRange & ingest_range,
-        const DMFilePtr & file);
+        const DMFilePtr & file,
+        bool clear_data_in_range);
 
     bool updateGCSafePoint();