fix empty segment cannot merge after gc

dbms/src/Debug/dbgFuncMisc.cpp

@@ -22,28 +22,73 @@
 
 namespace DB
 {
+inline size_t getThreadIdForLog(const String & line)
+{
+    auto sub_line = line.substr(line.find("thread_id="));
+    std::regex rx(R"((0|[1-9][0-9]*))");
+    std::smatch m;
+    if (regex_search(sub_line, m, rx))
+        return std::stoi(m[1]);
+    else
+        return 0;
+}
+
+// Usage example:
+// The first argument is the key you want to search.
+// For example, we want to search the key 'RSFilter exclude rate' in log file, and get the value following it.
+// So we can use it as the first argument.
+// But many kind of thread can print this keyword,
+// so we can use the second argument to specify a keyword that may just be printed by a specific kind of thread.
+// Here we use 'Rough set filter' to specify we just want to search read thread.
+// And the complete command is the following:
+//   DBGInvoke search_log_for_key('RSFilter exclude rate', 'Rough set filter')
+// TODO: this is still a too hack way to do test, but cannot think a better way now.
 void dbgFuncSearchLogForKey(Context & context, const ASTs & args, DBGInvoker::Printer output)
 {
-    if (args.size() < 1)
-        throw Exception("Args not matched, should be: key", ErrorCodes::BAD_ARGUMENTS);
+    if (args.size() < 2)
+        throw Exception("Args not matched, should be: key, thread_hint", ErrorCodes::BAD_ARGUMENTS);
 
     String key = safeGet<String>(typeid_cast<const ASTLiteral &>(*args[0]).value);
+    // the candidate line must be printed by a thread which also print a line contains `thread_hint`
+    String thread_hint = safeGet<String>(typeid_cast<const ASTLiteral &>(*args[1]).value);
     auto log_path = context.getConfigRef().getString("logger.log");
 
     std::ifstream file(log_path);
-    std::vector<String> line_candidates;
-    String line;
-    while (std::getline(file, line))
+    // get the lines containing `thread_hint` and `key`
+    std::vector<String> thread_hint_line_candidates;
+    std::vector<String> key_line_candidates;
     {
-        if ((line.find(key) != String::npos) && (line.find("DBGInvoke") == String::npos))
-            line_candidates.emplace_back(line);
+        String line;
+        while (std::getline(file, line))
+        {
+            if ((line.find(thread_hint) != String::npos) && (line.find("DBGInvoke") == String::npos))
+                thread_hint_line_candidates.emplace_back(line);
+            else if ((line.find(key) != String::npos) && (line.find("DBGInvoke") == String::npos))
+                key_line_candidates.emplace_back(line);
+        }
     }
-    if (line_candidates.empty())
+    // get target thread id
+    if (thread_hint_line_candidates.empty() || key_line_candidates.empty())
     {
         output("Invalid");
         return;
     }
-    auto & target_line = line_candidates.back();
+    size_t target_thread_id = getThreadIdForLog(thread_hint_line_candidates.back());
+    if (target_thread_id == 0)
+    {
+        output("Invalid");
+        return;
+    }
+    String target_line;
+    for (auto iter = key_line_candidates.rbegin(); iter != key_line_candidates.rend(); iter++)
+    {
+        if (getThreadIdForLog(*iter) == target_thread_id)
+        {
+            target_line = *iter;
+            break;
+        }
+    }
+    // try parse the first number following the key
     auto sub_line = target_line.substr(target_line.find(key));
     std::regex rx(R"([+-]?([0-9]+([.][0-9]*)?|[.][0-9]+))");
     std::smatch m;

dbms/src/Storages/DeltaMerge/ColumnFile/ColumnFileBig.cpp

@@ -35,8 +35,16 @@ void ColumnFileBig::calculateStat(const DMContext & context)
     auto index_cache = context.db_context.getGlobalContext().getMinMaxIndexCache();
     auto hash_salt = context.hash_salt;
 
-    auto pack_filter
-        = DMFilePackFilter::loadFrom(file, index_cache, hash_salt, {segment_range}, EMPTY_FILTER, {}, context.db_context.getFileProvider(), context.getReadLimiter());
+    auto pack_filter = DMFilePackFilter::loadFrom(
+        file,
+        index_cache,
+        hash_salt,
+        /*set_cache_if_miss*/ false,
+        {segment_range},
+        EMPTY_FILTER,
+        {},
+        context.db_context.getFileProvider(),
+        context.getReadLimiter());
 
     std::tie(valid_rows, valid_bytes) = pack_filter.validRowsAndBytes();
 }

dbms/src/Storages/DeltaMerge/DeltaMergeStore.cpp

@@ -1593,7 +1593,7 @@ UInt64 DeltaMergeStore::onSyncGc(Int64 limit)
         }
 
         assert(segment != nullptr);
