Do background compaction when the ratio of stable data covered by delete range is too large

.gitignore

@@ -251,6 +251,7 @@ website/presentations
 build_docker
 docker/builder/tics
 release-centos7/tiflash
+release-centos7/tiflash-*
 release-centos7/build-release
 release-darwin/tiflash
 release-darwin/build-release

dbms/src/Interpreters/Settings.h

@@ -261,6 +261,7 @@ struct Settings
     M(SettingUInt64, dt_bg_gc_check_interval, 600, "Background gc thread check interval, the unit is second.")\
     M(SettingInt64, dt_bg_gc_max_segments_to_check_every_round, 15, "Max segments to check in every gc round, value less than or equal to 0 means gc no segments.")\
     M(SettingFloat, dt_bg_gc_ratio_threhold_to_trigger_gc, 1.2, "Trigger segment's gc when the ratio of invalid version exceed this threhold. Values smaller than or equal to 1.0 means gc all segments")\
+    M(SettingFloat, dt_bg_gc_delta_delete_ratio_to_trigger_gc, 0.8, "Trigger segment's gc when the ratio of delta delete range to stable exceeds this ratio.")\
     M(SettingUInt64, dt_insert_max_rows, 0, "Max rows of insert blocks when write into DeltaTree Engine. By default 0 means no limit.")\
     M(SettingBool, dt_enable_rough_set_filter, true, "Whether to parse where expression as Rough Set Index filter or not.") \
     M(SettingBool, dt_raw_filter_range, true, "Do range filter or not when read data in raw mode in DeltaTree Engine.")\

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

@@ -319,6 +319,8 @@ class DeltaValueSnapshot : public std::enable_shared_from_this<DeltaValueSnapsho
     size_t getBytes() const { return bytes; }
     size_t getDeletes() const { return deletes; }
 
+    RowKeyRange getSquashDeleteRange() const;
+
     const auto & getStorageSnapshot() { return storage_snap; }
     const auto & getSharedDeltaIndex() { return shared_delta_index; }
 };

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

@@ -122,11 +122,22 @@ DeltaSnapshotPtr DeltaValueSpace::createSnapshot(const DMContext & context, bool
     return snap;
 }
 
+RowKeyRange DeltaValueSnapshot::getSquashDeleteRange() const
+{
+    RowKeyRange squashed_delete_range = RowKeyRange::newNone(is_common_handle, rowkey_column_size);
+    for (auto iter = packs.cbegin(); iter != packs.cend(); ++iter)
+    {
+        const auto & pack = *iter;
+        if (auto dp_delete = pack->tryToDeleteRange(); dp_delete)
+            squashed_delete_range = squashed_delete_range.merge(dp_delete->getDeleteRange());
+    }
+    return squashed_delete_range;
+}
+
 // ================================================
 // DeltaValueReader
 // ================================================
 
-
 DeltaValueReader::DeltaValueReader(const DMContext &        context,
                                    const DeltaSnapshotPtr & delta_snap_,
                                    const ColumnDefinesPtr & col_defs_,

dbms/src/Storages/DeltaMerge/DeltaMergeStore.h

@@ -172,9 +172,9 @@ class DeltaMergeStore : private boost::noncopyable
 
     enum TaskRunThread
     {
-        Thread_BG_Thread_Pool,
-        Thread_FG,
-        Thread_BG_GC,
+        BackgroundThreadPool,
+        Foreground,
+        BackgroundGCThread,
     };
 
     static std::string toString(ThreadType type)
@@ -204,21 +204,6 @@ class DeltaMergeStore : private boost::noncopyable
         }
     }
 
-    static std::string toString(TaskRunThread type)
-    {
-        switch (type)
-        {
-        case Thread_BG_Thread_Pool:
-            return "BackgroundThreadPool";
-        case Thread_FG:
-            return "Foreground";
-        case Thread_BG_GC:
-            return "BackgroundGCThread";
-        default:
-            return "Unknown";
-        }
-    }
-
     static std::string toString(TaskType type)
     {
         switch (type)
@@ -240,6 +225,21 @@ class DeltaMergeStore : private boost::noncopyable
         }
     }
 
