Refactor transaction commit (#2468)

### What problem does this PR solve? - Add txn text in query context - Update log ### Type of change - [x] Refactoring --------- Signed-off-by: Jin Hai <[email protected]>
infiniflow · Jan 17, 2025 · 197a243 · 197a243
1 parent c332e77
commit 197a243
Show file tree

Hide file tree

Showing 6 changed files with 20 additions and 11 deletions.
diff --git a/src/main/query_context.cpp b/src/main/query_context.cpp
@@ -422,13 +422,16 @@ QueryResult QueryContext::HandleAdminStatement(const AdminStatement *admin_state
 
 void QueryContext::BeginTxn(const BaseStatement *base_statement) {
     if (session_ptr_->GetTxn() == nullptr) {
-        bool is_checkpoint = base_statement != nullptr && base_statement->type_ == StatementType::kFlush;
         Txn *new_txn = nullptr;
-        // TODO: more type check and setting
-        if (is_checkpoint) {
-            new_txn = storage_->txn_manager()->BeginTxn(nullptr, TransactionType::kCheckpoint);
+        if(base_statement == nullptr) {
+            new_txn = storage_->txn_manager()->BeginTxn(MakeUnique<String>(""), TransactionType::kNormal);
         } else {
-            new_txn = storage_->txn_manager()->BeginTxn(nullptr, TransactionType::kNormal);
+            // TODO: more type check and setting
+            if (base_statement->type_ == StatementType::kFlush) {
+                new_txn = storage_->txn_manager()->BeginTxn(MakeUnique<String>(base_statement->ToString()), TransactionType::kCheckpoint);
+            } else {
+                new_txn = storage_->txn_manager()->BeginTxn(MakeUnique<String>(base_statement->ToString()), TransactionType::kNormal);
+            }
         }
         session_ptr_->SetTxn(new_txn);
     }

diff --git a/src/main/query_context.cppm b/src/main/query_context.cppm
@@ -90,7 +90,7 @@ public:
 
     inline u64 GetNextNodeID() { return ++current_max_node_id_; }
 
-    void BeginTxn(const BaseStatement *statement = nullptr);
+    void BeginTxn(const BaseStatement *statement);
 
     TxnTimeStamp CommitTxn();
 

diff --git a/src/storage/txn/txn.cpp b/src/storage/txn/txn.cpp
@@ -540,6 +540,11 @@ TxnState Txn::GetTxnState() const {
     return txn_context_ptr_->state_;
 }
 
+TransactionType Txn::GetTxnType() const {
+    std::shared_lock<std::shared_mutex> r_locker(rw_locker_);
+    return txn_context_ptr_->txn_type_;
+}
+
 bool Txn::IsWriteTransaction() const { return txn_context_ptr_->is_write_transaction_; }
 
 void Txn::SetTxnRollbacking(TxnTimeStamp rollback_ts) {

diff --git a/src/storage/txn/txn.cppm b/src/storage/txn/txn.cppm
@@ -206,6 +206,8 @@ public:
 
     TxnState GetTxnState() const;
 
+    TransactionType GetTxnType() const;
+
     bool IsWriteTransaction() const;
 
     void SetTxnCommitted();

diff --git a/src/storage/txn/txn_manager.cpp b/src/storage/txn/txn_manager.cpp
@@ -68,8 +68,7 @@ Txn *TxnManager::BeginTxn(UniquePtr<String> txn_text, TransactionType txn_type)
     // Record the start ts of the txn
     TxnTimeStamp begin_ts = current_ts_ + 1;
 
-    bool ckp_txn = txn_type == TransactionType::kCheckpoint;
-    if (ckp_txn) {
+    if (txn_type == TransactionType::kCheckpoint) {
         if (ckp_begin_ts_ != UNCOMMIT_TS) {
             // not set ckp_begin_ts_ may not truncate the wal file.
             LOG_WARN(fmt::format("Another checkpoint txn is started in {}, new checkpoint {} will do nothing", ckp_begin_ts_, begin_ts));
@@ -395,7 +394,7 @@ void TxnManager::CleanupTxn(Txn *txn, bool commit) {
 bool TxnManager::InCheckpointProcess(TxnTimeStamp commit_ts) {
     std::lock_guard guard(locker_);
     if (commit_ts > ckp_begin_ts_) {
-        LOG_TRACE(fmt::format("Full checkpoint begin in {}, cur txn commit_ts: {}, swap to new wal file", ckp_begin_ts_, commit_ts));
+        LOG_TRACE(fmt::format("Full/Delta checkpoint begin at {}, cur txn commit_ts: {}, swap to new wal file", ckp_begin_ts_, commit_ts));
         ckp_begin_ts_ = UNCOMMIT_TS;
         return true;
     }

diff --git a/src/unit_test/test_helper/sql_runner.cpp b/src/unit_test/test_helper/sql_runner.cpp
@@ -80,14 +80,14 @@ SharedPtr<DataTable> SQLRunner::Run(const String &sql_text, bool print) {
         UnrecoverableError(parsed_result->error_message_);
     }
 
-    query_context_ptr->BeginTxn();
-
     //    LogicalPlanner logical_planner(query_context_ptr.get());
     //    Optimizer optimizer(query_context_ptr.get());
     //    PhysicalPlanner physical_planner(query_context_ptr.get());
     //    FragmentBuilder fragment_builder(query_context_ptr.get());
     BaseStatement *statement = (*parsed_result->statements_ptr_)[0];
 
+    query_context_ptr->BeginTxn(statement);
+
     SharedPtr<BindContext> bind_context;
     query_context_ptr->logical_planner()->Build(statement, bind_context);
     query_context_ptr->set_max_node_id(bind_context->GetNewLogicalNodeId());