-        if (segment->hasAbandoned() || segment->getLastCheckGCSafePoint() >= gc_safe_point || segment_snap == nullptr)
+        if (segment->hasAbandoned() || segment_snap == nullptr)
             continue;
 
         const auto segment_id = segment->segmentId();
@@ -1602,43 +1602,52 @@ UInt64 DeltaMergeStore::onSyncGc(Int64 limit)
         // meet empty segment, try merge it
         if (segment_snap->getRows() == 0)
         {
+            // release segment_snap before checkSegmentUpdate, otherwise this segment is still in update status.
+            segment_snap = {};
             checkSegmentUpdate(dm_context, segment, ThreadType::BG_GC);
             continue;
         }
 
-        // Avoid recheck this segment when gc_safe_point doesn't change regardless whether we trigger this segment's DeltaMerge or not.
-        // Because after we calculate StableProperty and compare it with this gc_safe_point,
-        // there is no need to recheck it again using the same gc_safe_point.
-        // On the other hand, if it should do DeltaMerge using this gc_safe_point, and the DeltaMerge is interruptted by other process,
-        // it's still worth to wait another gc_safe_point to check this segment again.
-        segment->setLastCheckGCSafePoint(gc_safe_point);
-        dm_context->min_version = gc_safe_point;
-
-        // calculate StableProperty if needed
-        if (!segment->getStable()->isStablePropertyCached())
-            segment->getStable()->calculateStableProperty(*dm_context, segment_range, isCommonHandle());
-
         try
         {
             // Check whether we should apply gc on this segment
-            const bool should_compact
-                = GC::shouldCompactStable(
-                      segment,
-                      gc_safe_point,
-                      global_context.getSettingsRef().dt_bg_gc_ratio_threhold_to_trigger_gc,
-                      log)
-                || GC::shouldCompactDeltaWithStable(
-                      *dm_context,
-                      segment_snap,
-                      segment_range,
-                      global_context.getSettingsRef().dt_bg_gc_delta_delete_ratio_to_trigger_gc,
-                      log);
+            bool should_compact = false;
+            if (GC::shouldCompactDeltaWithStable(
+                    *dm_context,
+                    segment_snap,
+                    segment_range,
+                    global_context.getSettingsRef().dt_bg_gc_delta_delete_ratio_to_trigger_gc,
+                    log))
+            {
+                should_compact = true;
+            }
+            else if (segment->getLastCheckGCSafePoint() < gc_safe_point)
+            {
+                // Avoid recheck this segment when gc_safe_point doesn't change regardless whether we trigger this segment's DeltaMerge or not.
+                // Because after we calculate StableProperty and compare it with this gc_safe_point,
+                // there is no need to recheck it again using the same gc_safe_point.
+                // On the other hand, if it should do DeltaMerge using this gc_safe_point, and the DeltaMerge is interruptted by other process,
+                // it's still worth to wait another gc_safe_point to check this segment again.
+                segment->setLastCheckGCSafePoint(gc_safe_point);
+                dm_context->min_version = gc_safe_point;
+
+                // calculate StableProperty if needed
+                if (!segment->getStable()->isStablePropertyCached())
+                    segment->getStable()->calculateStableProperty(*dm_context, segment_range, isCommonHandle());
+
+                should_compact = GC::shouldCompactStable(
+                    segment,
+                    gc_safe_point,
+                    global_context.getSettingsRef().dt_bg_gc_ratio_threhold_to_trigger_gc,
+                    log);
+            }
             bool finish_gc_on_segment = false;
             if (should_compact)
             {
                 if (segment = segmentMergeDelta(*dm_context, segment, TaskRunThread::BackgroundGCThread, segment_snap); segment)
                 {
                     // Continue to check whether we need to apply more tasks on this segment
+                    segment_snap = {};
                     checkSegmentUpdate(dm_context, segment, ThreadType::BG_GC);
                     gc_segments_num++;
                     finish_gc_on_segment = true;

dbms/src/Storages/DeltaMerge/File/DMFilePackFilter.h

@@ -42,13 +42,14 @@ class DMFilePackFilter
     static DMFilePackFilter loadFrom(const DMFilePtr & dmfile,
                                      const MinMaxIndexCachePtr & index_cache,
                                      UInt64 hash_salt,
+                                     bool set_cache_if_miss,
                                      const RowKeyRanges & rowkey_ranges,
                                      const RSOperatorPtr & filter,
                                      const IdSetPtr & read_packs,
                                      const FileProviderPtr & file_provider,
                                      const ReadLimiterPtr & read_limiter)
     {
-        auto pack_filter = DMFilePackFilter(dmfile, index_cache, hash_salt, rowkey_ranges, filter, read_packs, file_provider, read_limiter);
+        auto pack_filter = DMFilePackFilter(dmfile, index_cache, hash_salt, set_cache_if_miss, rowkey_ranges, filter, read_packs, file_provider, read_limiter);
         pack_filter.init();
         return pack_filter;
     }
@@ -101,6 +102,7 @@ class DMFilePackFilter
     DMFilePackFilter(const DMFilePtr & dmfile_,
                      const MinMaxIndexCachePtr & index_cache_,
                      UInt64 hash_salt_,
+                     bool set_cache_if_miss_,
                      const RowKeyRanges & rowkey_ranges_, // filter by handle range
                      const RSOperatorPtr & filter_, // filter by push down where clause
                      const IdSetPtr & read_packs_, // filter by pack index
@@ -109,6 +111,7 @@ class DMFilePackFilter
         : dmfile(dmfile_)
         , index_cache(index_cache_)
         , hash_salt(hash_salt_)
+        , set_cache_if_miss(set_cache_if_miss_)
         , rowkey_ranges(rowkey_ranges_)
         , filter(filter_)
         , read_packs(read_packs_)
@@ -214,6 +217,7 @@ class DMFilePackFilter
                           const DMFilePtr & dmfile,
                           const FileProviderPtr & file_provider,
                           const MinMaxIndexCachePtr & index_cache,
+                          bool set_cache_if_miss,
                           ColId col_id,
                           const ReadLimiterPtr & read_limiter)
     {
@@ -250,13 +254,17 @@ class DMFilePackFilter
             }
         };
         MinMaxIndexPtr minmax_index;
-        if (index_cache)
+        if (index_cache && set_cache_if_miss)
         {
             minmax_index = index_cache->getOrSet(dmfile->colIndexCacheKey(file_name_base), load);
         }
         else
         {
-            minmax_index = load();
+            // try load from the cache first
+            if (index_cache)
+                minmax_index = index_cache->get(dmfile->colIndexCacheKey(file_name_base));
+            if (!minmax_index)
+                minmax_index = load();
         }
         indexes.emplace(col_id, RSIndex(type, minmax_index));
     }
@@ -269,13 +277,14 @@ class DMFilePackFilter
         if (!dmfile->isColIndexExist(col_id))
             return;
 
-        loadIndex(param.indexes, dmfile, file_provider, index_cache, col_id, read_limiter);
+        loadIndex(param.indexes, dmfile, file_provider, index_cache, set_cache_if_miss, col_id, read_limiter);
     }
 
 private:
     DMFilePtr dmfile;
     MinMaxIndexCachePtr index_cache;
     UInt64 hash_salt;
