Merge pull request ClickHouse#83015 from ClickHouse/backport/25.6/82485

clickhouse-gh[bot] · web-flow · commit bab75879b1db · 2025-07-01T19:25:54.000+01:00
Backport ClickHouse#82485 to 25.6: Keeper improvement: move changelog files between disk in a background thread
diff --git a/src/Coordination/Changelog.cpp b/src/Coordination/Changelog.cpp
@@ -2,6 +2,7 @@
 #include <filesystem>
 #include <mutex>
 #include <ranges>
+#include <variant>
 #include <Coordination/Changelog.h>
 #include <Coordination/Keeper4LWInfo.h>
 #include <Coordination/KeeperContext.h>
@@ -47,6 +48,7 @@ namespace ErrorCodes
     extern const int NOT_IMPLEMENTED;
     extern const int BAD_ARGUMENTS;
     extern const int LOGICAL_ERROR;
+    extern const int SYSTEM_ERROR;
 }
 
 namespace
@@ -125,6 +127,41 @@ Checksum computeRecordChecksum(const ChangelogRecord & record)
     return hash.get64();
 }
 
+struct RemoveChangelog
+{
+};
+
+struct MoveChangelog
+{
+    std::string new_path;
+    DiskPtr new_disk;
+};
+
+}
+
+using ChangelogFileOperationVariant = std::variant<RemoveChangelog, MoveChangelog>;
+
+struct ChangelogFileOperation
+{
+    explicit ChangelogFileOperation(ChangelogFileDescriptionPtr changelog_, ChangelogFileOperationVariant operation_)
+        : changelog(std::move(changelog_))
+        , operation(std::move(operation_))
+    {}
+
+    ChangelogFileDescriptionPtr changelog;
+    ChangelogFileOperationVariant operation;
+    std::atomic<bool> done = false;
+};
+
+void ChangelogFileDescription::waitAllAsyncOperations()
+{
+    for (const auto & op : file_operations)
+    {
+        if (auto op_locked = op.lock())
+            op_locked->done.wait(false);
+    }
+
+    file_operations.clear();
 }
 
 /// Appendable log writer
@@ -134,17 +171,20 @@ Checksum computeRecordChecksum(const ChangelogRecord & record)
 /// At least 1 log record should be contained in each log
 class ChangelogWriter
 {
+    using MoveChangelogCallback = std::function<void(ChangelogFileDescriptionPtr, std::string, DiskPtr)>;
 public:
     ChangelogWriter(
         std::map<uint64_t, ChangelogFileDescriptionPtr> & existing_changelogs_,
         LogEntryStorage & entry_storage_,
         KeeperContextPtr keeper_context_,
-        LogFileSettings log_file_settings_)
+        LogFileSettings log_file_settings_,
+        MoveChangelogCallback move_changelog_cb_)
         : existing_changelogs(existing_changelogs_)
         , entry_storage(entry_storage_)
         , log_file_settings(log_file_settings_)
         , keeper_context(std::move(keeper_context_))
         , log(getLogger("Changelog"))
+        , move_changelog_cb(std::move(move_changelog_cb_))
     {
     }
 
@@ -190,25 +230,8 @@ class ChangelogWriter
                             current_file_description->extension);
                     }
 
-                    if (disk == log_disk)
-                    {
-                        if (path != new_path)
-                        {
-                            try
-                            {
-                                disk->moveFile(path, new_path);
-                            }
-                            catch (...)
-                            {
-                                tryLogCurrentException(log, fmt::format("File rename failed on disk {}", disk->getName()));
-                            }
-                            current_file_description->path = std::move(new_path);
-                        }
-                    }
-                    else
-                    {
-                        moveChangelogBetweenDisks(log_disk, current_file_description, disk, new_path, keeper_context);
-                    }
+                    if (move_changelog_cb)
+                        move_changelog_cb(current_file_description, std::move(new_path), disk);
                 }
             }
             else
@@ -516,6 +539,8 @@ class ChangelogWriter
     KeeperContextPtr keeper_context;
 
     LoggerPtr const log;
