Storages: Fix VersionChain::getBytes (#10322)

ref #10323

Co-authored-by: JaySon <tshent@qq.com>

Author: JinheLin

dbms/src/Storages/DeltaMerge/Remote/RNMVCCIndexCache.h

@@ -19,6 +19,7 @@
 #include <Storages/DeltaMerge/VersionChain/VersionChain_fwd.h>
 #include <Storages/KVStore/Types.h>
 #include <common/types.h>
+#include <fmt/format.h>
 
 #include <boost/noncopyable.hpp>
 
@@ -57,6 +58,20 @@ class RNMVCCIndexCache : private boost::noncopyable
                 && segment_id == other.segment_id && segment_epoch == other.segment_epoch
                 && delta_index_epoch == other.delta_index_epoch && is_version_chain == other.is_version_chain;
         }
+
+        String toString() const
+        {
+            return fmt::format(
+                "<store_id={}, table_id={}, segment_id={}, segment_epoch={}, "
+                "delta_index_epoch={}, keyspace={}, is_version_chain={}>",
+                store_id,
+                table_id,
+                segment_id,
+                segment_epoch,
+                delta_index_epoch,
+                keyspace_id,
+                is_version_chain);
+        }
     };
 
     // Returns a cached or newly created delta index, which is assigned to the specified segment(at)epoch.

dbms/src/Storages/DeltaMerge/SegmentReadTask.cpp

@@ -871,8 +871,24 @@ void SegmentReadTask::updateMVCCIndexSize(ReadMode read_mode, size_t initial_ind
 
     auto & cache = dm_context->global_context.getSharedContextDisagg()->rn_mvcc_index_cache;
     assert(cache != nullptr);
-    if (cache_key->is_version_chain && getVersionChainBytes(*(segment->getVersionChain())) != initial_index_bytes)
-        cache->setVersionChain(*cache_key, segment->getVersionChain());
+    static constexpr size_t LOGGING_INDEX_BYTES_THRESHOLD = 1024 * 1024;
+    if (cache_key->is_version_chain)
+    {
+        auto current_index_bytes = getVersionChainBytes(*(segment->getVersionChain()));
+        if (current_index_bytes >= LOGGING_INDEX_BYTES_THRESHOLD)
+        {
+            LOG_INFO(
+                read_snapshot->log,
+                "Version chain index size is {}, cache key is {}, initial_index_bytes is {}",
+                current_index_bytes,
+                cache_key->toString(),
+                initial_index_bytes);
+        }
+        if (current_index_bytes != initial_index_bytes)
+        {
+            cache->setVersionChain(*cache_key, segment->getVersionChain());
+        }
+    }
     else if (
         !cache_key->is_version_chain && read_snapshot->delta->getSharedDeltaIndex()->getBytes() != initial_index_bytes)
         cache->setDeltaIndex(*cache_key, read_snapshot->delta->getSharedDeltaIndex());

dbms/src/Storages/DeltaMerge/VersionChain/VersionChain.cpp

@@ -393,6 +393,7 @@ DeltaValueReader VersionChain<HandleType>::createDeltaValueReader(
 template <ExtraHandleType HandleType>
 size_t VersionChain<HandleType>::getBytes() const
 {
+    std::lock_guard lock(mtx);
     // Ignore the size of `dmfile_or_delete_range_list` because it is small.
     return base_versions->capacity() * sizeof(RowID) + new_handle_to_row_ids.getBytes();
 }

dbms/src/Storages/DeltaMerge/VersionChain/VersionChain.h

@@ -142,7 +142,7 @@ class VersionChain
     friend class tests::NewHandleIndexTest;
     friend class tests::VersionChainTest;
 
-    std::mutex mtx;
+    mutable std::mutex mtx;
     UInt32 replayed_rows_and_deletes = 0; // delta.getRows() + delta.getDeletes()
     // After replaySnapshot, base_versions->size() == delta.getRows().
     // The records in delta correspond one-to-one with base_versions.