Storage: let stable meta using protobuf format

dbms/src/Storages/DeltaMerge/CMakeLists.txt

@@ -17,6 +17,7 @@ add_subdirectory(File/dtpb)
 include(${TiFlash_SOURCE_DIR}/cmake/dbms_glob_sources.cmake)
 
 add_subdirectory(./Remote/Proto)
+add_subdirectory(./Proto)
 
 add_headers_and_sources(delta_merge .)
 add_headers_and_sources(delta_merge ./BitmapFilter)
@@ -33,7 +34,7 @@ add_headers_and_sources(delta_merge ./Decode)
 add_headers_and_sources(delta_merge ./StoragePool)
 
 add_library(delta_merge ${delta_merge_headers} ${delta_merge_sources})
-target_link_libraries(delta_merge PRIVATE dbms page)
+target_link_libraries(delta_merge PRIVATE dbms page DeltaMergeProto)
 
 add_subdirectory(workload)
 

dbms/src/Storages/DeltaMerge/Proto/CMakeLists.txt

@@ -0,0 +1,20 @@
+# Copyright 2024 PingCAP, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+file(GLOB PROTO_FILES CONFIGURE_DEPENDS *.proto)
+protobuf_generate_cpp(rtproto_srcs rtproto_hdrs ${PROTO_FILES})
+
+add_library(DeltaMergeProto ${rtproto_srcs})
+target_include_directories(DeltaMergeProto PUBLIC ${Protobuf_INCLUDE_DIR})
+target_compile_options(DeltaMergeProto PRIVATE -Wno-unused-parameter)

dbms/src/Storages/DeltaMerge/Proto/stable.proto

@@ -0,0 +1,28 @@
+// Copyright 2024 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+syntax = "proto3";
+
+package DB.DM.PB;
+
+message FileIdentifier {
+    uint64 file_id = 1;
+}
+
+message StableLayerMeta {
+    uint64 valid_rows = 1;
+    uint64 valid_bytes = 2;
+    uint64 num_files = 3;
+    repeated FileIdentifier file_ids = 4;
+}

dbms/src/Storages/DeltaMerge/StableValueSpace.cpp

@@ -29,6 +29,7 @@
 #include <Storages/Page/V3/Universal/UniversalPageStorage.h>
 #include <Storages/PathPool.h>
 
