Some compileable code

src/Common/ZooKeeper/ZooKeeperIO.cpp

@@ -4,6 +4,11 @@ namespace Coordination
 {
 
 
+void write(uint8_t x, WriteBuffer & out)
+{
+    writeBinary(x, out);
+}
+
 void write(size_t x, WriteBuffer & out)
 {
     x = __builtin_bswap64(x);
@@ -64,6 +69,11 @@ void write(const Error & x, WriteBuffer & out)
     write(static_cast<int32_t>(x), out);
 }
 
+void read(uint8_t & x, ReadBuffer & in)
+{
+    readBinary(x, in);
+}
+
 void read(size_t & x, ReadBuffer & in)
 {
     readBinary(x, in);

src/Common/ZooKeeper/ZooKeeperIO.h

@@ -13,9 +13,11 @@ namespace Coordination
 
 using namespace DB;
 
+void write(uint8_t x, WriteBuffer & out);
 void write(size_t x, WriteBuffer & out);
 void write(int64_t x, WriteBuffer & out);
 void write(int32_t x, WriteBuffer & out);
+
 void write(OpNum x, WriteBuffer & out);
 void write(bool x, WriteBuffer & out);
 void write(const std::string & s, WriteBuffer & out);
@@ -38,6 +40,7 @@ void write(const std::vector<T> & arr, WriteBuffer & out)
         write(elem, out);
 }
 
+void read(uint8_t & x, ReadBuffer & in);
 void read(size_t & x, ReadBuffer & in);
 void read(int64_t & x, ReadBuffer & in);
 void read(int32_t & x, ReadBuffer & in);

src/Coordination/NuKeeperServer.cpp

@@ -25,7 +25,7 @@ NuKeeperServer::NuKeeperServer(
     ResponsesQueue & responses_queue_)
     : server_id(server_id_)
     , coordination_settings(coordination_settings_)
-    , state_machine(nuraft::cs_new<NuKeeperStateMachine>(responses_queue_, coordination_settings))
+    , state_machine(nuraft::cs_new<NuKeeperStateMachine>(responses_queue_, config.getString("test_keeper_server.snapshot_storage_path"), coordination_settings))
     , state_manager(nuraft::cs_new<NuKeeperStateManager>(server_id, "test_keeper_server", config, coordination_settings))
     , responses_queue(responses_queue_)
 {

src/Coordination/NuKeeperStateMachine.cpp

@@ -29,14 +29,21 @@ NuKeeperStorage::RequestForSession parseRequest(nuraft::buffer & data)
     return request_for_session;
 }
 
-NuKeeperStateMachine::NuKeeperStateMachine(ResponsesQueue & responses_queue_, const CoordinationSettingsPtr & coordination_settings_)
+NuKeeperStateMachine::NuKeeperStateMachine(ResponsesQueue & responses_queue_, const std::string & snapshots_path_, const CoordinationSettingsPtr & coordination_settings_)
     : coordination_settings(coordination_settings_)
     , storage(coordination_settings->dead_session_check_period_ms.totalMilliseconds())
+    , snapshot_manager(snapshots_path_)
     , responses_queue(responses_queue_)
     , last_committed_idx(0)
     , log(&Poco::Logger::get("NuKeeperStateMachine"))
 {
-    LOG_DEBUG(log, "Created nukeeper state machine");
+}
+
+void NuKeeperStateMachine::init()
+{
+    LOG_DEBUG(log, "Trying to load state machine");
+    last_committed_idx = snapshot_manager.restoreFromLatestSnapshot(&storage);
+    LOG_DEBUG(log, "Loaded snapshot with last commited log index {}", last_committed_idx);
 }
 
 nuraft::ptr<nuraft::buffer> NuKeeperStateMachine::commit(const size_t log_idx, nuraft::buffer & data)
@@ -76,16 +83,12 @@ nuraft::ptr<nuraft::buffer> NuKeeperStateMachine::commit(const size_t log_idx, n
 bool NuKeeperStateMachine::apply_snapshot(nuraft::snapshot & s)
 {
     LOG_DEBUG(log, "Applying snapshot {}", s.get_last_log_idx());
-    StorageSnapshotPtr snapshot;
+    if (s.get_last_log_idx() != latest_snapshot_meta->get_last_log_idx())
-    {
+        throw Exception(ErrorCodes::LOGICAL_ERROR, "Required to apply snapshot with last log index {}, but our last log index is {}",
-        std::lock_guard<std::mutex> lock(snapshots_lock);
+                        s.get_last_log_idx(), latest_snapshot_meta->get_last_log_idx());
-        auto entry = snapshots.find(s.get_last_log_idx());
+
-        if (entry == snapshots.end())
-            return false;
-        snapshot = entry->second;
-    }
     std::lock_guard lock(storage_lock);
-    storage = snapshot->storage;
+    snapshot_manager.deserializeSnapshotFromBuffer(&storage, latest_snapshot_buf);
     last_committed_idx = s.get_last_log_idx();
     return true;
 }
@@ -94,41 +97,7 @@ nuraft::ptr<nuraft::snapshot> NuKeeperStateMachine::last_snapshot()
 {
    // Just return the latest snapshot.
     std::lock_guard<std::mutex> lock(snapshots_lock);
-    auto entry = snapshots.rbegin();
+    return latest_snapshot_meta;
-    if (entry == snapshots.rend())
-        return nullptr;
-
-    return entry->second->snapshot;
-}
-
-NuKeeperStateMachine::StorageSnapshotPtr NuKeeperStateMachine::createSnapshotInternal(nuraft::snapshot & s)
-{
-    nuraft::ptr<nuraft::buffer> snp_buf = s.serialize();
-    nuraft::ptr<nuraft::snapshot> ss = nuraft::snapshot::deserialize(*snp_buf);
-    std::lock_guard lock(storage_lock);
-    return std::make_shared<NuKeeperStateMachine::StorageSnapshot>(ss, storage);
-}
-
-NuKeeperStateMachine::StorageSnapshotPtr NuKeeperStateMachine::readSnapshot(nuraft::snapshot & s, nuraft::buffer & in)
-{
-    nuraft::ptr<nuraft::buffer> snp_buf = s.serialize();
-    nuraft::ptr<nuraft::snapshot> ss = nuraft::snapshot::deserialize(*snp_buf);
-    NuKeeperStorageSerializer serializer;
-
-    ReadBufferFromNuraftBuffer reader(in);
-    NuKeeperStorage new_storage(coordination_settings->dead_session_check_period_ms.totalMilliseconds());
-    serializer.deserialize(new_storage, reader);
-    return std::make_shared<StorageSnapshot>(ss, new_storage);
-}
-
-
-void NuKeeperStateMachine::writeSnapshot(const NuKeeperStateMachine::StorageSnapshotPtr & snapshot, nuraft::ptr<nuraft::buffer> & out)
-{
-    NuKeeperStorageSerializer serializer;
-
-    WriteBufferFromNuraftBuffer writer;
-    serializer.serialize(snapshot->storage, writer);
-    out = writer.getBuffer();
 }
 
 void NuKeeperStateMachine::create_snapshot(
@@ -136,26 +105,12 @@ void NuKeeperStateMachine::create_snapshot(
     nuraft::async_result<bool>::handler_type & when_done)
 {
     LOG_DEBUG(log, "Creating snapshot {}", s.get_last_log_idx());
-    auto snapshot = createSnapshotInternal(s);
+    std::lock_guard lock(storage_lock);
-    {
+    NuKeeperStorageSnapshot snapshot(&storage, s.get_last_log_idx());
-        std::lock_guard<std::mutex> lock(snapshots_lock);
+    latest_snapshot_buf = snapshot_manager.serializeSnapshotToBuffer(snapshot);
-        snapshots[s.get_last_log_idx()] = snapshot;
+    auto result_path = snapshot_manager.serializeSnapshotBufferToDisk(*latest_snapshot_buf, s.get_last_log_idx());
-        size_t num = snapshots.size();
+    LOG_DEBUG(log, "Created snapshot {} with path {}", s.get_last_log_idx(), result_path);
-        if (num > coordination_settings->max_stored_snapshots)
-        {
-            auto entry = snapshots.begin();
 
-            for (size_t i = 0; i < num - coordination_settings->max_stored_snapshots; ++i)
-            {
-                if (entry == snapshots.end())
-                    break;
-                entry = snapshots.erase(entry);
-            }
-        }
-
-    }
-
-    LOG_DEBUG(log, "Created snapshot {}", s.get_last_log_idx());
     nuraft::ptr<std::exception> except(nullptr);
     bool ret = true;
     when_done(ret, except);
@@ -170,20 +125,24 @@ void NuKeeperStateMachine::save_logical_snp_obj(
 {
     LOG_DEBUG(log, "Saving snapshot {} obj_id {}", s.get_last_log_idx(), obj_id);
 
+    // Object ID == 0: it contains dummy value, create snapshot context.
     if (obj_id == 0)
     {
-        auto new_snapshot = createSnapshotInternal(s);
+        std::lock_guard lock(storage_lock);
-        std::lock_guard<std::mutex> lock(snapshots_lock);
+        NuKeeperStorageSnapshot snapshot(&storage, s.get_last_log_idx());
-        snapshots.try_emplace(s.get_last_log_idx(), std::move(new_snapshot));
+        latest_snapshot_buf = snapshot_manager.serializeSnapshotToBuffer(snapshot);
     }
     else
     {
-        auto received_snapshot = readSnapshot(s, data);
+        latest_snapshot_buf = nuraft::buffer::clone(data);
-
-        std::lock_guard<std::mutex> lock(snapshots_lock);
-        snapshots[s.get_last_log_idx()] = std::move(received_snapshot);
     }
 
+    nuraft::ptr<nuraft::buffer> snp_buf = s.serialize();
+    latest_snapshot_meta = nuraft::snapshot::deserialize(*snp_buf);
+
+    auto result_path = snapshot_manager.serializeSnapshotBufferToDisk(*latest_snapshot_buf, s.get_last_log_idx());
+    LOG_DEBUG(log, "Created snapshot {} with path {}", s.get_last_log_idx(), result_path);
+
     obj_id++;
 }
 
@@ -196,29 +155,19 @@ int NuKeeperStateMachine::read_logical_snp_obj(
 {
 
     LOG_DEBUG(log, "Reading snapshot {} obj_id {}", s.get_last_log_idx(), obj_id);
-    StorageSnapshotPtr required_snapshot;
-    {
-        std::lock_guard<std::mutex> lock(snapshots_lock);
-        auto entry = snapshots.find(s.get_last_log_idx());
-        if (entry == snapshots.end())
-        {
-            // Snapshot doesn't exist.
-            data_out = nullptr;
-            is_last_obj = true;
-            return 0;
-        }
-        required_snapshot = entry->second;
-    }
-
     if (obj_id == 0)
     {
-        auto new_snapshot = createSnapshotInternal(s);
+        data_out = nuraft::buffer::alloc(sizeof(int32_t));
-        writeSnapshot(new_snapshot, data_out);
+        nuraft::buffer_serializer bs(data_out);
+        bs.put_i32(0);
         is_last_obj = false;
     }
     else
     {
-        writeSnapshot(required_snapshot, data_out);
+        if (s.get_last_log_idx() != latest_snapshot_meta->get_last_log_idx())
+            throw Exception(ErrorCodes::LOGICAL_ERROR, "Required to apply snapshot with last log index {}, but our last log index is {}",
+                            s.get_last_log_idx(), latest_snapshot_meta->get_last_log_idx());
+        data_out = nuraft::buffer::clone(*latest_snapshot_buf);
         is_last_obj = true;
     }
     return 0;

src/Coordination/NuKeeperStateMachine.h

@@ -5,6 +5,7 @@
 #include <common/logger_useful.h>
 #include <Coordination/ThreadSafeQueue.h>
 #include <Coordination/CoordinationSettings.h>
+#include <Coordination/NuKeeperStorageSerializer.h>
 
 namespace DB
 {
@@ -14,7 +15,9 @@ using ResponsesQueue = ThreadSafeQueue<NuKeeperStorage::ResponseForSession>;
 class NuKeeperStateMachine : public nuraft::state_machine
 {
 public:
-    NuKeeperStateMachine(ResponsesQueue & responses_queue_, const CoordinationSettingsPtr & coordination_settings_);
+    NuKeeperStateMachine(ResponsesQueue & responses_queue_, const std::string & snapshots_path_, const CoordinationSettingsPtr & coordination_settings_);
+
+    void init();
 
     nuraft::ptr<nuraft::buffer> pre_commit(const size_t /*log_idx*/, nuraft::buffer & /*data*/) override { return nullptr; }
 
@@ -58,29 +61,16 @@ public:
     void shutdownStorage();
 
 private:
-    struct StorageSnapshot
-    {
-        StorageSnapshot(const nuraft::ptr<nuraft::snapshot> & s, const NuKeeperStorage & storage_)
-            : snapshot(s)
-            , storage(storage_)
-        {}
 
-        nuraft::ptr<nuraft::snapshot> snapshot;
+    nuraft::ptr<nuraft::snapshot> latest_snapshot_meta;
-        NuKeeperStorage storage;
+    nuraft::ptr<nuraft::buffer> latest_snapshot_buf;
-    };
-
-    using StorageSnapshotPtr = std::shared_ptr<StorageSnapshot>;
-
-    StorageSnapshotPtr createSnapshotInternal(nuraft::snapshot & s);
-
-    StorageSnapshotPtr readSnapshot(nuraft::snapshot & s, nuraft::buffer & in);
-
-    static void writeSnapshot(const StorageSnapshotPtr & snapshot, nuraft::ptr<nuraft::buffer> & out);
 
     CoordinationSettingsPtr coordination_settings;
 
     NuKeeperStorage storage;
 
+    NuKeeperSnapshotManager snapshot_manager;
+
     ResponsesQueue & responses_queue;
     /// Mutex for snapshots
     std::mutex snapshots_lock;
@@ -88,9 +78,6 @@ private:
     /// Lock for storage
     std::mutex storage_lock;
 
-    /// Fake snapshot storage
-    std::map<uint64_t, StorageSnapshotPtr> snapshots;
-
     /// Last committed Raft log number.
     std::atomic<size_t> last_committed_idx;
     Poco::Logger * log;

src/Coordination/NuKeeperStorage.h

@@ -18,6 +18,9 @@ struct NuKeeperStorageRequest;
 using NuKeeperStorageRequestPtr = std::shared_ptr<NuKeeperStorageRequest>;
 using ResponseCallback = std::function<void(const Coordination::ZooKeeperResponsePtr &)>;
 using ChildrenSet = std::unordered_set<std::string>;
+using SessionAndTimeout = std::unordered_map<int64_t, int64_t>;
+
+struct NuKeeperStorageSnapshot;
 
 class NuKeeperStorage
 {
@@ -54,7 +57,6 @@ public:
     using Container = SnapshotableHashTable<Node>;
     using Ephemerals = std::unordered_map<int64_t, std::unordered_set<std::string>>;
     using SessionAndWatcher = std::unordered_map<int64_t, std::unordered_set<std::string>>;
-    using SessionAndTimeout = std::unordered_map<int64_t, int64_t>;
     using SessionIDs = std::vector<int64_t>;
 
     using Watches = std::map<String /* path, relative of root_path */, SessionIDs>;
@@ -89,6 +91,12 @@ public:
         return result;
     }
 
+    void addSessionID(int64_t session_id, int64_t session_timeout_ms)
+    {
+        session_and_timeout.emplace(session_id, session_timeout_ms);
+        session_expiry_queue.update(session_id, session_timeout_ms);
+    }
+
     ResponsesForSessions processRequest(const Coordination::ZooKeeperRequestPtr & request, int64_t session_id, std::optional<int64_t> new_last_zxid);
 
     void finalize();

src/Coordination/NuKeeperStorageSerializer.cpp

@@ -1,13 +1,48 @@
 #include <Coordination/NuKeeperStorageSerializer.h>
 #include <IO/WriteHelpers.h>
+#include <Compression/CompressedReadBuffer.h>
+#include <Compression/CompressedWriteBuffer.h>
 #include <IO/ReadHelpers.h>
 #include <Common/ZooKeeper/ZooKeeperIO.h>
+#include <Coordination/ReadBufferFromNuraftBuffer.h>
+#include <Coordination/WriteBufferFromNuraftBuffer.h>
+#include <IO/WriteBufferFromFile.h>
+#include <IO/ReadBufferFromFile.h>
+#include <IO/copyData.h>
+#include <filesystem>
 
 namespace DB
 {
 
+namespace ErrorCodes
+{
+    extern const int UNKNOWN_FORMAT_VERSION;
+}
+
 namespace
 {
+    size_t getSnapshotPathUpToLogIdx(const String & snapshot_path)
+    {
+        std::filesystem::path path(snapshot_path);
+        std::string filename = path.stem();
+        Strings name_parts;
+        splitInto<'_'>(name_parts, filename);
+        return parse<size_t>(name_parts[1]);
+    }
+
+    std::string getSnapshotFileName(size_t up_to_log_idx)
+    {
+        return std::string{"snapshot_"} + std::to_string(up_to_log_idx) + ".bin";
+    }
+
+    String parentPath(const String & path)
+    {
+        auto rslash_pos = path.rfind('/');
+        if (rslash_pos > 0)
+            return path.substr(0, rslash_pos);
+        return "/";
+    }
+
     void writeNode(const NuKeeperStorage::Node & node, WriteBuffer & out)
     {
         Coordination::write(node.data, out);
@@ -29,62 +64,163 @@ namespace
     }
 }
 
-void NuKeeperStorageSerializer::serialize(const NuKeeperStorage & storage, WriteBuffer & out)
+
+void NuKeeperStorageSnapshot::serialize(const NuKeeperStorageSnapshot & snapshot, WriteBuffer & out)
 {
-    Coordination::write(storage.zxid, out);
+    Coordination::write(static_cast<uint8_t>(snapshot.version), out);
-    Coordination::write(storage.session_id_counter, out);
+    Coordination::write(snapshot.zxid, out);
-    Coordination::write(storage.container.size(), out);
+    Coordination::write(snapshot.session_id, out);
-    //for (const auto & [path, node] : storage.container)
+    Coordination::write(snapshot.snapshot_container_size, out);
-    //{
+    for (auto it = snapshot.begin; it != snapshot.end; ++it)
-    //    Coordination::write(path, out);
+    {
-    //    writeNode(node, out);
+        const auto & path = it->key;
-    //}
+        const auto & node = it->value;
-    Coordination::write(storage.ephemerals.size(), out);
+        Coordination::write(path, out);
-    for (const auto & [session_id, paths] : storage.ephemerals)
+        writeNode(node, out);
+    }
+
+    Coordination::write(snapshot.session_and_timeout.size(), out);
+    for (const auto & [session_id, timeout] : snapshot.session_and_timeout)
     {
         Coordination::write(session_id, out);
-        Coordination::write(paths.size(), out);
+        Coordination::write(timeout, out);
-        for (const auto & path : paths)
-            Coordination::write(path, out);
     }
 }
 
-void NuKeeperStorageSerializer::deserialize(NuKeeperStorage & storage, ReadBuffer & in)
+void NuKeeperStorageSnapshot::deserialize(NuKeeperStorage & storage, ReadBuffer & in)
 {
-    int64_t session_id_counter, zxid;
+    uint8_t version;
+    Coordination::read(version, in);
+    if (static_cast<SnapshotVersion>(version) > SnapshotVersion::V0)
+        throw Exception(ErrorCodes::UNKNOWN_FORMAT_VERSION, "Unsupported snapshot version {}", version);
+
+    int64_t session_id, zxid;
     Coordination::read(zxid, in);
-    Coordination::read(session_id_counter, in);
-    storage.zxid = zxid;
-    storage.session_id_counter = session_id_counter;
-
-    size_t container_size;
-    Coordination::read(container_size, in);
-
-    //size_t current_size = 0;
-    //while (current_size < container_size)
-    //{
-    //    std::string path;
-    //    Coordination::read(path, in);
-    //    NuKeeperStorage::Node node;
-    //    readNode(node, in);
-    //    storage.container[path] = node;
-    //    current_size++;
-    //}
-    size_t ephemerals_size;
-    Coordination::read(ephemerals_size, in);
-    while (storage.ephemerals.size() < ephemerals_size)
-    {
-        int64_t session_id;
-        size_t ephemerals_for_session;
     Coordination::read(session_id, in);
-        Coordination::read(ephemerals_for_session, in);
+    storage.zxid = zxid;
-        while (storage.ephemerals[session_id].size() < ephemerals_for_session)
+    storage.session_id_counter = session_id;
+
+    size_t snapshot_container_size;
+    Coordination::read(snapshot_container_size, in);
+
+    size_t current_size = 0;
+    while (current_size < snapshot_container_size)
     {
-            std::string ephemeral_path;
+        std::string path;
-            Coordination::read(ephemeral_path, in);
+        Coordination::read(path, in);
-            storage.ephemerals[session_id].emplace(ephemeral_path);
+        NuKeeperStorage::Node node;
+        readNode(node, in);
+        storage.container.insertOrReplace(path, node);
+        if (path != "/")
+        {
+            auto parent_path = parentPath(path);
+            storage.container.updateValue(parent_path, [&path] (NuKeeperStorage::Node & value) { value.children.insert(path); });
         }
+
+        if (node.ephemeral_owner != 0)
+            storage.ephemerals[node.ephemeral_owner].insert(path);
+
+        current_size++;
+    }
+
+    size_t active_sessions_size;
+    Coordination::read(active_sessions_size, in);
+    size_t current_session_size = 0;
+
+    while (current_session_size < active_sessions_size)
+    {
+        int64_t active_session_id, timeout;
+        Coordination::read(active_session_id, in);
+        Coordination::read(timeout, in);
+        storage.addSessionID(active_session_id, timeout);
+        current_session_size++;
     }
 }
 
+NuKeeperStorageSnapshot::NuKeeperStorageSnapshot(NuKeeperStorage * storage_, size_t up_to_log_idx_)
+    : storage(storage_)
+    , up_to_log_idx(up_to_log_idx_)
+    , zxid(storage->getZXID())
+    , session_id(storage->session_id_counter)
+{
+    storage->enableSnapshotMode();
+    snapshot_container_size = storage->container.snapshotSize();
+    begin = storage->getSnapshotIteratorBegin();
+    end = storage->getSnapshotIteratorEnd();
+    session_and_timeout = storage->getActiveSessions();
+}
+
+NuKeeperStorageSnapshot::~NuKeeperStorageSnapshot()
+{
+    storage->clearGarbageAfterSnapshot();
+    storage->disableSnapshotMode();
+}
+
+NuKeeperSnapshotManager::NuKeeperSnapshotManager(const std::string & snapshots_path_)
+    : snapshots_path(snapshots_path_)
+{
+    namespace fs = std::filesystem;
+
+    if (!fs::exists(snapshots_path))
+        fs::create_directories(snapshots_path);
+
+    for (const auto & p : fs::directory_iterator(snapshots_path))
+    {
+        size_t snapshot_up_to = getSnapshotPathUpToLogIdx(p.path());
+        existing_snapshots[snapshot_up_to] = p.path();
+    }
+}
+
+
+std::string NuKeeperSnapshotManager::serializeSnapshotBufferToDisk(nuraft::buffer & buffer, size_t up_to_log_idx)
+{
+    ReadBufferFromNuraftBuffer reader(buffer);
+
+    std::string new_snapshot_path = std::filesystem::path{snapshots_path} / getSnapshotFileName(up_to_log_idx);
+
+    WriteBufferFromFile plain_buf(new_snapshot_path);
+    copyData(reader, plain_buf);
+    plain_buf.sync();
+    return new_snapshot_path;
+}
+
+nuraft::ptr<nuraft::buffer> NuKeeperSnapshotManager::deserializeSnapshotBufferFromDisk(size_t up_to_log_idx) const
+{
+    const std::string & snapshot_path = existing_snapshots.at(up_to_log_idx);
+    WriteBufferFromNuraftBuffer writer;
+    ReadBufferFromFile reader(snapshot_path);
+    copyData(reader, writer);
+    return writer.getBuffer();
+}
+
+nuraft::ptr<nuraft::buffer> NuKeeperSnapshotManager::serializeSnapshotToBuffer(const NuKeeperStorageSnapshot & snapshot)
+{
+    WriteBufferFromNuraftBuffer writer;
+    CompressedWriteBuffer compressed_writer(writer);
+
+    NuKeeperStorageSnapshot::serialize(snapshot, compressed_writer);
+    return writer.getBuffer();
+}
+
+
+void NuKeeperSnapshotManager::deserializeSnapshotFromBuffer(NuKeeperStorage * storage, nuraft::ptr<nuraft::buffer> buffer)
+{
+    ReadBufferFromNuraftBuffer reader(buffer);
+    CompressedReadBuffer compressed_reader(reader);
+    NuKeeperStorageSnapshot::deserialize(*storage, compressed_reader);
+}
+
+size_t NuKeeperSnapshotManager::restoreFromLatestSnapshot(NuKeeperStorage * storage) const
+{
+    if (existing_snapshots.empty())
+        return 0 ;
+
+    auto log_id = existing_snapshots.rbegin()->first;
+    auto buffer = deserializeSnapshotBufferFromDisk(log_id);
+    deserializeSnapshotFromBuffer(storage, buffer);
+    return log_id;
+}
+
+
+
 }

src/Coordination/NuKeeperStorageSerializer.h

@@ -1,4 +1,5 @@
 #pragma once
+#include <libnuraft/nuraft.hxx> // Y_IGNORE
 #include <Coordination/NuKeeperStorage.h>
 #include <IO/WriteBuffer.h>
 #include <IO/ReadBuffer.h>
@@ -6,12 +7,49 @@
 namespace DB
 {
 
-class NuKeeperStorageSerializer
+enum SnapshotVersion : uint8_t
+{
+    V0 = 0,
+};
+
+struct NuKeeperStorageSnapshot
 {
 public:
-    static void serialize(const NuKeeperStorage & storage, WriteBuffer & out);
+    NuKeeperStorageSnapshot(NuKeeperStorage * storage_, size_t up_to_log_idx_);
+    ~NuKeeperStorageSnapshot();
+
+    static void serialize(const NuKeeperStorageSnapshot & snapshot, WriteBuffer & out);
 
     static void deserialize(NuKeeperStorage & storage, ReadBuffer & in);
+
+    NuKeeperStorage * storage;
+
+    SnapshotVersion version = SnapshotVersion::V0;
+    size_t up_to_log_idx;
+    int64_t zxid;
+    int64_t session_id;
+    size_t snapshot_container_size;
+    NuKeeperStorage::Container::const_iterator begin;
+    NuKeeperStorage::Container::const_iterator end;
+    SessionAndTimeout session_and_timeout;
+};
+
+class NuKeeperSnapshotManager
+{
+public:
+    explicit NuKeeperSnapshotManager(const std::string & snapshots_path_);
+
+    size_t restoreFromLatestSnapshot(NuKeeperStorage * storage) const;
+
+    std::string serializeSnapshotBufferToDisk(nuraft::buffer & buffer, size_t up_to_log_idx);
+    nuraft::ptr<nuraft::buffer> deserializeSnapshotBufferFromDisk(size_t up_to_log_idx) const;
+
+    static nuraft::ptr<nuraft::buffer> serializeSnapshotToBuffer(const NuKeeperStorageSnapshot & snapshot);
+    static void deserializeSnapshotFromBuffer(NuKeeperStorage * storage, nuraft::ptr<nuraft::buffer> buffer);
+
+private:
+    const std::string snapshots_path;
+    std::map<size_t, std::string> existing_snapshots;
 };
 
 }

src/Coordination/SnapshotableHashTable.h

@@ -50,6 +50,33 @@ public:
         return false;
     }
 
+
+    void insertOrReplace(const std::string & key, const V & value)
+    {
+        auto it = map.find(key);
+        if (it == map.end())
+        {
+            ListElem elem{key, value, true};
+            auto itr = list.insert(list.end(), elem);
+            map.emplace(itr->key, itr);
+        }
+        else
+        {
+            auto list_itr = it->second;
+            if (snapshot_mode)
+            {
+                ListElem elem{key, value, true};
+                list_itr->active_in_map = false;
+                auto new_list_itr = list.insert(list.end(), elem);
+                map[new_list_itr->key] = new_list_itr;
+            }
+            else
+            {
+                list_itr->value = value;
+            }
+        }
+    }
+
     bool erase(const std::string & key)
     {
         auto it = map.find(key);