Add metrics

Signed-off-by: JaySon-Huang <tshent@qq.com>

Author: JaySon-Huang

dbms/src/Common/CurrentMetrics.cpp

@@ -88,7 +88,10 @@
     M(ConnectionPoolSize)                       \
     M(MemoryTrackingQueryStorageTask)           \
     M(MemoryTrackingFetchPages)                 \
-    M(MemoryTrackingSharedColumnData)
+    M(MemoryTrackingSharedColumnData)           \
+    M(DT_NumSegments)                           \
+    M(DT_NumMemTable)                           \
+    M(DT_BytesMemTable)
 
 namespace CurrentMetrics
 {

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

@@ -73,19 +73,6 @@ class ColumnFile
         INMEMORY_FILE = 4,
     };
 
-    struct Cache
-    {
-        explicit Cache(const Block & header)
-            : block(header.cloneWithColumns(header.cloneEmptyColumns()))
-        {}
-        explicit Cache(Block && block)
-            : block(std::move(block))
-        {}
-
-        std::mutex mutex;
-        Block block;
-    };
-    using CachePtr = std::shared_ptr<Cache>;
     using ColIdToOffset = std::unordered_map<ColId, size_t>;
 
 public:
@@ -138,7 +125,13 @@ class ColumnFile
     /// been persisted in the disk and their data will be immutable.
     virtual bool isAppendable() const { return false; }
     virtual void disableAppend() {}
