storage: handle IngestSST using segment split

dbms/src/Common/CurrentMetrics.cpp

@@ -49,6 +49,7 @@
     M(DT_SnapshotOfReadRaw)                     \
     M(DT_SnapshotOfSegmentSplit)                \
     M(DT_SnapshotOfSegmentMerge)                \
+    M(DT_SnapshotOfSegmentIngest)               \
     M(DT_SnapshotOfDeltaMerge)                  \
     M(DT_SnapshotOfDeltaCompact)                \
     M(DT_SnapshotOfPlaceIndex)                  \

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_split)                                \
     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/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)
     {
@@ -243,6 +246,7 @@ void MemTableSet::ingestColumnFiles(const RowKeyRange & range, const ColumnFiles
 
     for (const auto & f : new_column_files)
     {
+        RUNTIME_CHECK(f->isBigFile());
         appendColumnFileInner(f);
     }
 }

dbms/src/Storages/DeltaMerge/DeltaMergeStore.h

@@ -266,11 +266,15 @@ class DeltaMergeStore : private boost::noncopyable
 
     void preIngestFile(const String & parent_path, PageId file_id, size_t file_size);
 
+    /// You must ensure external files are ordered and do not overlap. Otherwise exceptions will be thrown.
+    /// You must ensure all of the external files are contained by the range. Otherwise exceptions will be thrown.
     void ingestFiles(const DMContextPtr & dm_context, //
                      const RowKeyRange & range,
                      const std::vector<DM::ExternalDTFileInfo> & external_files,
                      bool clear_data_in_range);
 
+    /// You must ensure external files are ordered and do not overlap. Otherwise exceptions will be thrown.
+    /// You must ensure all of the external files are contained by the range. Otherwise exceptions will be thrown.
     void ingestFiles(const Context & db_context, //
                      const DB::Settings & db_settings,
                      const RowKeyRange & range,
@@ -504,29 +508,30 @@ 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 the new segment. The returned segment may be nullptr. In this case, it IS NOT A FAILURE.
+     *          It means the ingest does not cause a segment created and the current segment
+     *          is still valid.
      */
-    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);
     }
@@ -553,7 +558,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();
 

dbms/src/Storages/DeltaMerge/DeltaMergeStore_Ingest.cpp

@@ -19,6 +19,13 @@
 #include <Storages/DeltaMerge/File/DMFile.h>
 #include <Storages/DeltaMerge/Segment.h>
 
+#include <magic_enum.hpp>
+
+namespace CurrentMetrics
+{
+extern const Metric DT_SnapshotOfSegmentIngest;
+}
+
 namespace ProfileEvents
 {
 extern const Event DMWriteFile;
@@ -34,6 +41,8 @@ namespace FailPoints
 extern const char pause_when_ingesting_to_dt_store[];
 extern const char force_set_segment_ingest_packs_fail[];
 extern const char segment_merge_after_ingest_packs[];
+extern const char force_ingest_via_delta[];
+extern const char force_ingest_via_split[];
 } // namespace FailPoints
 
 namespace DM
