Raft cache persit improvement

cmake/find_kvproto.cmake

@@ -1,7 +1,11 @@
 # Currently kvproto should always use bundled library.
 
 if (NOT EXISTS "${ClickHouse_SOURCE_DIR}/contrib/kvproto/cpp/kvproto/errorpb.pb.h")
-   message (FATAL_ERROR "kvproto submodule in contrib/kvproto is missing.")
+	if (EXISTS "${ClickHouse_SOURCE_DIR}/contrib/kvproto/proto/errorpb.proto")
+		message (FATAL_ERROR "kvproto cpp files in contrib/kvproto is missing. Try go to contrib/kvproto, and run ./generate_cpp.sh")
+	else()
+		message (FATAL_ERROR "kvproto submodule in contrib/kvproto is missing. Try run 'git submodule update --init --recursive', and go to contrib/kvproto, and run ./generate_cpp.sh")
+	endif()
 endif ()
 
 message(STATUS "Using kvproto: ${ClickHouse_SOURCE_DIR}/contrib/kvproto/cpp")

dbms/src/Common/ErrorCodes.cpp

@@ -384,6 +384,7 @@ namespace ErrorCodes
     extern const int THEFLASH_ENCODER_ERROR = 9002;
     extern const int THEFLASH_SESSION_ERROR = 9003;
     extern const int DECIMAL_OVERFLOW_ERROR = 9004;
+    extern const int SIZE_CHECK_FAILED = 9004;
     extern const int LOCK_EXCEPTION = 10000;
 }
 

dbms/src/Common/PersistedContainer.h

@@ -10,34 +10,71 @@
 #include <Poco/Path.h>
 
 #include <Core/Types.h>
+#include <IO/HashingReadBuffer.h>
+#include <IO/HashingWriteBuffer.h>
 #include <IO/ReadBufferFromFile.h>
 #include <IO/ReadHelpers.h>
 #include <IO/WriteBufferFromFile.h>
 #include <IO/WriteHelpers.h>
 
-// TODO find a way to unify PersistedContainerSetOrVector and PersistedContainerMap, and Write & Read should be able to use lambda.
+namespace DB
+{
+
+namespace ErrorCodes
+{
+extern const int CHECKSUM_DOESNT_MATCH;
+}
+
+constexpr UInt8 PERSISTED_CONTAINER_MAGIC_WORD = 0xFF;
+constexpr size_t HASH_CODE_LENGTH = sizeof(DB::HashingWriteBuffer::uint128);
 
-template <bool is_set, typename T, template <typename E = T, typename...> class Container, class Write, class Read>
-struct PersistedContainerSetOrVector
+template <bool is_map, bool is_set, class Trait>
+struct PersistedContainer : public Trait
 {
 public:
-    PersistedContainerSetOrVector(const std::string & path_) : path(path_) {}
+    using Container = typename Trait::Container;
+    using Write = typename Trait::Write;
+    using Read = typename Trait::Read;
 
-    Container<T> & get() { return container; }
+    explicit PersistedContainer(const std::string & path_) : path(path_) {}
+
+    auto & get() { return container; }
 
     void persist()
     {
         std::string tmp_file_path = path + ".tmp." + DB::toString(Poco::Timestamp().epochMicroseconds());
-        DB::WriteBufferFromFile file_buf(tmp_file_path, DBMS_DEFAULT_BUFFER_SIZE, O_WRONLY | O_TRUNC | O_CREAT);
-        size_t size = container.size();
-        writeIntBinary(size, file_buf);
-        for (const T & t : container)
         {
-            write(t, file_buf);
-        }
-        file_buf.next();
-        file_buf.sync();
+            DB::WriteBufferFromFile file_buf(tmp_file_path, DBMS_DEFAULT_BUFFER_SIZE, O_WRONLY | O_TRUNC | O_CREAT);
+            file_buf.seek(HASH_CODE_LENGTH);
 
+            DB::HashingWriteBuffer hash_buf(file_buf);
+            size_t size = container.size();
+            writeIntBinary(size, hash_buf);
+            if constexpr (is_map)
+            {
+                for (auto && [k, v] : container)
+                {
+                    writeIntBinary(PERSISTED_CONTAINER_MAGIC_WORD, hash_buf);
+                    write(k, v, hash_buf);
+                }
+            }
+            else
+            {
+                for (const auto & t : container)
+                {
+                    writeIntBinary(PERSISTED_CONTAINER_MAGIC_WORD, hash_buf);
+                    write(t, hash_buf);
+                }
+            }
+            hash_buf.next();
+
+            auto hashcode = hash_buf.getHash();
+            file_buf.seek(0);
+            writeIntBinary(hashcode.first, file_buf);
+            writeIntBinary(hashcode.second, file_buf);
+
+            file_buf.sync();
+        }
         Poco::File(tmp_file_path).renameTo(path);
     }
 
@@ -46,100 +83,80 @@ struct PersistedContainerSetOrVector
         if (!Poco::File(path).exists())
             return;
         DB::ReadBufferFromFile file_buf(path, DBMS_DEFAULT_BUFFER_SIZE, O_RDONLY);
+
+        DB::HashingReadBuffer::uint128 expected_hashcode;
+        readIntBinary(expected_hashcode.first, file_buf);
+        readIntBinary(expected_hashcode.second, file_buf);
+
+        DB::HashingReadBuffer hash_buf(file_buf);
         size_t size;
-        readIntBinary(size, file_buf);
+        readIntBinary(size, hash_buf);
+        UInt8 word;
         for (size_t i = 0; i < size; ++i)
         {
-            if constexpr (is_set)
+            readIntBinary(word, hash_buf);
+            if (word != PERSISTED_CONTAINER_MAGIC_WORD)
+                throw DB::Exception("Magic word does not match!", DB::ErrorCodes::CHECKSUM_DOESNT_MATCH);
+
+            if constexpr (is_map)
+            {
+                const auto && [k, v] = read(hash_buf);
+                container.emplace(k, v);
+            }
+            else if constexpr (is_set)
             {
-                container.insert(std::move(read(file_buf)));
+                container.insert(std::move(read(hash_buf)));
             }
             else
             {
-                container.push_back(std::move(read(file_buf)));
+                container.push_back(std::move(read(hash_buf)));
             }
         }
+        auto hashcode = hash_buf.getHash();
+        if (hashcode != expected_hashcode)
+            throw DB::Exception("Hashcode does not match!", DB::ErrorCodes::CHECKSUM_DOESNT_MATCH);
     }
 
     void drop()
     {
         Poco::File f(path);
         if (f.exists())
             f.remove(false);
-        Container<T> tmp;
+        Container tmp;
         container.swap(tmp);
     }
 
 private:
     std::string path;