+    bool set_cache_if_miss;
     RowKeyRanges rowkey_ranges;
     RSOperatorPtr filter;
     IdSetPtr read_packs;

dbms/src/Storages/DeltaMerge/File/DMFileReader.cpp

@@ -222,7 +222,7 @@ DMFileReader::DMFileReader(
     , read_columns(read_columns_)
     , enable_clean_read(enable_clean_read_)
     , max_read_version(max_read_version_)
-    , pack_filter(dmfile_, index_cache_, hash_salt_, rowkey_ranges_, filter_, read_packs_, file_provider_, read_limiter)
+    , pack_filter(dmfile_, index_cache_, hash_salt_, /*set_cache_if_miss*/ true, rowkey_ranges_, filter_, read_packs_, file_provider_, read_limiter)
     , handle_res(pack_filter.getHandleRes())
     , use_packs(pack_filter.getUsePacks())
     , skip_packs_by_column(read_columns.size(), 0)
@@ -315,7 +315,7 @@ Block DMFileReader::read()
     // 0 means no limit
     size_t read_pack_limit = (single_file_mode || read_one_pack_every_time) ? 1 : 0;
 
-    auto & pack_stats = dmfile->getPackStats();
+    const auto & pack_stats = dmfile->getPackStats();
     size_t read_rows = 0;
     size_t not_clean_rows = 0;
 
@@ -382,7 +382,7 @@ Block DMFileReader::read()
                 }
                 else if (cd.id == TAG_COLUMN_ID)
                 {
-                    column = cd.type->createColumnConst(read_rows, Field((UInt64)(pack_stats[start_pack_id].first_tag)));
+                    column = cd.type->createColumnConst(read_rows, Field(static_cast<UInt64>(pack_stats[start_pack_id].first_tag)));
                 }
 
                 res.insert(ColumnWithTypeAndName{column, cd.type, cd.name, cd.id});