+
 namespace DB
 {
 namespace ErrorCodes
@@ -82,31 +83,93 @@ void StableValueSpace::saveMeta(WriteBatchWrapper & meta_wb)
 {
     MemoryWriteBuffer buf(0, 8192);
     // The method must call `buf.count()` to get the last seralized size before `buf.tryGetReadBuffer`
-    auto data_size = saveMeta(buf);
+    auto data_size = serializeMetaToBuf(buf);
     meta_wb.putPage(id, 0, buf.tryGetReadBuffer(), data_size);
 }
 
-UInt64 StableValueSpace::saveMeta(WriteBuffer & buf) const
+UInt64 StableValueSpace::serializeMetaToBuf(WriteBuffer & buf) const
 {
     writeIntBinary(STORAGE_FORMAT_CURRENT.stable, buf);
-    writeIntBinary(valid_rows, buf);
-    writeIntBinary(valid_bytes, buf);
-    writeIntBinary(static_cast<UInt64>(files.size()), buf);
-    for (const auto & f : files)
-        writeIntBinary(f->pageId(), buf);
-
+    if (STORAGE_FORMAT_CURRENT.stable == StableFormat::V1)
+    {
+        writeIntBinary(valid_rows, buf);
+        writeIntBinary(valid_bytes, buf);
+        writeIntBinary(static_cast<UInt64>(files.size()), buf);
+        for (const auto & f : files)
+            writeIntBinary(f->pageId(), buf);
+    }
+    else if (STORAGE_FORMAT_CURRENT.stable == StableFormat::V2)
+    {
+        PB::StableLayerMeta meta;
+        meta.set_valid_rows(valid_rows);
+        meta.set_valid_bytes(valid_bytes);
+        meta.set_num_files(files.size());
+        for (const auto & f : files)
+            meta.add_file_ids()->set_file_id(f->pageId());
+
+        auto data = meta.SerializeAsString();
+        writeStringBinary(data, buf);
+    }
+    else
+    {
+        throw Exception("Unexpected version: {}", STORAGE_FORMAT_CURRENT.stable);
+    }
     return buf.count();
 }
 
+namespace
+{
+PB::StableLayerMeta derializeMetaV1FromBuf(ReadBuffer & buf)
+{
+    PB::StableLayerMeta meta;
+    UInt64 valid_rows, valid_bytes, size;
+    readIntBinary(valid_rows, buf);
+    readIntBinary(valid_bytes, buf);
+    readIntBinary(size, buf);
+    meta.set_valid_rows(valid_rows);
+    meta.set_valid_bytes(valid_bytes);
+    meta.set_num_files(size);
+    for (size_t i = 0; i < size; ++i)
+    {
+        UInt64 page_id;
+        readIntBinary(page_id, buf);
+        meta.add_file_ids()->set_file_id(page_id);
+    }
+    return meta;
+}
+
+PB::StableLayerMeta derializeMetaV2FromBuf(ReadBuffer & buf)
+{
+    PB::StableLayerMeta meta;
+    String data;
+    readStringBinary(data, buf);
+    RUNTIME_CHECK_MSG(meta.ParseFromString(data), "Failed to parse StableLayerMeta from string: {}", data);
+    return meta;
+}
+
+PB::StableLayerMeta derializeMetaFromBuf(ReadBuffer & buf)
+{
+    UInt64 version;
+    readIntBinary(version, buf);
+    if (version == StableFormat::V1)
+        return derializeMetaV1FromBuf(buf);
+    else if (version == StableFormat::V2)
+        return derializeMetaV2FromBuf(buf);
+    else
+        throw Exception("Unexpected version: {}", version);
+}
+} // namespace
+
 std::string StableValueSpace::serializeMeta() const
 {
     WriteBufferFromOwnString wb;
-    saveMeta(wb);
+    serializeMetaToBuf(wb);
     return wb.releaseStr();
 }
 
 StableValueSpacePtr StableValueSpace::restore(DMContext & dm_context, PageIdU64 id)
 {
+    // read meta page
     Page page = dm_context.storage_pool->metaReader()->read(id); // not limit restore
     ReadBufferFromMemory buf(page.data.begin(), page.data.size());
     return StableValueSpace::restore(dm_context, buf, id);
@@ -116,20 +179,11 @@ StableValueSpacePtr StableValueSpace::restore(DMContext & dm_context, ReadBuffer
 {
     auto stable = std::make_shared<StableValueSpace>(id);
 
-    UInt64 version, valid_rows, valid_bytes, size;
-    readIntBinary(version, buf);
-    if (version != StableFormat::V1)
-        throw Exception("Unexpected version: " + DB::toString(version));
-
-    readIntBinary(valid_rows, buf);
-    readIntBinary(valid_bytes, buf);
-    readIntBinary(size, buf);
-    UInt64 page_id;
+    auto metapb = derializeMetaFromBuf(buf);
     auto remote_data_store = dm_context.global_context.getSharedContextDisagg()->remote_data_store;
-    for (size_t i = 0; i < size; ++i)
+    for (size_t i = 0; i < metapb.num_files(); ++i)
     {
-        readIntBinary(page_id, buf);
-
+        UInt64 page_id = metapb.file_ids(i).file_id();
         DMFilePtr dmfile;
         auto path_delegate = dm_context.path_pool->getStableDiskDelegator();
         if (remote_data_store)
@@ -170,8 +224,8 @@ StableValueSpacePtr StableValueSpace::restore(DMContext & dm_context, ReadBuffer
         stable->files.push_back(dmfile);
     }
 
-    stable->valid_rows = valid_rows;
-    stable->valid_bytes = valid_bytes;
+    stable->valid_rows = metapb.valid_rows();
+    stable->valid_bytes = metapb.valid_bytes();
 
     return stable;
 }
@@ -192,22 +246,11 @@ StableValueSpacePtr StableValueSpace::createFromCheckpoint( //
     ReadBufferFromMemory buf(page.data.begin(), page.data.size());
 
     // read stable meta info
-    UInt64 version, valid_rows, valid_bytes, size;
-    {
-        readIntBinary(version, buf);
-        if (version != StableFormat::V1)
-            throw Exception("Unexpected version: " + DB::toString(version));
-
-        readIntBinary(valid_rows, buf);
-        readIntBinary(valid_bytes, buf);
-        readIntBinary(size, buf);
-    }
-
+    auto metapb = derializeMetaFromBuf(buf);
     auto remote_data_store = dm_context.global_context.getSharedContextDisagg()->remote_data_store;
-    for (size_t i = 0; i < size; ++i)
+    for (size_t i = 0; i < metapb.num_files(); ++i)
     {
-        UInt64 page_id;
-        readIntBinary(page_id, buf);
+        UInt64 page_id = metapb.file_ids(i).file_id();
         auto full_page_id = UniversalPageIdFormat::toFullPageId(
             UniversalPageIdFormat::toFullPrefix(
                 dm_context.keyspace_id,
@@ -234,8 +277,8 @@ StableValueSpacePtr StableValueSpace::createFromCheckpoint( //
         stable->files.push_back(dmfile);
     }
 
-    stable->valid_rows = valid_rows;
-    stable->valid_bytes = valid_bytes;
+    stable->valid_rows = metapb.valid_rows();
+    stable->valid_bytes = metapb.valid_bytes();
 
     return stable;
 }

dbms/src/Storages/DeltaMerge/StableValueSpace.h

@@ -19,6 +19,8 @@
 #include <Storages/DeltaMerge/File/DMFilePackFilter_fwd.h>
 #include <Storages/DeltaMerge/File/DMFile_fwd.h>
 #include <Storages/DeltaMerge/Index/RSResult.h>
+#include <Storages/DeltaMerge/Proto/stable.pb.h>
+#include <Storages/DeltaMerge/ReadMode.h>
 #include <Storages/DeltaMerge/RowKeyRange.h>
 #include <Storages/DeltaMerge/SkippableBlockInputStream.h>
 #include <Storages/Page/PageStorage_fwd.h>
@@ -39,7 +41,7 @@ using StableValueSpacePtr = std::shared_ptr<StableValueSpace>;
 class StableValueSpace : public std::enable_shared_from_this<StableValueSpace>
 {
 public:
-    StableValueSpace(PageIdU64 id_)
+    explicit StableValueSpace(PageIdU64 id_)
         : id(id_)
         , log(Logger::get())
     {}
@@ -127,7 +129,7 @@ class StableValueSpace : public std::enable_shared_from_this<StableValueSpace>
         // number of rows having at least one version(include delete)
         UInt64 num_rows;
 
-        const String toDebugString() const
+        String toDebugString() const
         {
             return "StableProperty: gc_hint_version [" + std::to_string(this->gc_hint_version) + "] num_versions ["
                 + std::to_string(this->num_versions) + "] num_puts[" + std::to_string(this->num_puts) + "] num_rows["
@@ -148,18 +150,18 @@ class StableValueSpace : public std::enable_shared_from_this<StableValueSpace>
     {
         StableValueSpacePtr stable;
 
-        PageIdU64 id;
-        UInt64 valid_rows;
-        UInt64 valid_bytes;
+        PageIdU64 id{};
+        UInt64 valid_rows{};
+        UInt64 valid_bytes{};
 
-        bool is_common_handle;
-        size_t rowkey_column_size;
+        bool is_common_handle{};
+        size_t rowkey_column_size{};
 
         /// TODO: The members below are not actually snapshots, they should not be here.
 
         ColumnCachePtrs column_caches;
 
-        Snapshot(StableValueSpacePtr stable_)
+        explicit Snapshot(StableValueSpacePtr stable_)
             : stable(stable_)
             , log(stable->log)
         {}
@@ -263,19 +265,19 @@ class StableValueSpace : public std::enable_shared_from_this<StableValueSpace>
     size_t avgRowBytes(const ColumnDefines & read_columns);
 
 private:
-    UInt64 saveMeta(WriteBuffer & buf) const;
+    UInt64 serializeMetaToBuf(WriteBuffer & buf) const;
 
 private:
     const PageIdU64 id;
 
     // Valid rows is not always the sum of rows in file,
     // because after logical split, two segments could reference to a same file.
-    UInt64 valid_rows; /* At most. The actual valid rows may be lower than this value. */
-    UInt64 valid_bytes; /* At most. The actual valid bytes may be lower than this value. */
+    UInt64 valid_rows{}; /* At most. The actual valid rows may be lower than this value. */
+    UInt64 valid_bytes{}; /* At most. The actual valid bytes may be lower than this value. */
 
     DMFiles files;
 
-    StableProperty property;
+    StableProperty property{};
     std::atomic<bool> is_property_cached = false;
 
     LoggerPtr log;

dbms/src/Storages/DeltaMerge/tests/gtest_segment.cpp

@@ -232,6 +232,22 @@ try
 }
 CATCH
 
+TEST_F(SegmentOperationTest, CurrentV2RestoreFromStableV1)
+try
+{
+    auto current = STORAGE_FORMAT_CURRENT;
+    STORAGE_FORMAT_CURRENT = STORAGE_FORMAT_V5;
+    writeSegment(DELTA_MERGE_FIRST_SEGMENT_ID, 100);
+    flushSegmentCache(DELTA_MERGE_FIRST_SEGMENT_ID);
+    mergeSegmentDelta(DELTA_MERGE_FIRST_SEGMENT_ID);
+
+    STORAGE_FORMAT_CURRENT = STORAGE_FORMAT_V6;
+    auto segment = Segment::restoreSegment(log, *dm_context, DELTA_MERGE_FIRST_SEGMENT_ID);
+    ASSERT_EQ(segment->stable->getRows(), 100);
+    STORAGE_FORMAT_CURRENT = current;
+}
+CATCH
+
 TEST_F(SegmentOperationTest, WriteDuringSegmentSplit)
 try
 {

dbms/src/Storages/FormatVersion.h

@@ -47,6 +47,7 @@ namespace StableFormat
 using Version = Int64;
 
 inline static constexpr Version V1 = 1;
+inline static constexpr Version V2 = 2; // Meta using protobuf
 } // namespace StableFormat
 
 namespace DeltaFormat
@@ -130,6 +131,15 @@ inline static const StorageFormatVersion STORAGE_FORMAT_V5 = StorageFormatVersio
     .identifier = 5,
 };
 
+inline static const StorageFormatVersion STORAGE_FORMAT_V6 = StorageFormatVersion{
+    .segment = SegmentFormat::V2,
+    .dm_file = DMFileFormat::V3,
+    .stable = StableFormat::V2, // diff
+    .delta = DeltaFormat::V3,
+    .page = PageFormat::V3,
+    .identifier = 6,
+};
+
 // STORAGE_FORMAT_V100 is used for S3 only
 inline static const StorageFormatVersion STORAGE_FORMAT_V100 = StorageFormatVersion{
     .segment = SegmentFormat::V2,
@@ -156,6 +166,8 @@ inline const StorageFormatVersion & toStorageFormat(UInt64 setting)
         return STORAGE_FORMAT_V4;
     case 5:
         return STORAGE_FORMAT_V5;
+    case 6:
+        return STORAGE_FORMAT_V6;
     case 100:
         return STORAGE_FORMAT_V100;
     default: