Performace: move BaseEntry::txn_id_ to non-atomic variable.

Performace: move BaseEntry::CheckVisible inline.

src/storage/meta/entry/base_entry.cppm

@@ -21,11 +21,14 @@ export module base_entry;
 import stl;
 import default_values;
 import txn;
+import txn_manager;
+import infinity_exception;
+import third_party;
+import txn_state;
 
 namespace infinity {
 
 class Catalog;
-class TxnManager;
 
 export enum class EntryType : i8 {
     kDatabase,
@@ -66,10 +69,26 @@ public:
 
     SharedPtr<String> encode_ptr() const { return encode_; }
 
-    virtual bool CheckVisible(Txn *txn) const;
+    // return if this entry is visible to the `txn`
+    virtual bool CheckVisible(Txn *txn) const {
+        TxnTimeStamp begin_ts = txn->BeginTS();
+        if (begin_ts >= commit_ts_ || txn_id_ == txn->TxnID()) { 
+            return true;
+        }
+        TxnManager *txn_mgr = txn->txn_mgr();
+        if (txn_mgr == nullptr) { // when replay
+            UnrecoverableError(fmt::format("Replay should not reach here. begin_ts: {}, commit_ts_: {} txn_id: {}, txn_id_: {}",
+                                           begin_ts,
+                                           commit_ts_,
+                                           txn->TxnID(),
+                                           txn_id_));
+        }
+        // Check if the entry is in committing process, because commit_ts of the base_entry is set in the Txn::CommitBottom
+        return txn_mgr->CheckIfCommitting(txn_id_, begin_ts);
+    }
 
 public:
-    atomic_u64 txn_id_{0};
+    TransactionID txn_id_{0};
     TxnTimeStamp begin_ts_{0};
     atomic_u64 commit_ts_{UNCOMMIT_TS};
     const bool deleted_;

src/storage/meta/entry/db_entry.cpp

@@ -55,7 +55,7 @@ DBEntry::DBEntry(DBMeta *db_meta,
                  TxnTimeStamp begin_ts)
     : BaseEntry(EntryType::kDatabase, is_delete, DBEntry::EncodeIndex(*db_name)), db_meta_(db_meta), db_entry_dir_(db_entry_dir), db_name_(db_name) {
     begin_ts_ = begin_ts;
-    txn_id_.store(txn_id);
+    txn_id_ = txn_id;
 }
 
 SharedPtr<DBEntry> DBEntry::NewDBEntry(DBMeta *db_meta,
@@ -238,7 +238,7 @@ nlohmann::json DBEntry::Serialize(TxnTimeStamp max_commit_ts) {
     {
         std::shared_lock<std::shared_mutex> lck(this->rw_locker());
         json_res["db_name"] = *this->db_name_;
-        json_res["txn_id"] = this->txn_id_.load();
+        json_res["txn_id"] = this->txn_id_;
         json_res["begin_ts"] = this->begin_ts_;
         json_res["commit_ts"] = this->commit_ts_.load();
         json_res["deleted"] = this->deleted_;

src/storage/meta/entry/table_entry.cpp

@@ -208,7 +208,7 @@ void TableEntry::RemoveIndexEntry(const String &index_name, TransactionID txn_id
 
 /// replay
 void TableEntry::UpdateEntryReplay(const SharedPtr<TableEntry> &table_entry) {
-    txn_id_.store(table_entry->txn_id_);
+    txn_id_ = table_entry->txn_id_;
     begin_ts_ = table_entry->begin_ts_;
     commit_ts_.store(table_entry->commit_ts_);
     row_count_ = table_entry->row_count();
@@ -944,7 +944,7 @@ nlohmann::json TableEntry::Serialize(TxnTimeStamp max_commit_ts) {
         json_res["table_entry_type"] = this->table_entry_type_;
         json_res["begin_ts"] = this->begin_ts_;
         json_res["commit_ts"] = this->commit_ts_.load();
-        json_res["txn_id"] = this->txn_id_.load();
+        json_res["txn_id"] = this->txn_id_;
         json_res["deleted"] = this->deleted_;
         if (!this->deleted_) {
             json_res["table_entry_dir"] = *this->table_entry_dir_;
@@ -1180,8 +1180,6 @@ void TableEntry::Cleanup() {
     LOG_TRACE(fmt::format("Cleaned dir: {}", *table_entry_dir_));
 }
 
-IndexReader TableEntry::GetFullTextIndexReader(Txn *txn) {
-    return fulltext_column_index_cache_.GetIndexReader(txn, this);
-}
+IndexReader TableEntry::GetFullTextIndexReader(Txn *txn) { return fulltext_column_index_cache_.GetIndexReader(txn, this); }
 
 } // namespace infinity

src/storage/meta/entry/table_index_entry.cpp

@@ -191,7 +191,7 @@ nlohmann::json TableIndexEntry::Serialize(TxnTimeStamp max_commit_ts) {
     Vector<SharedPtr<SegmentIndexEntry>> segment_index_entry_candidates;
     {
         std::shared_lock<std::shared_mutex> lck(this->rw_locker_);
-        json["txn_id"] = this->txn_id_.load();
+        json["txn_id"] = this->txn_id_;
         json["begin_ts"] = this->begin_ts_;
         json["commit_ts"] = this->commit_ts_.load();
         json["deleted"] = this->deleted_;

src/storage/txn/txn_manager.cpp

@@ -97,7 +97,7 @@ bool TxnManager::CheckIfCommitting(TransactionID txn_id, TxnTimeStamp begin_ts)
     std::lock_guard guard(rw_locker_);
     auto iter = txn_map_.find(txn_id);
     if (iter == txn_map_.end()) {
-        return true; // committed
+        return true; // Txn is already committed
     }
     Txn *txn = iter->second.get();
     auto state = txn->GetTxnState();