storage: Merge multiple segments in gc thread

dbms/src/Common/FailPoint.cpp

@@ -79,6 +79,7 @@ std::unordered_map<String, std::shared_ptr<FailPointChannel>> FailPointHelper::f
 
 #define APPLY_FOR_FAILPOINTS(M)                              \
     M(skip_check_segment_update)                             \
+    M(gc_skip_update_safe_point)                             \
     M(force_set_page_file_write_errno)                       \
     M(force_split_io_size_4k)                                \
     M(minimum_block_size_for_cross_join)                     \

dbms/src/Storages/DeltaMerge/DMContext.h

@@ -72,6 +72,10 @@ struct DMContext : private boost::noncopyable
     const size_t delta_small_column_file_bytes;
     // The expected stable pack rows.
     const size_t stable_pack_rows;
+    // The rows of segment to be regarded as small. Small segments will be merged.
+    const size_t small_segment_rows;
+    // The bytes of segment to be regarded as small. Small segments will be merged.
+    const size_t small_segment_bytes;
 
     // The number of points to check for calculating region split.
     const size_t region_split_check_points = 128;
@@ -111,6 +115,8 @@ struct DMContext : private boost::noncopyable
         , delta_small_column_file_rows(settings.dt_segment_delta_small_column_file_rows)
         , delta_small_column_file_bytes(settings.dt_segment_delta_small_column_file_size)
         , stable_pack_rows(settings.dt_segment_stable_pack_rows)
+        , small_segment_rows(settings.dt_segment_limit_rows / 3)
+        , small_segment_bytes(settings.dt_segment_limit_size / 3)
         , enable_logical_split(settings.dt_enable_logical_split)
         , read_delta_only(settings.dt_read_delta_only)
         , read_stable_only(settings.dt_read_stable_only)

dbms/src/Storages/DeltaMerge/DeltaMergeStore.cpp

@@ -105,7 +105,6 @@ std::pair<bool, bool> DeltaMergeStore::MergeDeltaTaskPool::tryAddTask(const Back
     switch (task.type)
     {
     case TaskType::Split:
-    case TaskType::Merge:
     case TaskType::MergeDelta:
         is_heavy = true;
         // reserve some task space for light tasks
@@ -1178,8 +1177,6 @@ void DeltaMergeStore::checkSegmentUpdate(const DMContextPtr & dm_context, const
                                 || delta_bytes - delta_last_try_split_bytes >= delta_cache_limit_bytes))
         || (segment_rows >= segment_limit_rows * 3 || segment_bytes >= segment_limit_bytes * 3);
 
-    bool should_merge = segment_rows < segment_limit_rows / 3 && segment_bytes < segment_limit_bytes / 3;
-
     // Don't do compact on starting up.
     bool should_compact = (thread_type != ThreadType::Init) && std::max(static_cast<Int64>(column_file_count) - delta_last_try_compact_column_files, 0) >= 10;
 
@@ -1236,7 +1233,7 @@ void DeltaMergeStore::checkSegmentUpdate(const DMContextPtr & dm_context, const
             {
                 delta_last_try_flush_rows = delta_rows;
                 delta_last_try_flush_bytes = delta_bytes;
-                try_add_background_task(BackgroundTask{TaskType::Flush, dm_context, segment, {}});
+                try_add_background_task(BackgroundTask{TaskType::Flush, dm_context, segment});
             }
         }
     }
