Fix: refactor wait cleanup alg.

Author: small-turtle-1

src/main/query_context.cpp

@@ -50,6 +50,7 @@ import session_manager;
 import base_statement;
 import parser_result;
 import parser_assert;
+import plan_fragment;
 
 namespace infinity {
 
@@ -106,6 +107,10 @@ QueryResult QueryContext::Query(const String &query) {
 
 QueryResult QueryContext::QueryStatement(const BaseStatement *statement) {
     QueryResult query_result;
+    SharedPtr<LogicalNode> logical_plan = nullptr;
+    UniquePtr<PlanFragment> plan_fragment = nullptr;
+    UniquePtr<PhysicalOperator> physical_plan = nullptr;
+
 //    ProfilerStart("Query");
 //    BaseProfiler profiler;
 //    profiler.Begin();
@@ -127,7 +132,7 @@ QueryResult QueryContext::QueryStatement(const BaseStatement *statement) {
         }
 
         current_max_node_id_ = bind_context->GetNewLogicalNodeId();
-        SharedPtr<LogicalNode> logical_plan = logical_planner_->LogicalPlan();
+        logical_plan = logical_planner_->LogicalPlan();
         StopProfile(QueryPhase::kLogicalPlan);
 //        LOG_WARN(fmt::format("Before optimizer cost: {}", profiler.ElapsedToString()));
         // Apply optimized rule to the logical plan
@@ -137,13 +142,13 @@ QueryResult QueryContext::QueryStatement(const BaseStatement *statement) {
 
         // Build physical plan
         StartProfile(QueryPhase::kPhysicalPlan);
-        UniquePtr<PhysicalOperator> physical_plan = physical_planner_->BuildPhysicalOperator(logical_plan);
+        physical_plan = physical_planner_->BuildPhysicalOperator(logical_plan);
         StopProfile(QueryPhase::kPhysicalPlan);
 //        LOG_WARN(fmt::format("Before pipeline cost: {}", profiler.ElapsedToString()));
         StartProfile(QueryPhase::kPipelineBuild);
         // Fragment Builder, only for test now.
         // SharedPtr<PlanFragment> plan_fragment = fragment_builder.Build(physical_plan);
-        auto plan_fragment = fragment_builder_->BuildFragment(physical_plan.get());
+        plan_fragment = fragment_builder_->BuildFragment(physical_plan.get());
         StopProfile(QueryPhase::kPipelineBuild);
 
         auto notifier = MakeUnique<Notifier>();
@@ -159,14 +164,7 @@ QueryResult QueryContext::QueryStatement(const BaseStatement *statement) {
         StopProfile(QueryPhase::kExecution);
 //        LOG_WARN(fmt::format("Before commit cost: {}", profiler.ElapsedToString()));
         StartProfile(QueryPhase::kCommit);
-        try {
-            this->CommitTxn();
-        } catch (RecoverableException &e) {
-            StopProfile();
-            this->RollbackTxn();
-            query_result.result_table_ = nullptr;
-            query_result.status_.Init(e.ErrorCode(), e.what());
-        }
+        this->CommitTxn();
         StopProfile(QueryPhase::kCommit);
 
     } catch (RecoverableException &e) {

src/storage/bg_task/periodic_trigger.cpp

@@ -28,7 +28,7 @@ import third_party;
 namespace infinity {
 
 void CleanupPeriodicTrigger::Trigger() {
-    TxnTimeStamp visible_ts = txn_mgr_->GetMinUnflushedTS();
+    TxnTimeStamp visible_ts = wal_mgr_->GetLastCkpTS() + 1;
     // if (visible_ts == last_visible_ts_) {
     //     LOG_TRACE(fmt::format("Skip cleanup. visible timestamp: {}", visible_ts));
     //     return;
@@ -40,17 +40,15 @@ void CleanupPeriodicTrigger::Trigger() {
     last_visible_ts_ = visible_ts;
     LOG_TRACE(fmt::format("Cleanup visible timestamp: {}", visible_ts));
 
-    auto buffer_mgr = txn_mgr_->GetBufferMgr();
-    auto cleanup_task = MakeShared<CleanupTask>(catalog_, visible_ts, buffer_mgr);
+    auto cleanup_task = MakeShared<CleanupTask>(catalog_, visible_ts, buffer_mgr_);
     bg_processor_->Submit(std::move(cleanup_task));
 }
 
 void CheckpointPeriodicTrigger::Trigger() {
     auto checkpoint_task = MakeShared<CheckpointTask>(is_full_checkpoint_);
     LOG_INFO(fmt::format("Trigger {} periodic checkpoint.", is_full_checkpoint_ ? "FULL" : "DELTA"));
     if (!wal_mgr_->TrySubmitCheckpointTask(std::move(checkpoint_task))) {
-        LOG_INFO(
-            fmt::format("Skip {} checkpoint(time) because there is already a checkpoint task running.", is_full_checkpoint_ ? "FULL" : "DELTA"));
+        LOG_INFO(fmt::format("Skip {} checkpoint(time) because there is already a checkpoint task running.", is_full_checkpoint_ ? "FULL" : "DELTA"));
     }
 }
 

src/storage/bg_task/periodic_trigger.cppm

@@ -20,8 +20,8 @@ import stl;
 import background_process;
 import compaction_process;
 import catalog;
-import txn_manager;
 import wal_manager;
+import buffer_manager;
 
 namespace infinity {
 
@@ -51,15 +51,20 @@ private:
 
 export class CleanupPeriodicTrigger final : public PeriodicTrigger {
 public:
-    CleanupPeriodicTrigger(std::chrono::milliseconds interval, BGTaskProcessor *bg_processor, Catalog *catalog, TxnManager *txn_mgr)
-        : PeriodicTrigger(interval), bg_processor_(bg_processor), catalog_(catalog), txn_mgr_(txn_mgr) {}
+    CleanupPeriodicTrigger(std::chrono::milliseconds interval,
+                           BGTaskProcessor *bg_processor,
+                           Catalog *catalog,
+                           WalManager *wal_mgr,
+                           BufferManager *buffer_mgr)
+        : PeriodicTrigger(interval), bg_processor_(bg_processor), catalog_(catalog), wal_mgr_(wal_mgr), buffer_mgr_(buffer_mgr) {}
 
     virtual void Trigger() override;
 
 private:
     BGTaskProcessor *const bg_processor_{};
     Catalog *const catalog_{};
-    TxnManager *const txn_mgr_{};
+    WalManager *const wal_mgr_{};
+    BufferManager *const buffer_mgr_{};
 
     TxnTimeStamp last_visible_ts_{0};
 };

src/storage/meta/catalog.cpp

@@ -77,8 +77,6 @@ Catalog::~Catalog() {
     mem_index_commit_thread_.join();
 }
 
-void Catalog::SetTxnMgr(TxnManager *txn_mgr) { txn_mgr_ = txn_mgr; }
-
 // do not only use this method to create database
 // it will not record database in transaction, so when you commit transaction
 // it will lose operation
@@ -887,8 +885,6 @@ bool Catalog::SaveDeltaCatalog(TxnTimeStamp max_commit_ts, String &delta_catalog
     // Check the SegmentEntry's for flush the data to disk.
     UniquePtr<CatalogDeltaEntry> flush_delta_entry = global_catalog_delta_entry_->PickFlushEntry(max_commit_ts);
 
-    DeferFn defer_fn([&]() { txn_mgr_->RemoveWaitFlushTxns(flush_delta_entry->txn_ids()); });
-
     if (flush_delta_entry->operations().empty()) {
         LOG_TRACE("Save delta catalog ops is empty. Skip flush.");
         return true;

src/storage/meta/catalog.cppm

@@ -106,8 +106,6 @@ public:
 
     ~Catalog();
 
-    void SetTxnMgr(TxnManager *txn_mgr);
-
 public:
     // Database related functions
     Tuple<DBEntry *, Status> CreateDatabase(const String &db_name,
@@ -291,8 +289,6 @@ public:
 
     ProfileHistory history{DEFAULT_PROFILER_HISTORY_SIZE};
 
-    TxnManager *txn_mgr_{nullptr};
-
 private: // TODO: remove this
     std::shared_mutex &rw_locker() { return db_meta_map_.rw_locker_; }
 

src/storage/storage.cpp

@@ -124,7 +124,7 @@ void Storage::Init() {
         std::chrono::seconds cleanup_interval = static_cast<std::chrono::seconds>(config_ptr_->CleanupInterval());
         if (cleanup_interval.count() > 0) {
             periodic_trigger_thread_->AddTrigger(
-                MakeUnique<CleanupPeriodicTrigger>(cleanup_interval, bg_processor_.get(), new_catalog_.get(), txn_mgr_.get()));
+                MakeUnique<CleanupPeriodicTrigger>(cleanup_interval, bg_processor_.get(), new_catalog_.get(), wal_mgr_.get(), buffer_mgr_.get()));
         } else {
             LOG_WARN("Cleanup interval is not set, auto cleanup task will not be triggered");
         }

src/storage/txn/txn_manager.cpp

@@ -43,9 +43,7 @@ TxnManager::TxnManager(Catalog *catalog,
                        TxnTimeStamp start_ts,
                        bool enable_compaction)
     : catalog_(catalog), buffer_mgr_(buffer_mgr), bg_task_processor_(bg_task_processor), wal_mgr_(wal_mgr), start_ts_(start_ts), is_running_(false),
-      enable_compaction_(enable_compaction) {
-    catalog_->SetTxnMgr(this);
-}
+      enable_compaction_(enable_compaction) {}
 
 Txn *TxnManager::BeginTxn(UniquePtr<String> txn_text) {
     // Check if the is_running_ is true
@@ -68,7 +66,7 @@ Txn *TxnManager::BeginTxn(UniquePtr<String> txn_text) {
 
     // Storage txn in txn manager
     txn_map_[new_txn_id] = new_txn;
-    ts_map_.emplace(ts, new_txn_id);
+    // ts_map_.emplace(ts, new_txn_id);
     beginned_txns_.emplace_back(new_txn);
     rw_locker_.unlock();
 
@@ -184,8 +182,6 @@ void TxnManager::AddDeltaEntry(UniquePtr<CatalogDeltaEntry> delta_entry) {
         UnrecoverableError("TxnManager is not running, cannot add delta entry");
     }
     i64 wal_size = wal_mgr_->WalSize();
-    const auto &txn_ids = delta_entry->txn_ids();
-    this->AddWaitFlushTxn(txn_ids);
     bg_task_processor_->Submit(MakeShared<AddDeltaEntryTask>(std::move(delta_entry), wal_size));
 }
 
@@ -294,46 +290,4 @@ void TxnManager::FinishTxn(Txn *txn) {
     }
 }
 
-void TxnManager::AddWaitFlushTxn(const Vector<TransactionID> &txn_ids) {
-    std::stringstream ss;
-    for (auto txn_id : txn_ids) {
-        ss << txn_id << " ";
-    }
-    LOG_INFO(fmt::format("Add txns: {} to wait flush set", ss.str()));
-    std::unique_lock w_lock(rw_locker_);
-    wait_flush_txns_.insert(txn_ids.begin(), txn_ids.end());
-}
-
-void TxnManager::RemoveWaitFlushTxns(const Vector<TransactionID> &txn_ids) {
-    std::stringstream ss2;
-    for (auto txn_id : txn_ids) {
-        ss2 << txn_id << " ";
-    }
-    LOG_INFO(fmt::format("Remove txn: {} from wait flush set", ss2.str()));
-    std::unique_lock w_lock(rw_locker_);
-    for (auto txn_id : txn_ids) {
-        if (!wait_flush_txns_.erase(txn_id)) {
-            UnrecoverableError(fmt::format("Txn: {} not found in wait flush set", txn_id));
-        }
-    }
-}
-
-TxnTimeStamp TxnManager::GetMinUnflushedTS() {
-    std::unique_lock w_locker(rw_locker_);
-    for (auto iter = ts_map_.begin(); iter != ts_map_.end();) {
-        auto &[ts, txn_id] = *iter;
-        if (txn_map_.find(txn_id) != txn_map_.end()) {
-            LOG_INFO(fmt::format("Txn: {} found in txn map", txn_id));
-            return ts;
-        }
-        if (wait_flush_txns_.find(txn_id) != wait_flush_txns_.end()) {
-            LOG_INFO(fmt::format("Txn: {} wait flush", txn_id));
-            return ts;
-        }
-        iter = ts_map_.erase(iter);
-    }
-    LOG_INFO(fmt::format("No txn is active, return the next ts {}", start_ts_));
-    return start_ts_;
-}
-
 } // namespace infinity

src/storage/txn/txn_manager.cppm

@@ -88,12 +88,7 @@ public:
 private:
     void FinishTxn(Txn *txn);
 
-    void AddWaitFlushTxn(const Vector<TransactionID> &txn_ids);
-
 public:
-    void RemoveWaitFlushTxns(const Vector<TransactionID> &txn_ids);
-
-    TxnTimeStamp GetMinUnflushedTS();
 
     bool enable_compaction() const { return enable_compaction_; }
 
@@ -113,9 +108,6 @@ private:
     Map<TxnTimeStamp, WalEntry *> wait_conflict_ck_{}; // sorted by commit ts
 
     Atomic<TxnTimeStamp> start_ts_{}; // The next txn ts
-    // Deque<TxnTimeStamp> ts_queue_{}; // the ts queue
-    Map<TxnTimeStamp, TransactionID> ts_map_{}; // optimize the data structure
-    HashSet<TransactionID> wait_flush_txns_{};
 
     // For stop the txn manager
     atomic_bool is_running_{false};

src/storage/wal/wal_manager.cpp

@@ -58,10 +58,14 @@ module wal_manager;
 
 namespace infinity {
 
-WalManager::WalManager(Storage *storage, String wal_dir, u64 wal_size_threshold, u64 delta_checkpoint_interval_wal_bytes, FlushOptionType flush_option)
+WalManager::WalManager(Storage *storage,
+                       String wal_dir,
+                       u64 wal_size_threshold,
+                       u64 delta_checkpoint_interval_wal_bytes,
+                       FlushOptionType flush_option)
     : cfg_wal_size_threshold_(wal_size_threshold), cfg_delta_checkpoint_interval_wal_bytes_(delta_checkpoint_interval_wal_bytes), wal_dir_(wal_dir),
       wal_path_(wal_dir + "/" + WalFile::TempWalFilename()), storage_(storage), running_(false), flush_option_(flush_option), last_ckp_wal_size_(0),
-      checkpoint_in_progress_(false), last_ckp_ts_(UNCOMMIT_TS), last_full_ckp_ts_(UNCOMMIT_TS) {}
+      checkpoint_in_progress_(false), last_ckp_ts_(0), last_full_ckp_ts_(0) {}
 
 WalManager::~WalManager() {
     if (running_.load()) {
@@ -142,6 +146,8 @@ i64 WalManager::WalSize() const {
     return wal_size_;
 }
 
+TxnTimeStamp WalManager::GetLastCkpTS() { return last_ckp_ts_; }
+
 // Flush is scheduled regularly. It collects a batch of transactions, sync
 // wal and do parallel committing. Each sync cost ~1s. Each checkpoint cost
 // ~10s. So it's necessary to sync for a batch of transactions, and to
@@ -274,25 +280,26 @@ void WalManager::Checkpoint(ForceCheckpointTask *ckp_task, TxnTimeStamp max_comm
 void WalManager::CheckpointInner(bool is_full_checkpoint, Txn *txn, TxnTimeStamp max_commit_ts, i64 wal_size) {
     DeferFn defer([&] { checkpoint_in_progress_.store(false); });
 
+    TxnTimeStamp last_ckp_ts = GetLastCkpTS();
     if (is_full_checkpoint) {
         if (max_commit_ts == last_full_ckp_ts_) {
             LOG_INFO(fmt::format("Skip full checkpoint because the max_commit_ts {} is the same as the last full checkpoint", max_commit_ts));
             return;
         }
-        if (last_full_ckp_ts_ != UNCOMMIT_TS && last_full_ckp_ts_ >= max_commit_ts) {
+        if (last_full_ckp_ts_ >= max_commit_ts) {
             UnrecoverableError(
                 fmt::format("WalManager::UpdateLastFullMaxCommitTS last_full_ckp_ts_ {} >= max_commit_ts {}", last_full_ckp_ts_, max_commit_ts));
         }
-        if (last_ckp_ts_ != UNCOMMIT_TS && last_ckp_ts_ > max_commit_ts) {
-            UnrecoverableError(fmt::format("WalManager::UpdateLastFullMaxCommitTS last_ckp_ts_ {} >= max_commit_ts {}", last_ckp_ts_, max_commit_ts));
+        if (last_ckp_ts > max_commit_ts) {
+            UnrecoverableError(fmt::format("WalManager::UpdateLastFullMaxCommitTS last_ckp_ts {} >= max_commit_ts {}", last_ckp_ts, max_commit_ts));
         }
     } else {
-        if (max_commit_ts == last_ckp_ts_) {
+        if (max_commit_ts == last_ckp_ts) {
             LOG_INFO(fmt::format("Skip delta checkpoint because the max_commit_ts {} is the same as the last checkpoint", max_commit_ts));
             return;
         }
-        if (last_ckp_ts_ >= max_commit_ts) {
-            UnrecoverableError(fmt::format("WalManager::UpdateLastMaxCommitTS last_ckp_ts_ {} >= max_commit_ts {}", last_ckp_ts_, max_commit_ts));
+        if (last_ckp_ts >= max_commit_ts) {
+            UnrecoverableError(fmt::format("WalManager::UpdateLastMaxCommitTS last_ckp_ts {} >= max_commit_ts {}", last_ckp_ts, max_commit_ts));
         }
     }
     try {

src/storage/wal/wal_manager.cppm

@@ -70,6 +70,8 @@ public:
 
     i64 GetLastCkpWalSize();
 
+    TxnTimeStamp GetLastCkpTS();
+
 private:
     // Checkpoint Helper
     void CheckpointInner(bool is_full_checkpoint, Txn *txn, TxnTimeStamp max_commit_ts, i64 wal_size);
@@ -123,7 +125,7 @@ private:
     i64 last_ckp_wal_size_{};
     Atomic<bool> checkpoint_in_progress_{false};
 
-    // Only Checkpoint thread access following members
+    // Only Checkpoint/Cleanup thread access following members
     TxnTimeStamp last_ckp_ts_{};
     TxnTimeStamp last_full_ckp_ts_{};
 };