Keep track of the number of async tables loading jobs. If there are some running jobs, do not update tail_ptr in TransactionLog::removeOldEntries

tuanpach · tuanpach · commit 349c46105730 · 2025-07-07T07:29:52.000Z
diff --git a/src/Databases/DatabaseOrdinary.cpp b/src/Databases/DatabaseOrdinary.cpp
@@ -36,6 +36,7 @@
 #include <Common/logger_useful.h>
 #include <Common/quoteString.h>
 #include <Common/typeid_cast.h>
+#include <Interpreters/TransactionLog.h>
 
 #include <boost/algorithm/string/replace.hpp>
 
@@ -409,13 +410,15 @@ LoadTaskPtr DatabaseOrdinary::loadTableFromMetadataAsync(
     const ASTPtr & ast,
     LoadingStrictnessLevel mode)
 {
+    TransactionLog::increaseAsyncTablesLoadingJobNumber();
     std::scoped_lock lock(mutex);
     auto job = makeLoadJob(
         std::move(load_after),
         TablesLoaderBackgroundLoadPoolId,
         fmt::format("load table {}", name.getFullName()),
-        [this, local_context, file_path, name, ast, mode] (AsyncLoader &, const LoadJobPtr &)
+        [this, local_context, file_path, name, ast, mode](AsyncLoader &, const LoadJobPtr &)
         {
+            SCOPE_EXIT(TransactionLog::decreaseAsyncTablesLoadingJobNumber(););
             loadTableFromMetadata(local_context, file_path, name, ast, mode);
         });
 
diff --git a/src/Interpreters/TransactionLog.cpp b/src/Interpreters/TransactionLog.cpp
@@ -1,3 +1,4 @@
+#include <atomic>
 #include <Core/ServerUUID.h>
 #include <IO/ReadBufferFromString.h>
 #include <IO/ReadHelpers.h>
@@ -275,6 +276,17 @@ void TransactionLog::removeOldEntries()
     if (!global_context->isServerCompletelyStarted())
         return;
 
+    /// Because `loadTableFromMetadataAsync` is running asynchronously, it is possible that the outdated parts are loading while the `tail_ptr` is updated here.
+    /// It might trigger assertion when the part `create_csn` is lower than `tail_ptr`. Refer: https://github.com/ClickHouse/ClickHouse/issues/60406
+    /// We keep track of `asyncTablesLoadingJobNumber`, and not update `tail_ptr` if there are running jobs.
+    if (!updated_tail_ptr.load(std::memory_order_relaxed) && asyncTablesLoadingJobNumber() != 0)
+    {
+        LOG_TRACE(log, "There are running async tables loading jobs, skip updating tail_ptr");
+        return;
+    }
+
+    updated_tail_ptr.store(true, std::memory_order_relaxed);
+
     /// Also similar problem is possible if some table was not attached during startup (for example, if table is detached permanently).
     /// Also we write CSNs into data parts without fsync, so it's theoretically possible that we wrote CSN, finished transaction,
     /// removed its entry from the log, but after that server restarts and CSN is not actually saved to metadata on disk.
@@ -638,4 +650,16 @@ void TransactionLog::sync() const
     waitForCSNLoaded(newest_csn);
 }
 
+void TransactionLog::increaseAsyncTablesLoadingJobNumber()
+{
+    async_tables_loading_job_number.fetch_add(1);
+}
+void TransactionLog::decreaseAsyncTablesLoadingJobNumber()
+{
+    async_tables_loading_job_number.fetch_sub(1);
+}
+Int64 TransactionLog::asyncTablesLoadingJobNumber()
+{
+    return async_tables_loading_job_number.load();
+}
 }
diff --git a/src/Interpreters/TransactionLog.h b/src/Interpreters/TransactionLog.h
@@ -1,11 +1,12 @@
 #pragma once
+#include <mutex>
+#include <unordered_map>
 #include <Interpreters/MergeTreeTransaction.h>
 #include <Interpreters/MergeTreeTransactionHolder.h>
-#include <Common/ZooKeeper/ZooKeeper.h>
-#include <Common/ThreadPool.h>
+#include <base/types.h>
 #include <boost/noncopyable.hpp>
-#include <mutex>
-#include <unordered_map>
+#include <Common/ThreadPool.h>
+#include <Common/ZooKeeper/ZooKeeper.h>
 
 namespace DB
 {
@@ -131,6 +132,10 @@ class TransactionLog final : public SingletonHelper<TransactionLog>
 
     void sync() const;
 
+    static void increaseAsyncTablesLoadingJobNumber();
+    static void decreaseAsyncTablesLoadingJobNumber();
+    static Int64 asyncTablesLoadingJobNumber();
+
 private:
     void loadLogFromZooKeeper() TSA_REQUIRES(mutex);
     void runUpdatingThread();
@@ -148,6 +153,8 @@ class TransactionLog final : public SingletonHelper<TransactionLog>
     static TransactionID deserializeTID(const String & csn_node_content);
     static String serializeTID(const TransactionID & tid);
 
+    inline static std::atomic<Int64> async_tables_loading_job_number{0};
+
     ZooKeeperPtr getZooKeeper() const;
 
     /// Some time a transaction could be committed concurrently, in order to resolve it provide failback_with_strict_load_csn
@@ -191,6 +198,7 @@ class TransactionLog final : public SingletonHelper<TransactionLog>
     String last_loaded_entry TSA_GUARDED_BY(mutex);
     /// The oldest CSN such that we store in log entries with TransactionIDs containing this CSN.
     std::atomic<CSN> tail_ptr = Tx::UnknownCSN;
+    std::atomic<bool> updated_tail_ptr{false};
 
     zkutil::EventPtr log_updated_event = std::make_shared<Poco::Event>();
 