+    static std::string toString(TaskRunThread type)
+    {
+        switch (type)
+        {
+        case BackgroundThreadPool:
+            return "BackgroundThreadPool";
+        case Foreground:
+            return "Foreground";
+        case BackgroundGCThread:
+            return "BackgroundGCThread";
+        default:
+            return "Unknown";
+        }
+    }
+
     struct BackgroundTask
     {
         TaskType type;
@@ -414,13 +414,19 @@ class DeltaMergeStore : private boost::noncopyable
 
     SegmentPair segmentSplit(DMContext & dm_context, const SegmentPtr & segment, bool is_foreground);
     void        segmentMerge(DMContext & dm_context, const SegmentPtr & left, const SegmentPtr & right, bool is_foreground);
-    SegmentPtr  segmentMergeDelta(DMContext & dm_context, const SegmentPtr & segment, const TaskRunThread thread);
+    SegmentPtr  segmentMergeDelta(DMContext &         dm_context,
+                                  const SegmentPtr &  segment,
+                                  const TaskRunThread thread,
+                                  SegmentSnapshotPtr  segment_snap = nullptr);
 
     bool updateGCSafePoint();
 
     bool handleBackgroundTask(bool heavy);
 
-    bool isSegmentValid(const SegmentPtr & segment);
+    // isSegmentValid should be protected by lock on `read_write_mutex`
+    inline bool isSegmentValid(std::shared_lock<std::shared_mutex> &, const SegmentPtr & segment) { return doIsSegmentValid(segment); }
+    inline bool isSegmentValid(std::unique_lock<std::shared_mutex> &, const SegmentPtr & segment) { return doIsSegmentValid(segment); }
+    bool        doIsSegmentValid(const SegmentPtr & segment);
 
     void restoreStableFiles();
 

dbms/src/Storages/DeltaMerge/RowKeyRange.h

@@ -1,4 +1,5 @@
 #pragma once
+#include <Columns/ColumnString.h>
 #include <Core/Types.h>
 #include <Functions/FunctionHelpers.h>
 #include <IO/WriteHelpers.h>

dbms/src/Storages/DeltaMerge/Segment.h

@@ -138,13 +138,13 @@ class Segment : private boost::noncopyable
                                        size_t                expected_block_size = DEFAULT_BLOCK_SIZE);
 
     /// Return a stream which is suitable for exporting data.
-    ///  reorgize_block: put those rows with the same pk rows into the same block or not.
+    ///  reorganize_block: put those rows with the same pk rows into the same block or not.
     BlockInputStreamPtr getInputStreamForDataExport(const DMContext &          dm_context,
                                                     const ColumnDefines &      columns_to_read,
                                                     const SegmentSnapshotPtr & segment_snap,
                                                     const RowKeyRange &        data_range,
                                                     size_t                     expected_block_size = DEFAULT_BLOCK_SIZE,
-                                                    bool                       reorgnize_block     = true) const;
+                                                    bool                       reorganize_block    = true) const;
 
     BlockInputStreamPtr getInputStreamRaw(const DMContext &          dm_context,
                                           const ColumnDefines &      columns_to_read,

dbms/src/Storages/DeltaMerge/StableValueSpace.cpp

@@ -333,15 +333,22 @@ SkippableBlockInputStreamPtr StableValueSpace::Snapshot::getInputStream(const DM
 
 RowsAndBytes StableValueSpace::Snapshot::getApproxRowsAndBytes(const DMContext & context, const RowKeyRange & range)
 {
-    if (valid_rows == 0)
+    // Avoid unnessary reading IO
+    if (valid_rows == 0 || range.none())
         return {0, 0};
+
     size_t match_packs       = 0;
     size_t total_match_rows  = 0;
     size_t total_match_bytes = 0;
+    // Usually, this method will be called for some "cold" key ranges. Loading the index
+    // into cache may pollute the cache and make the hot index cache invalid. Set the
+    // index cache to nullptr so that the cache won't be polluted.
+    // TODO: We can use the cache if the index happens to exist in the cache, but
+    // don't refill the cache if the index does not exist.
     for (auto & f : stable->files)
     {
-        auto   filter     = DMFilePackFilter::loadFrom(f,
-                                                 context.db_context.getGlobalContext().getMinMaxIndexCache(),
+        auto   filter     = DMFilePackFilter::loadFrom(f, //
+                                                 nullptr,
                                                  context.hash_salt,
                                                  range,
                                                  RSOperatorPtr{},

dbms/src/Storages/DeltaMerge/convertColumnTypeHelpers.cpp

@@ -16,7 +16,8 @@ namespace DB
 namespace ErrorCodes
 {
 extern const int BAD_ARGUMENTS;
-}
+extern const int NOT_IMPLEMENTED;
+} // namespace ErrorCodes
 
 namespace DM
 {