-    virtual bool append(
+
+    struct AppendResult
+    {
+        bool success = false; // whether the append is successful
+        size_t new_alloc_bytes = 0; // the new allocated bytes after append
+    };
+    virtual AppendResult append(
         const DMContext & /*dm_context*/,
         const Block & /*data*/,
         size_t /*offset*/,

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

@@ -41,6 +41,7 @@ void ColumnFileInMemory::fillColumns(const ColumnDefines & col_defs, size_t col_
             // Copy data from cache
             const auto & type = getDataType(cd.id);
             auto col_data = type->createColumn();
+            col_data->reserve(rows);
             col_data->insertRangeFrom(*(cache->block.getByPosition(col_offset).column), 0, rows);
             // Cast if need
             auto col_converted = convertColumnByColumnDefineIfNeed(type, std::move(col_data), cd);
@@ -64,24 +65,30 @@ ColumnFileReaderPtr ColumnFileInMemory::getReader(
     return std::make_shared<ColumnFileInMemoryReader>(*this, col_defs);
 }
 
-bool ColumnFileInMemory::append(
+void ColumnFileInMemory::disableAppend() 
+{
+    disable_append = true;
+    // TODO: Call shrinkToFit() to release the extra memory of the cache block.
+}
+
+ColumnFile::AppendResult ColumnFileInMemory::append(
     const DMContext & context,
     const Block & data,
     size_t offset,
     size_t limit,
     size_t data_bytes)
 {
     if (disable_append)
-        return false;
+        return AppendResult{false, 0};
 
     std::scoped_lock lock(cache->mutex);
     if (!isSameSchema(cache->block, data))
-        return false;
+        return AppendResult{false, 0};
 
     // check whether this instance overflows
     if (cache->block.rows() >= context.delta_cache_limit_rows
         || cache->block.bytes() >= context.delta_cache_limit_bytes)
-        return false;
+        return AppendResult{false, 0};
 
     size_t new_alloc_block_bytes = 0;
     for (size_t i = 0; i < cache->block.columns(); ++i)
@@ -96,15 +103,7 @@ bool ColumnFileInMemory::append(
 
     rows += limit;
     bytes += data_bytes;
-    LOG_INFO(
-        Logger::get(),
-        "Append rows to ColumnFileInMemory, new_rows={} new_bytes={} new_alloc_bytes={} total_rows={} total_bytes={}",
-        limit - offset,
-        data_bytes,
-        new_alloc_block_bytes,
-        rows,
-        bytes);
-    return true;
+    return AppendResult{true, new_alloc_block_bytes};
 }
 
 Block ColumnFileInMemory::readDataForFlush() const

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

@@ -30,6 +30,20 @@ class ColumnFileInMemory : public ColumnFile
     friend class ColumnFileInMemoryReader;
     friend struct Remote::Serializer;
 
+    struct Cache
+    {
+        explicit Cache(const Block & header)
+            : block(header.cloneWithColumns(header.cloneEmptyColumns()))
+        {}
+        explicit Cache(Block && block)
+            : block(std::move(block))
+        {}
+
+        std::mutex mutex;
+        Block block;
+    };
+    using CachePtr = std::shared_ptr<Cache>;
+
 private:
     ColumnFileSchemaPtr schema;
 
@@ -83,9 +97,13 @@ class ColumnFileInMemory : public ColumnFile
         ReadTag) const override;
 
     bool isAppendable() const override { return !disable_append; }
-    void disableAppend() override { disable_append = true; }
-    bool append(const DMContext & dm_context, const Block & data, size_t offset, size_t limit, size_t data_bytes)
-        override;
+    void disableAppend() override;
+    AppendResult append(
+        const DMContext & dm_context,
+        const Block & data,
+        size_t offset,
+        size_t limit,
+        size_t data_bytes) override;
 
     Block readDataForFlush() const;
 

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

@@ -22,9 +22,32 @@
 #include <Storages/DeltaMerge/WriteBatchesImpl.h>
 #include <Storages/PathPool.h>
 
+namespace CurrentMetrics
+{
+extern const Metric DT_NumMemTable;
+extern const Metric DT_BytesMemTable;
+} // namespace CurrentMetrics
+
 namespace DB::DM
 {
 
+MemTableSet::MemTableSet(const ColumnFiles & in_memory_files)
+    : holder_counter(CurrentMetrics::DT_NumMemTable, 1)
+    , holder_allocated_bytes(CurrentMetrics::DT_BytesMemTable, 0)
+    , column_files(in_memory_files)
+    , log(Logger::get())
+{
+    column_files_count = column_files.size();
+    for (const auto & file : column_files)
+    {
+        rows += file->getRows();
+        bytes += file->getBytes();
+        allocated_bytes += file->getAllocateBytes();
+        deletes += file->getDeletes();
+    }
+    holder_allocated_bytes.changeTo(allocated_bytes.load());
+}
+
 void MemTableSet::appendColumnFileInner(const ColumnFilePtr & column_file)
 {
     if (!column_files.empty())
@@ -182,16 +205,16 @@ void MemTableSet::appendColumnFile(const ColumnFilePtr & column_file)
 void MemTableSet::appendToCache(DMContext & context, const Block & block, size_t offset, size_t limit)
 {
     // If the `column_files` is not empty, and the last `column_file` is a `ColumnInMemoryFile`, we will merge the newly block into the last `column_file`.
-    bool success = false;
+    ColumnFile::AppendResult append_res;
     size_t append_bytes = block.bytes(offset, limit);
     if (!column_files.empty())
     {
         auto & last_column_file = column_files.back();
         if (last_column_file->isAppendable())
-            success = last_column_file->append(context, block, offset, limit, append_bytes);
+            append_res = last_column_file->append(context, block, offset, limit, append_bytes);
     }
 
-    if (!success)
+    if (!append_res.success)
     {
         /// Otherwise, create a new `ColumnInMemoryFile` and write into it.
 
@@ -202,13 +225,13 @@ void MemTableSet::appendToCache(DMContext & context, const Block & block, size_t
         // Must append the empty `new_column_file` to `column_files` before appending data to it,
         // because `appendColumnFileInner` will update stats related to `column_files` but we will update stats relate to `new_column_file` here.
         appendColumnFileInner(new_column_file);
-        success = new_column_file->append(context, block, offset, limit, append_bytes);
-        if (unlikely(!success))
+        append_res = new_column_file->append(context, block, offset, limit, append_bytes);
+        if (unlikely(!append_res.success))
             throw Exception("Write to MemTableSet failed", ErrorCodes::LOGICAL_ERROR);
     }
     rows += limit;
     bytes += append_bytes;
-    
+    allocated_bytes += append_res.new_alloc_bytes;
 }
 
 void MemTableSet::appendDeleteRange(const RowKeyRange & delete_range)

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

@@ -37,6 +37,10 @@ class MemTableSet
 #else
 public:
 #endif
+    // Keep track of the number of mem-table in memory.
+    CurrentMetrics::Increment holder_counter;
+    CurrentMetrics::Increment holder_allocated_bytes;
+
     // Note that we must update `column_files_count` for outer thread-safe after `column_files` changed
     ColumnFiles column_files;
 
@@ -54,18 +58,7 @@ class MemTableSet
     void appendColumnFileInner(const ColumnFilePtr & column_file);
 
 public:
-    explicit MemTableSet(const ColumnFiles & in_memory_files = {})
-        : column_files(in_memory_files)
-        , log(Logger::get())
-    {
-        column_files_count = column_files.size();
-        for (const auto & file : column_files)
-        {
-            rows += file->getRows();
-            bytes += file->getBytes();
-            deletes += file->getDeletes();
-        }
-    }
+    explicit MemTableSet(const ColumnFiles & in_memory_files = {});
 
     /**
      * Resets the logger by using the one from the segment.

dbms/src/Storages/DeltaMerge/Segment.cpp

@@ -128,6 +128,7 @@ extern const Metric DT_SnapshotOfPlaceIndex;
 extern const Metric DT_SnapshotOfReplayVersionChain;
 extern const Metric DT_SnapshotOfSegmentIngest;
 extern const Metric DT_SnapshotOfBitmapFilter;
+extern const Metric DT_NumSegments;
 } // namespace CurrentMetrics
 
 namespace DB
@@ -295,7 +296,8 @@ Segment::Segment( //
     PageIdU64 next_segment_id_,
     const DeltaValueSpacePtr & delta_,
     const StableValueSpacePtr & stable_)
-    : epoch(epoch_)
+    : holder_counter(CurrentMetrics::NumSegments)
+    , epoch(epoch_)
     , rowkey_range(rowkey_range_)
     , is_common_handle(rowkey_range.is_common_handle)
     , rowkey_column_size(rowkey_range.rowkey_column_size)

dbms/src/Storages/DeltaMerge/Segment.h

@@ -837,6 +837,10 @@ class Segment
 #else
 public:
 #endif
+
+    // Keep track of the number of segments in memory.
+    CurrentMetrics::Increment holder_counter;
+
     /// The version of this segment. After split / merge / mergeDelta / replaceData, epoch got increased by 1.
     const UInt64 epoch;