Fix: error txn_store pointer when replay wal command. (infiniflow#804)

### What problem does this PR solve?

The replay txn will set another `txn_store` when replay wal command,
which will make block report error:

```
terminate called after throwing an instance of 'infinity::UnrecoverableException@infinity_exception'
  what():  Multiple transactions are changing data of Segment: 0, Block: 0@src/storage/meta/entry/block_entry.cpp:165
Signal: SIGABRT (signal SIGABRT)
```

It is because the `txn` before is not cleared.

Issue link:infiniflow#802

### Type of change

- [x] Bug Fix (non-breaking change which fixes an issue)

Author: Ognimalf

src/executor/operator/physical_import.cpp

@@ -198,7 +198,7 @@ void PhysicalImport::ImportCSV(QueryContext *query_context, ImportOperatorState
     UniquePtr<ZxvParserCtx> parser_context = nullptr;
     Txn *txn = query_context->GetTxn();
     {
-        auto *buffer_mgr = txn->GetBufferMgr();
+        auto *buffer_mgr = txn->buffer_manager();
         u64 segment_id = Catalog::GetNextSegmentID(table_entry_);
         SharedPtr<SegmentEntry> segment_entry = SegmentEntry::NewImportSegmentEntry(table_entry_, segment_id, txn);
         UniquePtr<BlockEntry> block_entry = BlockEntry::NewBlockEntry(segment_entry.get(), 0, 0, table_entry_->ColumnCount(), txn);
@@ -278,7 +278,7 @@ void PhysicalImport::ImportJSONL(QueryContext *query_context, ImportOperatorStat
     Vector<ColumnVector> column_vectors;
     for (SizeT i = 0; i < table_entry_->ColumnCount(); ++i) {
         auto *block_column_entry = block_entry->GetColumnBlockEntry(i);
-        column_vectors.emplace_back(block_column_entry->GetColumnVector(txn->GetBufferMgr()));
+        column_vectors.emplace_back(block_column_entry->GetColumnVector(txn->buffer_manager()));
     }
     while (true) {
         SizeT end_pos = jsonl_str.find('\n', start_pos);
@@ -312,7 +312,7 @@ void PhysicalImport::ImportJSONL(QueryContext *query_context, ImportOperatorStat
             column_vectors.clear();
             for (SizeT i = 0; i < table_entry_->ColumnCount(); ++i) {
                 auto *block_column_entry = block_entry->GetColumnBlockEntry(i);
-                column_vectors.emplace_back(block_column_entry->GetColumnVector(txn->GetBufferMgr()));
+                column_vectors.emplace_back(block_column_entry->GetColumnVector(txn->buffer_manager()));
             }
         }
     }
@@ -361,7 +361,7 @@ void PhysicalImport::CSVRowHandler(void *context) {
     SizeT column_count = parser_context->parser_.CellCount();
 
     auto *txn = parser_context->txn_;
-    auto *buffer_mgr = txn->GetBufferMgr();
+    auto *buffer_mgr = txn->buffer_manager();
 
     auto segment_entry = parser_context->segment_entry_;
     UniquePtr<BlockEntry> block_entry = std::move(parser_context->block_entry_);

src/storage/meta/entry/block_entry.cpp

@@ -162,8 +162,11 @@ u16 BlockEntry::AppendData(TransactionID txn_id,
                            BufferManager *buffer_mgr) {
     std::unique_lock<std::shared_mutex> lck(this->rw_locker_);
     if (this->using_txn_id_ != 0 && this->using_txn_id_ != txn_id) {
-        UnrecoverableError(
-            fmt::format("Multiple transactions are changing data of Segment: {}, Block: {}", this->segment_entry_->segment_id(), this->block_id_));
+        UnrecoverableError(fmt::format("Multiple transactions are changing data of Segment: {}, Block: {}, using_txn_id_: {}, txn_id: {}",
+                                       this->segment_entry_->segment_id(),
+                                       this->block_id_,
+                                       this->using_txn_id_,
+                                       txn_id));
     }
 
     this->using_txn_id_ = txn_id;

src/storage/meta/entry/segment_index_entry.cpp

@@ -62,7 +62,7 @@ SegmentIndexEntry::SegmentIndexEntry(TableIndexEntry *table_index_entry, Segment
 
 SharedPtr<SegmentIndexEntry>
 SegmentIndexEntry::NewIndexEntry(TableIndexEntry *table_index_entry, SegmentID segment_id, Txn *txn, CreateIndexParam *param) {
-    auto *buffer_mgr = txn->GetBufferMgr();
+    auto *buffer_mgr = txn->buffer_manager();
 
     // FIXME: estimate index size.
     auto vector_file_worker = table_index_entry->CreateFileWorker(param, segment_id);
@@ -138,7 +138,7 @@ Status SegmentIndexEntry::CreateIndexPrepare(const IndexBase *index_base,
                                              bool check_ts) {
     TxnTimeStamp begin_ts = txn->BeginTS();
 
-    auto *buffer_mgr = txn->GetBufferMgr();
+    auto *buffer_mgr = txn->buffer_manager();
     switch (index_base->index_type_) {
         case IndexType::kIVFFlat: {
             if (column_def->type()->type() != LogicalType::kEmbedding) {

src/storage/meta/entry/table_index_entry.cpp

@@ -245,7 +245,7 @@ Tuple<FulltextIndexEntry *, Vector<SegmentIndexEntry *>, Status>
 TableIndexEntry::CreateIndexPrepare(TableEntry *table_entry, BlockIndex *block_index, Txn *txn, bool prepare, bool is_replay, bool check_ts) {
     FulltextIndexEntry *fulltext_index_entry = this->fulltext_index_entry_.get();
     if (fulltext_index_entry != nullptr && !IsFulltextIndexHomebrewed()) {
-        auto *buffer_mgr = txn->GetBufferMgr();
+        auto *buffer_mgr = txn->buffer_manager();
         for (const auto *segment_entry : block_index->segments_) {
             fulltext_index_entry->irs_index_->BatchInsert(table_entry, index_base_.get(), segment_entry, buffer_mgr);
         }

src/storage/txn/txn.cpp

@@ -171,8 +171,6 @@ void Txn::CheckTxn(const String &db_name) {
     }
 }
 
-BufferManager *Txn::GetBufferMgr() const { return this->txn_mgr_->GetBufferMgr(); }
-
 // Database OPs
 Status Txn::CreateDatabase(const String &db_name, ConflictType conflict_type) {
     this->CheckTxnStatus();

src/storage/txn/txn.cppm

@@ -146,8 +146,6 @@ public:
     Compact(TableEntry *table_entry, Vector<Pair<SharedPtr<SegmentEntry>, Vector<SegmentEntry *>>> &&segment_data, CompactSegmentsTaskType type);
 
     // Getter
-    BufferManager *GetBufferMgr() const;
-
     BufferManager *buffer_manager() const { return buffer_mgr_; }
 
     Catalog *GetCatalog() { return catalog_; }

src/storage/txn/txn_store.cpp

@@ -206,7 +206,7 @@ void TxnTableStore::PrepareCommit() {
 
     // Start to append
     LOG_TRACE(fmt::format("Transaction local storage table: {}, Start to prepare commit", *table_entry_->GetTableName()));
-    Catalog::Append(table_entry_, txn_->TxnID(), this, txn_->CommitTS(), txn_->GetBufferMgr());
+    Catalog::Append(table_entry_, txn_->TxnID(), this, txn_->CommitTS(), txn_->buffer_manager());
 
     // Attention: "compact" needs to be ahead of "delete"
     if (compact_state_.task_type_ != CompactSegmentsTaskType::kInvalid) {

src/storage/wal/wal_manager.cpp

@@ -742,10 +742,10 @@ void WalManager::WalCmdDeleteReplay(const WalCmdDelete &cmd, TransactionID txn_i
     }
 
     auto fake_txn = Txn::NewReplayTxn(storage_->buffer_manager(), storage_->txn_manager(), storage_->catalog(), txn_id);
-    auto table_store = MakeShared<TxnTableStore>(table_entry, fake_txn.get());
+    auto table_store = fake_txn->GetTxnTableStore(table_entry);
     table_store->Delete(cmd.row_ids_);
     fake_txn->FakeCommit(commit_ts);
-    Catalog::Delete(table_store->table_entry_, fake_txn->TxnID(), (void *)table_store.get(), fake_txn->CommitTS(), table_store->delete_state_);
+    Catalog::Delete(table_store->table_entry_, fake_txn->TxnID(), (void *)table_store, fake_txn->CommitTS(), table_store->delete_state_);
     Catalog::CommitWrite(table_store->table_entry_, fake_txn->TxnID(), commit_ts, table_store->txn_segments());
 }
 
@@ -777,15 +777,14 @@ void WalManager::WalCmdAppendReplay(const WalCmdAppend &cmd, TransactionID txn_i
     }
 
     auto fake_txn = Txn::NewReplayTxn(storage_->buffer_manager(), storage_->txn_manager(), storage_->catalog(), txn_id);
-
-    auto table_store = MakeShared<TxnTableStore>(table_entry, fake_txn.get());
+    auto table_store = fake_txn->GetTxnTableStore(table_entry);
     table_store->Append(cmd.block_);
 
     auto append_state = MakeUnique<AppendState>(table_store->blocks_);
     table_store->append_state_ = std::move(append_state);
 
     fake_txn->FakeCommit(commit_ts);
-    Catalog::Append(table_store->table_entry_, fake_txn->TxnID(), table_store.get(), commit_ts, storage_->buffer_manager());
+    Catalog::Append(table_store->table_entry_, fake_txn->TxnID(), table_store, commit_ts, storage_->buffer_manager());
     Catalog::CommitWrite(table_store->table_entry_, fake_txn->TxnID(), commit_ts, table_store->txn_segments());
 }