@@ -94,10 +103,14 @@ Segments DeltaMergeStore::ingestDTFilesUsingColumnFile(
 
             segment_range = segment->getRowKeyRange();
 
-            // Write could fail, because other threads could already updated the instance. Like split/merge, merge delta.
-            ColumnFiles column_files;
+            // Write could fail, because we do not lock the segment here.
+            // Thus, other threads may update the instance at any time, like split, merge, merge delta,
+            // causing the segment to be abandoned.
             WriteBatches wbs(*storage_pool, dm_context->getWriteLimiter());
 
+            DMFiles data_files;
+            data_files.reserve(files.size());
+
             for (const auto & file : files)
             {
                 /// Generate DMFile instance with a new ref_id pointed to the file_id.
@@ -106,19 +119,15 @@ Segments DeltaMergeStore::ingestDTFilesUsingColumnFile(
                 auto page_id = storage_pool->newDataPageIdForDTFile(delegate, __PRETTY_FUNCTION__);
 
                 auto ref_file = DMFile::restore(file_provider, file_id, page_id, file_parent_path, DMFile::ReadMetaMode::all());
-                auto column_file = std::make_shared<ColumnFileBig>(*dm_context, ref_file, segment_range);
-                if (column_file->getRows() != 0)
-                {
-                    column_files.emplace_back(std::move(column_file));
-                    wbs.data.putRefPage(page_id, file->pageId());
-                }
+                data_files.emplace_back(std::move(ref_file));
+                wbs.data.putRefPage(page_id, file->pageId());
             }
 
             // We have to commit those file_ids to PageStorage, because as soon as packs are written into segments,
             // they are visible for readers who require file_ids to be found in PageStorage.
             wbs.writeLogAndData();
 
-            bool ingest_success = segment->ingestColumnFiles(*dm_context, range.shrink(segment_range), column_files, clear_data_in_range);
+            bool ingest_success = segment->ingestDataToDelta(*dm_context, range.shrink(segment_range), data_files, clear_data_in_range);
             fiu_do_on(FailPoints::force_set_segment_ingest_packs_fail, { ingest_success = false; });
             if (ingest_success)
             {
@@ -145,12 +154,10 @@ Segments DeltaMergeStore::ingestDTFilesUsingColumnFile(
 
 /**
  * Accept a target ingest range and a vector of DTFiles, ingest these DTFiles (clipped by the target ingest range)
- * using logical split. All data in the target ingest range will be cleared, and replaced by the specified DTFiles.
+ * using logical split.
  *
- * `clear_data_in_range` must be true. Otherwise exceptions will be thrown.
- *
- * You must ensure DTFiles do not overlap. Otherwise you will lose data.
- * // TODO: How to check this? Maybe we can enforce external_files to be ordered.
+ * You must ensure DTFiles do not overlap. Otherwise this function will not work properly when clear_data_in_range == true.
+ * The check is performed in `ingestFiles`.
  *
  * WARNING: This function does not guarantee isolation. You may observe partial results when
  * querying related segments when this function is running.
@@ -162,20 +169,23 @@ Segments DeltaMergeStore::ingestDTFilesUsingSplit(
     const DMFiles & files,
     bool clear_data_in_range)
 {
-    RUNTIME_CHECK(clear_data_in_range == true);
-    RUNTIME_CHECK(
-        files.size() == external_files.size(),
-        files.size(),
-        external_files.size());
-    for (size_t i = 0; i < files.size(); ++i)
     {
-        RUNTIME_CHECK(files[i]->pageId() == external_files[i].id);
+        RUNTIME_CHECK(
+            files.size() == external_files.size(),
+            files.size(),
+            external_files.size());
+        for (size_t i = 0; i < files.size(); ++i)
+            RUNTIME_CHECK(
+                files[i]->pageId() == external_files[i].id,
+                files[i]->pageId(),
+                external_files[i].toString());
     }
 
     std::set<SegmentPtr> updated_segments;
 
     // First phase (DeleteRange Phase):
     // Write DeleteRange to the covered segments to ensure that all data in the `ingest_range` is cleared.
+    if (clear_data_in_range)
     {
         RowKeyRange remaining_delete_range = ingest_range;
         LOG_INFO(
@@ -220,16 +230,16 @@ Segments DeltaMergeStore::ingestDTFilesUsingSplit(
                 // segment may be abandoned, retry current range by finding the segment again.
             }
         }
-    }
 
-    LOG_DEBUG(
-        log,
-        "Table ingest using split - delete range phase - finished, updated_segments_n={}",
-        updated_segments.size());
+        LOG_DEBUG(
+            log,
+            "Table ingest using split - delete range phase - finished, updated_segments_n={}",
+            updated_segments.size());
+    }
 
     /*
      * In second phase (SplitIngest Phase),
-     * we will try to ingest DMFile one by one into the segments in order.
+     * we will try to ingest DMFile one by one into the segments.
      *
      * Consider the following case:
      * -Inf                                                                         +Inf
@@ -323,7 +333,7 @@ Segments DeltaMergeStore::ingestDTFilesUsingSplit(
                 segment->simpleInfo(),
                 segment_ingest_range.toDebugString());
 
-            const bool succeeded = ingestDTFileIntoSegmentUsingSplit(*dm_context, segment, segment_ingest_range, files[file_idx]);
+            const bool succeeded = ingestDTFileIntoSegmentUsingSplit(*dm_context, segment, segment_ingest_range, files[file_idx], clear_data_in_range);
             if (succeeded)
             {
                 updated_segments.insert(segment);
@@ -356,7 +366,8 @@ bool DeltaMergeStore::ingestDTFileIntoSegmentUsingSplit(
     DMContext & dm_context,
     const SegmentPtr & segment,
     const RowKeyRange & ingest_range,
-    const DMFilePtr & file)
+    const DMFilePtr & file,
+    bool clear_data_in_range)
 {
     const auto & segment_range = segment->getRowKeyRange();
 
@@ -379,16 +390,47 @@ bool DeltaMergeStore::ingestDTFileIntoSegmentUsingSplit(
     if (is_start_matching && is_end_matching)
     {
         /*
-         * The segment and the ingest range is perfectly matched. We can
-         * simply replace all of the data from this segment.
+         * The segment and the ingest range is perfectly matched.
          *
          * Example:
          *    │----------- Segment ----------│
          *    │-------- Ingest Range --------│
          */
-        const auto new_segment_or_null = segmentDangerouslyReplaceData(dm_context, segment, file);
-        const bool succeeded = new_segment_or_null != nullptr;
-        return succeeded;
+
+        auto delegate = dm_context.path_pool.getStableDiskDelegator();
+        auto file_provider = dm_context.db_context.getFileProvider();
+
+        WriteBatches wbs(*storage_pool, dm_context.getWriteLimiter());
+
+        // Generate DMFile instance with a new ref_id pointed to the file_id,
+        // because we may use the same DMFile to ingest into multiple segments.
+        auto new_page_id = storage_pool->newDataPageIdForDTFile(delegate, __PRETTY_FUNCTION__);
+        auto ref_file = DMFile::restore(
+            file_provider,
+            file->fileId(),
+            new_page_id,
+            file->parentPath(),
+            DMFile::ReadMetaMode::all());
+        wbs.data.putRefPage(new_page_id, file->pageId());
+
+        // We have to commit those file_ids to PageStorage before applying the ingest, because after the write
+        // they are visible for readers immediately, who require file_ids to be found in PageStorage.
+        wbs.writeLogAndData();
+
+        // clear_all_data_in_segment == clear_data_in_range is safe, because we have verified
+        // that current segment range is identical to the ingest range.
+        auto new_segment = segmentIngestData(dm_context, segment, ref_file, clear_data_in_range);
+        if (new_segment == nullptr)
+        {
+            // When ingest failed, just discard this ref file.
+            wbs.rollbackWrittenLogAndData();
+
+            // Failed, segment may be abandoned, or there is another update in progress.
+            std::this_thread::sleep_for(std::chrono::milliseconds(15));
+            return false;
+        }
+
+        return true;
     }
     else if (is_start_matching)
     {
@@ -463,6 +505,41 @@ void DeltaMergeStore::ingestFiles(
         throw Exception(msg);
     }
 
+    {
+        // `ingestDTFilesUsingSplit` requires external_files to be not overlapped. Otherwise the results will be incorrect.
+        // Here we verify the external_files are ordered and not overlapped.
+        // "Ordered" is actually not a hard requirement by `ingestDTFilesUsingSplit`. However "ordered" makes us easy to check overlap efficiently.
+        RowKeyValue last_end;
+        if (is_common_handle)
+            last_end = RowKeyValue::COMMON_HANDLE_MIN_KEY;
+        else
+            last_end = RowKeyValue::INT_HANDLE_MIN_KEY;
+
+        // Suppose we have keys: 1, 2, | 3, 4, 5, | 6, | 7, 8
+        // Our file ranges will be: [1, 3), [3, 6), [6, 7), [7, 9)
+        //                              ↑    ↑
+        //                              A    B
+        //                         We require A <= B.
+        for (const auto & ext_file : external_files)
+        {
+            RUNTIME_CHECK(
+                !ext_file.range.none(),
+                ext_file.toString());
+            RUNTIME_CHECK(
+                compare(last_end.toRowKeyValueRef(), ext_file.range.getStart()) <= 0,
+                last_end.toDebugString(),
+                ext_file.toString());
+            last_end = ext_file.range.end;
+        }
+
+        // Check whether all external files are contained by the range.
+        for (const auto & ext_file : external_files)
+        {
+            RUNTIME_CHECK(compare(range.getStart(), ext_file.range.getStart()) <= 0);
+            RUNTIME_CHECK(compare(range.getEnd(), ext_file.range.getEnd()) >= 0);
+        }
+    }
+
     EventRecorder write_block_recorder(ProfileEvents::DMWriteFile, ProfileEvents::DMWriteFileNS);
 
     auto delegate = dm_context->path_pool.getStableDiskDelegator();
@@ -491,12 +568,15 @@ void DeltaMergeStore::ingestFiles(
     }
 
     bool ingest_using_split = false;
-    if (clear_data_in_range && bytes >= dm_context->delta_small_column_file_bytes)
+    if (bytes >= dm_context->delta_small_column_file_bytes)
     {
         // We still write small ssts directly into the delta layer.
         ingest_using_split = true;
     }
 
+    fiu_do_on(FailPoints::force_ingest_via_delta, { ingest_using_split = false; });
+    fiu_do_on(FailPoints::force_ingest_via_split, { ingest_using_split = true; });
+
     {
         auto get_ingest_files = [&] {
             FmtBuffer fmt_buf;