Storage: Refine logging about reading snap, place index, index to wait

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

@@ -340,7 +340,7 @@ bool ColumnFileSetReader::shouldPlace(
         }
         else
         {
-            throw Exception("Unknown column file: " + column_file->toString(), ErrorCodes::LOGICAL_ERROR);
+            throw Exception(ErrorCodes::LOGICAL_ERROR, "Unknown column file: {}", column_file->toString());
         }
     }
 

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

@@ -16,9 +16,7 @@
 #include <Storages/DeltaMerge/ColumnFile/ColumnFileSetSnapshot.h>
 #include <Storages/DeltaMerge/DMContext.h>
 
-namespace DB
-{
-namespace DM
+namespace DB::DM
 {
 RowKeyRange ColumnFileSetSnapshot::getSquashDeleteRange() const
 {
@@ -30,5 +28,4 @@ RowKeyRange ColumnFileSetSnapshot::getSquashDeleteRange() const
     }
     return squashed_delete_range;
 }
-} // namespace DM
-} // namespace DB
+} // namespace DB::DM

dbms/src/Storages/DeltaMerge/ColumnFile/ColumnFileSetSnapshot.h

@@ -17,9 +17,7 @@
 #include <Storages/DeltaMerge/ColumnFile/ColumnFile.h>
 #include <Storages/DeltaMerge/Remote/Serializer_fwd.h>
 
-namespace DB
-{
-namespace DM
+namespace DB::DM
 {
 class ColumnFileSetSnapshot;
 using ColumnFileSetSnapshotPtr = std::shared_ptr<ColumnFileSetSnapshot>;
@@ -107,5 +105,4 @@ class ColumnFileSetSnapshot
     const auto & getDataProvider() const { return data_provider; }
 };
 
-} // namespace DM
-} // namespace DB
+} // namespace DB::DM

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

@@ -350,7 +350,7 @@ class DeltaValueSnapshot
 #else
 public:
 #endif
-    bool is_update{false};
+    const bool is_update{false};
 
     // The delta index of cached.
     DeltaIndexPtr shared_delta_index;
@@ -371,8 +371,7 @@ class DeltaValueSnapshot
         // We only allow one for_update snapshots to exist, so it cannot be cloned.
         RUNTIME_CHECK(!is_update);
 
-        auto c = std::make_shared<DeltaValueSnapshot>(type);
-        c->is_update = is_update;
+        auto c = std::make_shared<DeltaValueSnapshot>(type, is_update);
         c->shared_delta_index = shared_delta_index;
         c->delta_index_epoch = delta_index_epoch;
         c->mem_table_snap = mem_table_snap->clone();
@@ -383,8 +382,9 @@ class DeltaValueSnapshot
         return c;
     }
 
-    explicit DeltaValueSnapshot(CurrentMetrics::Metric type_)
-        : type(type_)
+    explicit DeltaValueSnapshot(CurrentMetrics::Metric type_, bool update_)
+        : is_update(update_)
+        , type(type_)
     {
         CurrentMetrics::add(type);
     }

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