-    Container<T> container;
+    Container container;
     Write write{};
     Read read{};
 };
 
-template <typename Key,
-          typename Value,
-          template <typename CKey = Key, typename CValue = Value, typename...> class Container,
-          class Write,
-          class Read>
-struct PersistedContainerMap
+template <typename K, typename V, template <typename...> class C, typename W, typename R>
+struct MapTrait
 {
-public:
-    PersistedContainerMap(const std::string & path_) : path(path_) {}
-
-    Container<Key, Value> & get() { return container; }
-
-    void persist()
-    {
-        std::string tmp_file_path = path + ".tmp." + DB::toString(Poco::Timestamp().epochMicroseconds());
-        DB::WriteBufferFromFile file_buf(tmp_file_path, DBMS_DEFAULT_BUFFER_SIZE, O_WRONLY | O_TRUNC | O_CREAT);
-        size_t size = container.size();
-        writeIntBinary(size, file_buf);
-        for (auto && [k, v] : container)
-        {
-            write(k, v, file_buf);
-        }
-        file_buf.next();
-        file_buf.sync();
-
-        Poco::File(tmp_file_path).renameTo(path);
-    }
-
-    void restore()
-    {
-        if (!Poco::File(path).exists())
-            return;
-        DB::ReadBufferFromFile file_buf(path, DBMS_DEFAULT_BUFFER_SIZE, O_RDONLY);
-        size_t size;
-        readIntBinary(size, file_buf);
-        for (size_t i = 0; i < size; ++i)
-        {
-            const auto && [k, v] = read(file_buf);
-            container.emplace(k, v);
-        }
-    }
-
-    void drop()
-    {
-        Poco::File f(path);
-        if (f.exists())
-            f.remove(false);
-        Container<Key, Value> tmp;
-        container.swap(tmp);
-    }
+    using Container = C<K, V>;
+    using Write = W;
+    using Read = R;
+};
 
-private:
-    std::string path;
-    Container<Key, Value> container;
-    Write write{};
-    Read read{};
+template <typename T, template <typename...> class C, typename W, typename R>
+struct VecSetTrait
+{
+    using Container = C<T>;
+    using Write = W;
+    using Read = R;
 };
 
-template <typename T, template <typename E = T, typename...> class Container, class Write, class Read>
-using PersistedContainerSet = PersistedContainerSetOrVector<true, T, Container, Write, Read>;
+template <typename K, typename V, template <typename...> class C, class Write, class Read>
+using PersistedContainerMap = PersistedContainer<true, false, MapTrait<K, V, C, Write, Read>>;
+
+template <typename T, template <typename...> class C, class Write, class Read>
+using PersistedContainerSet = PersistedContainer<false, true, VecSetTrait<T, C, Write, Read>>;
 