+
+    MoveChangelogCallback move_changelog_cb;
 };
 
 namespace
@@ -1452,11 +1477,11 @@ LogEntriesPtr LogEntryStorage::getLogEntriesBetween(uint64_t start, uint64_t end
         if (!read_info)
             return;
 
-        LOG_TRACE(log, "Reading from path {} {} entries", read_info->file_description->path, read_info->count);
         read_info->file_description->withLock(
             [&]
             {
                 const auto & [file_description, start_position, count] = *read_info;
+                LOG_TRACE(log, "Reading from path {} {} entries", file_description->path, read_info->count);
                 auto file = file_description->disk->readFile(file_description->path, getReadSettings());
                 file->seek(start_position, SEEK_SET);
 
@@ -1694,13 +1719,19 @@ Changelog::Changelog(
         if (existing_changelogs.empty())
             LOG_WARNING(log, "No logs exists in {}. It's Ok if it's the first run of clickhouse-keeper.", disk->getPath());
 
-        clean_log_thread = std::make_unique<ThreadFromGlobalPool>([this] { cleanLogThread(); });
+        background_changelog_operations_thread = std::make_unique<ThreadFromGlobalPool>([this] { backgroundChangelogOperationsThread(); });
 
         write_thread = std::make_unique<ThreadFromGlobalPool>([this] { writeThread(); });
 
         append_completion_thread = std::make_unique<ThreadFromGlobalPool>([this] { appendCompletionThread(); });
 
-        current_writer = std::make_unique<ChangelogWriter>(existing_changelogs, entry_storage, keeper_context, log_file_settings);
+        current_writer = std::make_unique<ChangelogWriter>(
+            existing_changelogs,
+            entry_storage,
+            keeper_context,
+            log_file_settings,
+            /*move_changelog_cb=*/[&](ChangelogFileDescriptionPtr changelog, std::string new_path, DiskPtr new_disk)
+            { moveChangelogAsync(std::move(changelog), std::move(new_path), std::move(new_disk)); });
     }
     catch (...)
     {
@@ -1844,7 +1875,7 @@ try
             removeAllLogsAfter(last_log_read_result->log_start_index);
 
             /// This log, even if it finished with error shouldn't be removed
-            chassert(existing_changelogs.find(last_log_read_result->log_start_index) != existing_changelogs.end());
+            chassert(existing_changelogs.contains(last_log_read_result->log_start_index));
             chassert(existing_changelogs.find(last_log_read_result->log_start_index)->first == existing_changelogs.rbegin()->first);
         };
 
@@ -2190,6 +2221,7 @@ void Changelog::writeAt(uint64_t index, const LogEntryPtr & log_entry)
             else
                 description = std::prev(index_changelog)->second;
 
+            description->waitAllAsyncOperations();
             /// if the changelog is broken at end, we cannot append it with new logs
             /// we create a new file starting with the required index
             if (description->broken_at_end)
@@ -2212,7 +2244,7 @@ void Changelog::writeAt(uint64_t index, const LogEntryPtr & log_entry)
             auto to_remove_itr = existing_changelogs.upper_bound(index);
             for (auto itr = to_remove_itr; itr != existing_changelogs.end();)
             {
-                itr->second->disk->removeFile(itr->second->path);
+                removeChangelogAsync(itr->second);
                 itr = existing_changelogs.erase(itr);
             }
         }
@@ -2247,6 +2279,7 @@ void Changelog::compact(uint64_t up_to_log_index)
     for (auto itr = existing_changelogs.begin(); itr != existing_changelogs.end();)
     {
         auto & changelog_description = *itr->second;
+        auto path = changelog_description.getPathSafe();
         /// Remove all completely outdated changelog files
         if (remove_all_logs || changelog_description.to_log_index <= up_to_log_index)
         {
@@ -2255,31 +2288,12 @@ void Changelog::compact(uint64_t up_to_log_index)
                 LOG_INFO(
                     log,
                     "Trying to remove log {} which is current active log for write. Possibly this node recovers from snapshot",
-                    changelog_description.path);
+                    path);
                 need_rotate = true;
             }
 
-            LOG_INFO(log, "Removing changelog {} because of compaction", changelog_description.path);
-
-            /// If failed to push to queue for background removing, then we will remove it now
-            if (!log_files_to_delete_queue.tryPush({changelog_description.path, changelog_description.disk}, 1))
-            {
-                try
-                {
-                    changelog_description.disk->removeFile(changelog_description.path);
-                    LOG_INFO(log, "Removed changelog {} because of compaction.", changelog_description.path);
-                }
-                catch (Exception & e)
-                {
-                    LOG_WARNING(
-                        log, "Failed to remove changelog {} in compaction, error message: {}", changelog_description.path, e.message());
-                }
-                catch (...)
-                {
-                    tryLogCurrentException(log);
-                }
-            }
-
+            LOG_INFO(log, "Removing changelog {} because of compaction", path);
+            removeChangelogAsync(itr->second);
             changelog_description.deleted = true;
 
             itr = existing_changelogs.erase(itr);
@@ -2434,11 +2448,11 @@ uint64_t Changelog::size() const
 void Changelog::shutdown()
 {
     LOG_DEBUG(log, "Shutting down Changelog");
-    if (!log_files_to_delete_queue.isFinished())
-        log_files_to_delete_queue.finish();
+    if (!changelog_operation_queue.isFinished())
+        changelog_operation_queue.finish();
 
-    if (clean_log_thread->joinable())
-        clean_log_thread->join();
+    if (background_changelog_operations_thread->joinable())
+        background_changelog_operations_thread->join();
 
     if (!write_operations.isFinished())
         write_operations.finish();
@@ -2482,26 +2496,83 @@ Changelog::~Changelog()
     }
 }
 
-void Changelog::cleanLogThread()
+void Changelog::backgroundChangelogOperationsThread()
 {
-    std::pair<std::string, DiskPtr> path_with_disk;
-    while (log_files_to_delete_queue.pop(path_with_disk))
+    ChangelogFileOperationPtr changelog_operation;
+    while (changelog_operation_queue.pop(changelog_operation))
     {
-        const auto & [path, disk] = path_with_disk;
-        try
+        if (std::holds_alternative<RemoveChangelog>(changelog_operation->operation))
         {
-            disk->removeFile(path);
-            LOG_INFO(log, "Removed changelog {} because of compaction.", path);
+            chassert(changelog_operation->changelog);
+            const auto & changelog = *changelog_operation->changelog;
+            try
+            {
+                changelog.disk->removeFile(changelog.path);
+                LOG_INFO(log, "Removed changelog {} because of compaction.", changelog.path);
+            }
+            catch (Exception & e)
+            {
+                LOG_WARNING(log, "Failed to remove changelog {} in compaction, error message: {}", changelog.path, e.message());
+            }
+            catch (...)
+            {
+                tryLogCurrentException(log);
+            }
         }
-        catch (Exception & e)
+        else if (auto * move_operation = std::get_if<MoveChangelog>(&changelog_operation->operation))
         {
-            LOG_WARNING(log, "Failed to remove changelog {} in compaction, error message: {}", path, e.message());
+            const auto & changelog = changelog_operation->changelog;
+
+            if (move_operation->new_disk == changelog->disk)
+            {
+                if (move_operation->new_path != changelog->path)
+                {
+                    try
+                    {
+                        changelog->disk->moveFile(changelog->path, move_operation->new_path);
+                    }
+                    catch (...)
+                    {
+                        tryLogCurrentException(log, fmt::format("File rename failed on disk {}", changelog->disk->getName()));
+                    }
+                    changelog->path = std::move(move_operation->new_path);
+                }
+            }
+            else
+            {
+                moveChangelogBetweenDisks(changelog->disk, changelog, move_operation->new_disk, move_operation->new_path, keeper_context);
+            }
         }
-        catch (...)
+        else
         {
-            tryLogCurrentException(log);
+            LOG_ERROR(log, "Unsupported operation detected for changelog {}", changelog_operation->changelog->path);
+            chassert(false);
         }
+        changelog_operation->done = true;
+    }
+}
+
+void Changelog::modifyChangelogAsync(ChangelogFileOperationPtr changelog_operation)
+{
+    if (!changelog_operation_queue.tryPush(changelog_operation, 60 * 1000))
+    {
+        throw DB::Exception(
+            ErrorCodes::SYSTEM_ERROR, "Background thread for changelog operations is stuck or not keeping up with operations");
     }
+
+    changelog_operation->changelog->file_operations.push_back(changelog_operation);
+}
+
+void Changelog::removeChangelogAsync(ChangelogFileDescriptionPtr changelog)
+{
+    modifyChangelogAsync(std::make_shared<ChangelogFileOperation>(std::move(changelog), RemoveChangelog{}));
+}
+
+void Changelog::moveChangelogAsync(ChangelogFileDescriptionPtr changelog, std::string new_path, DiskPtr new_disk)
+{
+    modifyChangelogAsync(
+        std::make_shared<ChangelogFileOperation>(
+            std::move(changelog), MoveChangelog{.new_path = std::move(new_path), .new_disk = std::move(new_disk)}));
 }
 
 void Changelog::setRaftServer(const nuraft::ptr<nuraft::raft_server> & raft_server_)
diff --git a/src/Coordination/Changelog.h b/src/Coordination/Changelog.h
@@ -67,6 +67,9 @@ struct ChangelogRecord
     nuraft::ptr<nuraft::buffer> blob;
 };
 