@@ -38,8 +38,7 @@ DeltaSnapshotPtr DeltaValueSpace::createSnapshot(
     if (abandoned.load(std::memory_order_relaxed))
         return {};
 
-    auto snap = std::make_shared<DeltaValueSnapshot>(type);
-    snap->is_update = for_update;
+    auto snap = std::make_shared<DeltaValueSnapshot>(type, for_update);
     snap->delta = this->shared_from_this();
 
     auto storage_snap = std::make_shared<StorageSnapshot>(
@@ -220,16 +219,18 @@ bool DeltaValueReader::shouldPlace(
 {
     auto [placed_rows, placed_delete_ranges] = my_delta_index->getPlacedStatus();
 
-    // Already placed.
+    // The placed_rows, placed_delete_range already contains the data in delta_snap
     if (placed_rows >= delta_snap->getRows() && placed_delete_ranges == delta_snap->getDeletes())
         return false;
 
-    if (relevant_range.all() || relevant_range == segment_range_ //
+    if (relevant_range.all() || relevant_range == segment_range_ // read all the data in this segment
         || delta_snap->getRows() - placed_rows > context.delta_cache_limit_rows //
-        || placed_delete_ranges != delta_snap->getDeletes())
+        || placed_delete_ranges != delta_snap->getDeletes() // new delete_range appended, must place it
+    )
         return true;
 
-    size_t rows_in_persisted_file_snap = delta_snap->getMemTableSetRowsOffset();
+    // otherwise check persisted_files and mem_tables
+    const size_t rows_in_persisted_file_snap = delta_snap->getMemTableSetRowsOffset();
     return persisted_files_reader->shouldPlace(context, relevant_range, start_ts, placed_rows)
         || mem_table_reader->shouldPlace(
             context,

dbms/src/Storages/DeltaMerge/DeltaIndex.h

@@ -18,9 +18,7 @@
 #include <Storages/DeltaMerge/DeltaTree.h>
 #include <Storages/DeltaMerge/Remote/RNDeltaIndexCache.h>
 #include <Storages/Page/PageDefinesBase.h>
-namespace DB
-{
-namespace DM
+namespace DB::DM
 {
 class DeltaIndex;
 using DeltaIndexPtr = std::shared_ptr<DeltaIndex>;
@@ -171,6 +169,7 @@ class DeltaIndex
         return delta_tree->getBytes();
     }
 
+    // Return <placed_rows, placed_deletes>
     std::pair<size_t, size_t> getPlacedStatus()
     {
         std::scoped_lock lock(mutex);
@@ -219,5 +218,4 @@ class DeltaIndex
     const std::optional<Remote::RNDeltaIndexCache::CacheKey> & getRNCacheKey() const { return rn_cache_key; }
 };
 
-} // namespace DM
-} // namespace DB
+} // namespace DB::DM

dbms/src/Storages/DeltaMerge/Remote/Serializer.cpp

@@ -135,8 +135,7 @@ SegmentSnapshotPtr Serializer::deserializeSegment(
 
     auto data_store = dm_context.global_context.getSharedContextDisagg()->remote_data_store;
 
-    auto delta_snap = std::make_shared<DeltaValueSnapshot>(CurrentMetrics::DT_SnapshotOfDisaggReadNodeRead);
-    delta_snap->is_update = false;
+    auto delta_snap = std::make_shared<DeltaValueSnapshot>(CurrentMetrics::DT_SnapshotOfDisaggReadNodeRead, false);
     delta_snap->mem_table_snap
         = deserializeColumnFileSet(dm_context, proto.column_files_memtable(), data_store, segment_range);
     delta_snap->persisted_files_snap

dbms/src/Storages/DeltaMerge/Segment.cpp

@@ -39,6 +39,7 @@
 #include <Storages/DeltaMerge/Filter/FilterHelper.h>
 #include <Storages/DeltaMerge/LateMaterializationBlockInputStream.h>
 #include <Storages/DeltaMerge/PKSquashingBlockInputStream.h>
+#include <Storages/DeltaMerge/Range.h>
 #include <Storages/DeltaMerge/Remote/DataStore/DataStore.h>
 #include <Storages/DeltaMerge/Remote/ObjectId.h>
 #include <Storages/DeltaMerge/Remote/RNDeltaIndexCache.h>
@@ -121,6 +122,23 @@ extern const int UNKNOWN_FORMAT_VERSION;
 
 namespace DM
 {
+String SegmentSnapshot::detailInfo() const
+{
+    return fmt::format(
+        "{{"
+        "stable_rows={} "
+        "persisted_rows={} persisted_dels={} persisted_cfs={} "
+        "mem_rows={} mem_dels={} mem_cfs={}"
+        "}}",
+        stable->getRows(),
+        delta->getPersistedFileSetSnapshot()->getRows(),
+        delta->getPersistedFileSetSnapshot()->getDeletes(),
+        delta->getPersistedFileSetSnapshot()->getColumnFileCount(),
+        delta->getMemTableSetSnapshot()->getRows(),
+        delta->getMemTableSetSnapshot()->getDeletes(),
+        delta->getMemTableSetSnapshot()->getColumnFileCount());
+}
+
 const static size_t SEGMENT_BUFFER_SIZE = 128; // More than enough.
 
 DMFilePtr writeIntoNewDMFile(
@@ -2401,7 +2419,7 @@ Segment::ReadInfo Segment::getReadInfo(
     ReadTag read_tag,
     UInt64 start_ts) const
 {
-    LOG_DEBUG(segment_snap->log, "Begin segment getReadInfo");
+    LOG_DEBUG(segment_snap->log, "Begin segment getReadInfo {}", simpleInfo());
 
     auto new_read_columns = arrangeReadColumns(getExtraHandleColumnDefine(is_common_handle), read_columns);
     auto pk_ver_col_defs = std::make_shared<ColumnDefines>(
@@ -2417,20 +2435,30 @@ Segment::ReadInfo Segment::getReadInfo(
 
     auto [my_delta_index, fully_indexed] = ensurePlace(dm_context, segment_snap, delta_reader, read_ranges, start_ts);
     auto compacted_index = my_delta_index->getDeltaTree()->getCompactedEntries();
-
-
     // Hold compacted_index reference, to prevent it from deallocated.
     delta_reader->setDeltaIndex(compacted_index);
 
-    LOG_DEBUG(segment_snap->log, "Finish segment getReadInfo");
+    LOG_DEBUG(
+        segment_snap->log,
+        "Finish segment getReadInfo, my_delta_index={} fully_indexed={} read_ranges={} "
+        "snap={} {}",
+        my_delta_index->toString(),
+        fully_indexed,
+        DB::DM::toDebugString(read_ranges),
+        segment_snap->detailInfo(),
+        simpleInfo());
 
     if (fully_indexed)
     {
         // Try update shared index, if my_delta_index is more advanced.
         bool ok = segment_snap->delta->getSharedDeltaIndex()->updateIfAdvanced(*my_delta_index);
         if (ok)
         {
-            LOG_DEBUG(segment_snap->log, "Segment updated delta index");
+            LOG_DEBUG(
+                segment_snap->log,
+                "Segment updated delta index, my_delta_index={} {}",
+                my_delta_index->toString(),
+                simpleInfo());
             // Update cache size.
             if (auto cache = dm_context.global_context.getSharedContextDisagg()->rn_delta_index_cache; cache)
                 cache->setDeltaIndex(segment_snap->delta->getSharedDeltaIndex());
@@ -2539,8 +2567,8 @@ std::pair<DeltaIndexPtr, bool> Segment::ensurePlace(
     auto my_delta_index = delta_snap->getSharedDeltaIndex()->tryClone(delta_snap->getRows(), delta_snap->getDeletes());
     auto my_delta_tree = my_delta_index->getDeltaTree();
 
-    bool relevant_place = dm_context.enable_relevant_place;
-    bool skippable_place = dm_context.enable_skippable_place;
+    const bool relevant_place = dm_context.enable_relevant_place;
+    const bool skippable_place = dm_context.enable_skippable_place;
 
     // Note that, when enable_relevant_place is false , we cannot use the range of this segment.
     // Because some block / delete ranges could contain some data / range that are not belong to current segment.
@@ -2552,7 +2580,9 @@ std::pair<DeltaIndexPtr, bool> Segment::ensurePlace(
 
     // Let's do a fast check, determine whether we need to do place or not.
     if (!delta_reader->shouldPlace(dm_context, my_delta_index, rowkey_range, relevant_range, start_ts))
+    {
         return {my_delta_index, false};
+    }
 
     CurrentMetrics::Increment cur_dm_segments{CurrentMetrics::DT_PlaceIndexUpdate};
     GET_METRIC(tiflash_storage_subtask_count, type_place_index_update).Increment();
@@ -2578,8 +2608,11 @@ std::pair<DeltaIndexPtr, bool> Segment::ensurePlace(
             auto offset = v.getBlockOffset();
             auto rows = block.rows();
 
-            if (unlikely(my_placed_rows != offset))
-                throw Exception("Place block offset not match", ErrorCodes::LOGICAL_ERROR);
+            RUNTIME_CHECK_MSG(
+                my_placed_rows == offset,
+                "Place block offset not match, my_placed_rows={} offset={}",
+                my_placed_rows,
+                offset);
 
             if (skippable_place)
                 fully_indexed &= placeUpsert<true>(
@@ -2629,25 +2662,24 @@ std::pair<DeltaIndexPtr, bool> Segment::ensurePlace(
         }
     }
 
-    if (unlikely(my_placed_rows != delta_snap->getRows() || my_placed_deletes != delta_snap->getDeletes()))
-    {
-        throw Exception(fmt::format(
-            "Placed status not match! Expected place rows:{}, deletes:{}, but actually placed rows:{}, deletes:{}",
-            delta_snap->getRows(),
-            delta_snap->getDeletes(),
-            my_placed_rows,
-            my_placed_deletes));
-    }
+    RUNTIME_CHECK_MSG(
+        my_placed_rows == delta_snap->getRows() && my_placed_deletes == delta_snap->getDeletes(),
+        "Placed status not match! Expected place rows:{}, deletes:{}, but actually placed rows:{}, deletes:{}",
+        delta_snap->getRows(),
+        delta_snap->getDeletes(),
+        my_placed_rows,
+        my_placed_deletes);
 
     my_delta_index->update(my_delta_tree, my_placed_rows, my_placed_deletes);
 
     LOG_DEBUG(
         segment_snap->log,
-        "Finish segment ensurePlace, read_ranges={} placed_items={} shared_delta_index={} my_delta_index={}",
+        "Finish segment ensurePlace, read_ranges={} placed_items={} shared_delta_index={} my_delta_index={} {}",
         DB::DM::toDebugString(read_ranges),
         items.size(),
         delta_snap->getSharedDeltaIndex()->toString(),
-        my_delta_index->toString());
+        my_delta_index->toString(),
+        simpleInfo());
 
     return {my_delta_index, fully_indexed};
 }

dbms/src/Storages/DeltaMerge/Segment.h

@@ -73,6 +73,8 @@ struct SegmentSnapshot : private boost::noncopyable
         // handle + version + flag
         return (sizeof(Int64) + sizeof(UInt64) + sizeof(UInt8)) * getRows();
     }
+
+    String detailInfo() const;
 };
 
 /// A segment contains many rows of a table. A table is split into segments by consecutive ranges.

dbms/src/Storages/KVStore/Read/LearnerReadWorker.cpp

@@ -341,7 +341,7 @@ RegionsReadIndexResult LearnerReadWorker::readIndex(
 
 void LearnerReadWorker::waitIndex(
     const LearnerReadSnapshot & regions_snapshot,
-    RegionsReadIndexResult & batch_read_index_result,
+    const RegionsReadIndexResult & batch_read_index_result,
     const UInt64 timeout_ms,
     Stopwatch & watch)
 {
@@ -436,7 +436,7 @@ void LearnerReadWorker::waitIndex(
         unavailable_regions.size(),
         mvcc_query_info.start_ts);
 
-    auto bypass_formatter = [&](const RegionQueryInfo & query_info) -> String {
+    auto bypass_formatter = [](const RegionQueryInfo & query_info) -> String {
         if (query_info.bypass_lock_ts == nullptr)
             return "";
         FmtBuffer buffer;
@@ -456,9 +456,11 @@ void LearnerReadWorker::waitIndex(
             regions_info.end(),
             [&](const auto & region_to_query, FmtBuffer & f) {
                 const auto & region = regions_snapshot.find(region_to_query.region_id)->second;
+                const auto index_to_wait = batch_read_index_result.find(region_to_query.region_id)->second.read_index();
                 f.fmtAppend(
-                    "(id:{} applied_index:{} bypass_locks:{})",
+                    "(region_id={} to_wait={} applied_index={} bypass_locks={})",
                     region_to_query.region_id,
+                    index_to_wait,
                     region->appliedIndex(),
                     bypass_formatter(region_to_query));
             },

dbms/src/Storages/KVStore/Read/LearnerReadWorker.h

@@ -150,7 +150,7 @@ class LearnerReadWorker
     /// wait index relate methods
     void waitIndex(
         const LearnerReadSnapshot & regions_snapshot,
-        RegionsReadIndexResult & batch_read_index_result,
+        const RegionsReadIndexResult & batch_read_index_result,
         UInt64 timeout_ms,
         Stopwatch & watch);