-template <typename T, template <typename E = T, typename...> class Container, class Write, class Read>
-using PersistedContainerVector = PersistedContainerSetOrVector<false, T, Container, Write, Read>;
+template <typename T, template <typename...> class C, class Write, class Read>
+using PersistedContainerVector = PersistedContainer<false, false, VecSetTrait<T, C, Write, Read>>;
 
 struct UInt64Write
 {
@@ -177,3 +194,5 @@ struct UInt64StringRead
 };
 using PersistedUnorderedUInt64ToStringMap
     = PersistedContainerMap<UInt64, std::string, std::unordered_map, UInt64StringWrite, UInt64StringRead>;
+
+} // namespace DB

dbms/src/Common/tests/persisted_container.cpp

@@ -5,15 +5,21 @@
 
 #include <Common/PersistedContainer.h>
 
+using namespace DB;
+
 int main(int, char **)
 {
+    auto clear_file = [=](std::string path) {
+        Poco::File file(path);
+        if (file.exists())
+            file.remove();
+    };
+
     {
         std::string file_path = "persisted_container_set_test.dat";
-        SCOPE_EXIT({
-            Poco::File file(file_path);
-            if (file.exists())
-                file.remove();
-        });
+        clear_file(file_path);
+        SCOPE_EXIT({ clear_file(file_path); });
+
         {
             PersistedUnorderedUInt64Set set(file_path);
             set.restore();
@@ -38,11 +44,9 @@ int main(int, char **)
 
     {
         std::string file_path = "persisted_container_map_test.dat";
-        SCOPE_EXIT({
-            Poco::File file(file_path);
-            if (file.exists())
-                file.remove();
-        });
+        clear_file(file_path);
+        SCOPE_EXIT({ clear_file(file_path); });
+
         {
             PersistedUnorderedUInt64ToStringMap map(file_path);
             map.restore();

dbms/src/DataStreams/DeletingDeletedBlockInputStream.cpp

@@ -11,7 +11,7 @@ Block DeletingDeletedBlockInputStream::readImpl()
     if (!block)
         return block;
     IColumn::Filter filter(block.rows());
-    size_t count = setFilterByDeleteMarkColumn(block, filter);
+    size_t count = setFilterByDelMarkColumn(block, filter);
     if (count > 0)
         deleteRows(block, filter);
     return block;

dbms/src/DataStreams/MvccTMTSortedBlockInputStream.cpp

@@ -1,4 +1,5 @@
 #include <DataStreams/MvccTMTSortedBlockInputStream.h>
+#include <Storages/MutableSupport.h>
 
 namespace DB
 {
@@ -42,7 +43,7 @@ void MvccTMTSortedBlockInputStream::merge(MutableColumns & merged_columns, std::
         setPrimaryKeyRef(next_key, current);
 
         bool key_differs = next_key != current_key;
-        
+
         if (key_differs && merged_rows >= max_block_size)
             return;
 
@@ -56,7 +57,7 @@ void MvccTMTSortedBlockInputStream::merge(MutableColumns & merged_columns, std::
             selected_row.reset();
 
             current_key.swap(next_key);
-        } 
+        }
 
         if ((*(current->all_columns[version_column_number]))[current->pos].template get<UInt64>() <= read_tso && 
             (selected_row.empty() 
@@ -87,7 +88,8 @@ void MvccTMTSortedBlockInputStream::merge(MutableColumns & merged_columns, std::
 
 bool MvccTMTSortedBlockInputStream::hasDeleteFlag()
 {
-    return (*(*selected_row.columns)[del_column_number])[selected_row.row_num].template get<UInt8>() > 0;
+    UInt8 val = (*(*selected_row.columns)[del_column_number])[selected_row.row_num].template get<UInt8>();
+    return MutableSupport::DelMark::isDel(val);
 }
 
 }

dbms/src/DataStreams/MvccTMTSortedBlockInputStream.h

@@ -9,9 +9,9 @@ namespace DB
 
 class MvccTMTSortedBlockInputStream : public MergingSortedBlockInputStream
 {
-public :
+public:
     MvccTMTSortedBlockInputStream(
-        const BlockInputStreams & inputs_, 
+        const BlockInputStreams & inputs_,
         const SortDescription & description_,
         const String & version_column,
         const String & del_column,
@@ -23,31 +23,30 @@ public :
         version_column_number = header.getPositionByName(version_column);
         del_column_number = header.getPositionByName(del_column);
     }
-    
+
     String getName() const override { return "MvccTMTSorted"; }
 
 protected:
     Block readImpl() override;
 
 private:
-    ssize_t version_column_number ;
-    ssize_t del_column_number ;
+    ssize_t version_column_number;
+    ssize_t del_column_number;
     Logger * log = &Logger::get("MvccTMTStortedBlockInputStream");
 
     bool finished = false;
 
     RowRef current_key;
     RowRef next_key;
     RowRef selected_row;
-    
+
     size_t read_tso;
 
     void merge(MutableColumns & merged_columns, std::priority_queue<SortCursor> & queue);
 
     bool hasDeleteFlag();
 
     void insertRow(MutableColumns &, size_t &);
-
 };
 
 }