dbms/src/Storages/DeltaMerge/File/DMFileReader.h

@@ -75,7 +75,7 @@ class DMFileReader
         // If you have no idea what it means, then simply set it to false.
         bool enable_clean_read_,
         // The the MVCC filter version. Used by clean read check.
-        UInt64 max_data_version_,
+        UInt64 max_read_version_,
         // filters
         const RowKeyRanges & rowkey_ranges_,
         const RSOperatorPtr & filter_,

dbms/src/Storages/DeltaMerge/File/DMFileWriter.cpp

@@ -128,17 +128,17 @@ void DMFileWriter::write(const Block & block, const BlockProperty & block_proper
 
     auto del_mark_column = tryGetByColumnId(block, TAG_COLUMN_ID).column;
 
-    const ColumnVector<UInt8> * del_mark = !del_mark_column ? nullptr : (const ColumnVector<UInt8> *)del_mark_column.get();
+    const ColumnVector<UInt8> * del_mark = !del_mark_column ? nullptr : static_cast<const ColumnVector<UInt8> *>(del_mark_column.get());
 
     for (auto & cd : write_columns)
     {
-        auto & col = getByColumnId(block, cd.id).column;
+        const auto & col = getByColumnId(block, cd.id).column;
         writeColumn(cd.id, *cd.type, *col, del_mark);
 
         if (cd.id == VERSION_COLUMN_ID)
             stat.first_version = col->get64(0);
         else if (cd.id == TAG_COLUMN_ID)
-            stat.first_tag = (UInt8)(col->get64(0));
+            stat.first_tag = static_cast<UInt8>(col->get64(0));
     }
 
     if (!options.flags.isSingleFile())

dbms/src/Storages/DeltaMerge/StableValueSpace.cpp

@@ -53,6 +53,7 @@ void StableValueSpace::setFiles(const DMFiles & files_, const RowKeyRange & rang
             auto pack_filter = DMFilePackFilter::loadFrom(file,
                                                           index_cache,
                                                           hash_salt,
+                                                          /*set_cache_if_miss*/ true,
                                                           {range},
                                                           EMPTY_FILTER,
                                                           {},
@@ -241,6 +242,7 @@ void StableValueSpace::calculateStableProperty(const DMContext & context, const
         auto pack_filter = DMFilePackFilter::loadFrom(file,
                                                       context.db_context.getGlobalContext().getMinMaxIndexCache(),
                                                       context.hash_salt,
+                                                      /*set_cache_if_miss*/ false,
                                                       {rowkey_range},
                                                       EMPTY_FILTER,
                                                       {},
@@ -359,21 +361,21 @@ RowsAndBytes StableValueSpace::Snapshot::getApproxRowsAndBytes(const DMContext &
     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.
+    // 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.
+    // So don't refill the cache if the index does not exist.
     for (auto & f : stable->files)
     {
-        auto filter = DMFilePackFilter::loadFrom(f, //
-                                                 nullptr,
-                                                 context.hash_salt,
-                                                 {range},
-                                                 RSOperatorPtr{},
-                                                 IdSetPtr{},
-                                                 context.db_context.getFileProvider(),
-                                                 context.getReadLimiter());
+        auto filter = DMFilePackFilter::loadFrom(
+            f,
+            context.db_context.getGlobalContext().getMinMaxIndexCache(),
+            context.hash_salt,
+            /*set_cache_if_miss*/ false,
+            {range},
+            RSOperatorPtr{},
+            IdSetPtr{},
+            context.db_context.getFileProvider(),
+            context.getReadLimiter());
         const auto & pack_stats = f->getPackStats();
         const auto & use_packs = filter.getUsePacks();
         for (size_t i = 0; i < pack_stats.size(); ++i)