+struct ChangelogFileOperation;
+using ChangelogFileOperationPtr = std::shared_ptr<ChangelogFileOperation>;
+
 /// changelog_fromindex_toindex.bin
 /// [fromindex, toindex] <- inclusive
 struct ChangelogFileDescription
@@ -85,6 +88,8 @@ struct ChangelogFileDescription
 
     bool deleted = false;
 
+    std::deque<std::weak_ptr<ChangelogFileOperation>> file_operations;
+
     /// How many entries should be stored in this log
     uint64_t expectedEntriesCountInLog() const { return to_log_index - from_log_index + 1; }
 
@@ -94,6 +99,14 @@ struct ChangelogFileDescription
         std::lock_guard lock(file_mutex);
         fn();
     }
+
+    std::string getPathSafe()
+    {
+        std::lock_guard lock(file_mutex);
+        return path;
+    }
+
+    void waitAllAsyncOperations();
 };
 
 using ChangelogFileDescriptionPtr = std::shared_ptr<ChangelogFileDescription>;
@@ -402,8 +415,12 @@ class Changelog
     /// Init writer for existing log with some entries already written
     void initWriter(ChangelogFileDescriptionPtr description);
 
-    /// Clean useless log files in a background thread
-    void cleanLogThread();
+    /// Thread for operations on changelog file, e.g. removing the file
+    void backgroundChangelogOperationsThread();
+
+    void modifyChangelogAsync(ChangelogFileOperationPtr changelog_operation);
+    void removeChangelogAsync(ChangelogFileDescriptionPtr changelog);
+    void moveChangelogAsync(ChangelogFileDescriptionPtr changelog, std::string new_path, DiskPtr new_disk);
 
     const String changelogs_detached_dir;
     const uint64_t rotate_interval;
@@ -417,10 +434,9 @@ class Changelog
     LogEntryStorage entry_storage;
 
     uint64_t max_log_id = 0;
-    /// For compaction, queue of delete not used logs
-    /// 128 is enough, even if log is not removed, it's not a problem
-    ConcurrentBoundedQueue<std::pair<std::string, DiskPtr>> log_files_to_delete_queue{128};
-    std::unique_ptr<ThreadFromGlobalPool> clean_log_thread;
+
+    ConcurrentBoundedQueue<ChangelogFileOperationPtr> changelog_operation_queue{std::numeric_limits<size_t>::max()};
+    std::unique_ptr<ThreadFromGlobalPool> background_changelog_operations_thread;
 
     struct AppendLog
     {
diff --git a/tests/integration/test_keeper_disks/test.py b/tests/integration/test_keeper_disks/test.py