@@ -1246,36 +1243,6 @@ void DeltaMergeStore::checkSegmentUpdate(const DMContextPtr & dm_context, const
     if (segment->getDelta()->isUpdating())
         return;
 
-    /// Now start trying structure update.
-
-    auto get_merge_sibling = [&]() -> SegmentPtr {
-        /// For complexity reason, currently we only try to merge with next segment. Normally it is good enough.
-
-        // The last segment cannot be merged.
-        if (segment->getRowKeyRange().isEndInfinite())
-            return {};
-        SegmentPtr next_segment;
-        {
-            std::shared_lock read_write_lock(read_write_mutex);
-
-            auto it = segments.find(segment->getRowKeyRange().getEnd());
-            // check legality
-            if (it == segments.end())
-                return {};
-            auto & cur_segment = it->second;
-            if (cur_segment.get() != segment.get())
-                return {};
-            ++it;
-            if (it == segments.end())
-                return {};
-            next_segment = it->second;
-            auto limit = dm_context->segment_limit_rows / 5;
-            if (next_segment->getEstimatedRows() >= limit)
-                return {};
-        }
-        return next_segment;
-    };
-
     auto try_fg_merge_delta = [&]() -> SegmentPtr {
         // If the table is already dropped, don't trigger foreground merge delta when executing `remove region peer`,
         // or the raft-log apply threads may be blocked.
@@ -1303,7 +1270,7 @@ void DeltaMergeStore::checkSegmentUpdate(const DMContextPtr & dm_context, const
         if (should_background_merge_delta)
         {
             delta_last_try_merge_delta_rows = delta_rows;
-            try_add_background_task(BackgroundTask{TaskType::MergeDelta, dm_context, segment, {}});
+            try_add_background_task(BackgroundTask{TaskType::MergeDelta, dm_context, segment});
             return true;
         }
         return false;
@@ -1313,12 +1280,12 @@ void DeltaMergeStore::checkSegmentUpdate(const DMContextPtr & dm_context, const
         {
             delta_last_try_split_rows = delta_rows;
             delta_last_try_split_bytes = delta_bytes;
-            try_add_background_task(BackgroundTask{TaskType::Split, dm_context, seg, {}});
+            try_add_background_task(BackgroundTask{TaskType::Split, dm_context, seg});
             return true;
         }
         return false;
     };
-    auto try_fg_split = [&](const SegmentPtr & my_segment) -> bool {
+    auto try_fg_split = [&](const SegmentPtr & my_segment) {
         auto my_segment_size = my_segment->getEstimatedBytes();
         auto my_should_split = my_segment_size >= dm_context->segment_force_split_bytes;
         if (my_should_split && !my_segment->isSplitForbidden())
@@ -1334,15 +1301,6 @@ void DeltaMergeStore::checkSegmentUpdate(const DMContextPtr & dm_context, const
         }
         return false;
     };
-    auto try_bg_merge = [&]() {
-        SegmentPtr merge_sibling;
-        if (should_merge && (merge_sibling = get_merge_sibling()))
-        {
-            try_add_background_task(BackgroundTask{TaskType::Merge, dm_context, segment, merge_sibling});
-            return true;
-        }
-        return false;
-    };
     auto try_bg_compact = [&]() {
         /// Compact task should be a really low priority task.
         /// And if the segment is flushing,
@@ -1352,7 +1310,7 @@ void DeltaMergeStore::checkSegmentUpdate(const DMContextPtr & dm_context, const
         if (should_compact && !segment->isFlushing())
         {
             delta_last_try_compact_column_files = column_file_count;
-            try_add_background_task(BackgroundTask{TaskType::Compact, dm_context, segment, {}});
+            try_add_background_task(BackgroundTask{TaskType::Compact, dm_context, segment});
             return true;
         }
         return false;
@@ -1361,7 +1319,7 @@ void DeltaMergeStore::checkSegmentUpdate(const DMContextPtr & dm_context, const
         if (should_place_delta_index)
         {
             delta_last_try_place_delta_index_rows = delta_rows;
-            try_add_background_task(BackgroundTask{TaskType::PlaceIndex, dm_context, segment, {}});
+            try_add_background_task(BackgroundTask{TaskType::PlaceIndex, dm_context, segment});
             return true;
         }
         return false;
@@ -1406,8 +1364,6 @@ void DeltaMergeStore::checkSegmentUpdate(const DMContextPtr & dm_context, const
         if (try_bg_merge_delta())
             return;
     }
-    if (try_bg_merge())
-        return;
     if (try_bg_compact())
         return;
     if (try_place_delta_index())

dbms/src/Storages/DeltaMerge/DeltaMergeStore.h

@@ -173,7 +173,6 @@ class DeltaMergeStore : private boost::noncopyable
     enum TaskType
     {
         Split,
-        Merge,
         MergeDelta,
         Compact,
         Flush,
@@ -221,8 +220,6 @@ class DeltaMergeStore : private boost::noncopyable
         {
         case Split:
             return "Split";
-        case Merge:
-            return "Merge";
         case MergeDelta:
             return "MergeDelta";
         case Compact:
@@ -259,7 +256,6 @@ class DeltaMergeStore : private boost::noncopyable
 
         DMContextPtr dm_context;
         SegmentPtr segment;
-        SegmentPtr next_segment;
 
         explicit operator bool() const { return segment != nullptr; }
     };
@@ -405,6 +401,27 @@ class DeltaMergeStore : private boost::noncopyable
     /// Iterator over all segments and apply gc jobs.
     UInt64 onSyncGc(Int64 limit);
 
+    /**
+     * Try to merge the segment in the current thread as the GC operation.
+     * This function may be blocking, and should be called in the GC background thread.
+     */
+    SegmentPtr gcTrySegmentMerge(const DMContextPtr & dm_context, const SegmentPtr & segment);
+
+    /**
+     * Try to merge delta in the current thread as the GC operation.
+     * This function may be blocking, and should be called in the GC background thread.
+     */
+    SegmentPtr gcTrySegmentMergeDelta(const DMContextPtr & dm_context, const SegmentPtr & segment, const SegmentPtr & prev_segment, const SegmentPtr & next_segment, DB::Timestamp gc_safe_point);
+
+    /**
+     * Starting from the given base segment, find continuous segments that could be merged.
+     *
+     * When there are mergeable segments, the baseSegment is returned in index 0 and mergeable segments are then placed in order.
+     *   It is ensured that there are at least 2 elements in the returned vector.
+     * When there is no mergeable segment, the returned vector will be empty.
+     */
+    std::vector<SegmentPtr> getMergeableSegments(const DMContextPtr & context, const SegmentPtr & baseSegment);
+
     /// Apply DDL `commands` on `table_columns`
     void applyAlters(const AlterCommands & commands, //
                      OptionTableInfoConstRef table_info,