Merge pull request #16094 from nvartolomei/nv/wal-metadata-compatibility

RFC: Add metadata section to MergeTreeWriteAheadLog

src/Storages/CMakeLists.txt

+add_subdirectory(MergeTree)
 add_subdirectory(System)
 
 if(ENABLE_TESTS)

src/Storages/MergeTree/CMakeLists.txt

+if(ENABLE_TESTS)
+    add_subdirectory(tests)
+endif()

src/Storages/MergeTree/MergeTreeWriteAheadLog.cpp

@@ -64,7 +64,7 @@ void MergeTreeWriteAheadLog::addPart(const Block & block, const String & part_na
     min_block_number = std::min(min_block_number, part_info.min_block);
     max_block_number = std::max(max_block_number, part_info.max_block);
 
-    writeIntBinary(static_cast<UInt8>(0), *out); /// version
+    writeIntBinary(WAL_VERSION, *out);
     writeIntBinary(static_cast<UInt8>(ActionType::ADD_PART), *out);
     writeStringBinary(part_name, *out);
     block_out->write(block);
@@ -80,7 +80,7 @@ void MergeTreeWriteAheadLog::dropPart(const String & part_name)
 {
     std::unique_lock lock(write_mutex);
 
-    writeIntBinary(static_cast<UInt8>(0), *out);
+    writeIntBinary(WAL_VERSION, *out);
     writeIntBinary(static_cast<UInt8>(ActionType::DROP_PART), *out);
     writeStringBinary(part_name, *out);
     out->next();
@@ -116,9 +116,13 @@ MergeTreeData::MutableDataPartsVector MergeTreeWriteAheadLog::restore(const Stor
 
         try
         {
+            ActionMetadata metadata;
+
             readIntBinary(version, *in);
-            if (version != 0)
-                throw Exception("Unknown WAL format version: " + toString(version), ErrorCodes::UNKNOWN_FORMAT_VERSION);
+            if (version > 0)
+            {
+                metadata.read(*in);
+            }
 
             readIntBinary(action_type, *in);
             readStringBinary(part_name, *in);
@@ -233,4 +237,29 @@ MergeTreeWriteAheadLog::tryParseMinMaxBlockNumber(const String & filename)
     return std::make_pair(min_block, max_block);
 }
 
+void MergeTreeWriteAheadLog::ActionMetadata::read(ReadBuffer & meta_in)
+{
+    readIntBinary(min_compatible_version, meta_in);
+    if (min_compatible_version > WAL_VERSION)
+        throw Exception("WAL metadata version " + toString(min_compatible_version)
+                        + " is not compatible with this ClickHouse version", ErrorCodes::UNKNOWN_FORMAT_VERSION);
+
+    size_t metadata_size;
+    readVarUInt(metadata_size, meta_in);
+
+    UInt32 metadata_start = meta_in.offset();
+
+    /// For the future: read metadata here.
+
+
+    /// Skip extra fields if any. If min_compatible_version is lower than WAL_VERSION it means
+    /// that the fields are not critical for the correctness.
+    meta_in.ignore(metadata_size - (meta_in.offset() - metadata_start));
+}
+
+void MergeTreeWriteAheadLog::ActionMetadata::write(WriteBuffer & meta_out) const
+{
+    writeIntBinary(min_compatible_version, meta_out);
+    writeVarUInt(static_cast<UInt32>(0), meta_out);
+}
 }

src/Storages/MergeTree/MergeTreeWriteAheadLog.h

@@ -28,6 +28,19 @@ public:
         DROP_PART = 1,
     };
 
+    struct ActionMetadata
+    {
+        /// The minimum version of WAL reader that can understand metadata written by current ClickHouse version.
+        /// This field must be increased when making backwards incompatible changes.
+        ///
+        /// The same approach can be used recursively inside metadata.
+        UInt8 min_compatible_version = 0;
+
+        void write(WriteBuffer & meta_out) const;
+        void read(ReadBuffer & meta_in);
+    };
+
+    constexpr static UInt8 WAL_VERSION = 0;
     constexpr static auto WAL_FILE_NAME = "wal";
     constexpr static auto WAL_FILE_EXTENSION = ".bin";
     constexpr static auto DEFAULT_WAL_FILE_NAME = "wal.bin";

src/Storages/MergeTree/tests/CMakeLists.txt

+add_executable (wal_action_metadata wal_action_metadata.cpp)
+target_link_libraries (wal_action_metadata PRIVATE dbms)

src/Storages/MergeTree/tests/wal_action_metadata.cpp

+#include <iostream>
+
+#include <IO/MemoryReadWriteBuffer.h>
+#include <Storages/MergeTree/MergeTreeWriteAheadLog.h>
+
+namespace DB
+{
+namespace ErrorCodes
+{
+    extern const int UNKNOWN_FORMAT_VERSION;
+}
+}
+
+int main(int, char **)
+{
+    try
+    {
+        {
+            std::cout << "test: dummy test" << std::endl;
+
+            DB::MergeTreeWriteAheadLog::ActionMetadata metadata_out;
+            DB::MemoryWriteBuffer buf{};
+
+            metadata_out.write(buf);
+            buf.finalize();
+
+            metadata_out.read(*buf.tryGetReadBuffer());
+        }
+
+        {
+            std::cout << "test: min compatibility" << std::endl;
+
+            DB::MergeTreeWriteAheadLog::ActionMetadata metadata_out;
+            metadata_out.min_compatible_version = DB::MergeTreeWriteAheadLog::WAL_VERSION + 1;
+            DB::MemoryWriteBuffer buf{};
+
+            metadata_out.write(buf);
+            buf.finalize();
+
+            try
+            {
+                metadata_out.read(*buf.tryGetReadBuffer());
+            }
+            catch (const DB::Exception & e)
+            {
+                if (e.code() != DB::ErrorCodes::UNKNOWN_FORMAT_VERSION)
+                {
+                    std::cerr << "Expected UNKNOWN_FORMAT_VERSION exception but got: "
+                        << e.what() << ", " << e.displayText() << std::endl;
+                }
+            }
+        }
+    }
+    catch (const DB::Exception & e)
+    {
+        std::cerr << e.what() << ", " << e.displayText() << std::endl;
+        return 1;
+    }
+
+    return 0;
+}