Fix: use checkpoint_row_count to replace row_count (#1136)

### What problem does this PR solve?

`SegmentEntry` and `TableEntry` use `row_count_` directly when doing
full checkpoint, which will cause error if other txn modified
`row_count_` concurrently.


### Type of change

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

Author: Ognimalf

src/storage/meta/entry/segment_entry.cpp

@@ -409,6 +409,8 @@ SharedPtr<BlockEntry> SegmentEntry::GetBlockEntryByID(BlockID block_id) const {
 nlohmann::json SegmentEntry::Serialize(TxnTimeStamp max_commit_ts) {
     nlohmann::json json_res;
 
+    this->checkpoint_row_count_ = 0;
+
     // const field
     json_res["segment_dir"] = *this->segment_dir_;
     json_res["row_capacity"] = this->row_capacity_;
@@ -420,7 +422,6 @@ nlohmann::json SegmentEntry::Serialize(TxnTimeStamp max_commit_ts) {
         json_res["min_row_ts"] = this->min_row_ts_;
         json_res["max_row_ts"] = std::min(this->max_row_ts_, max_commit_ts);
         json_res["deleted"] = this->deleted_;
-        json_res["row_count"] = this->row_count_;
         json_res["actual_row_count"] = this->actual_row_count_;
 
         json_res["commit_ts"] = TxnTimeStamp(this->commit_ts_);
@@ -436,8 +437,11 @@ nlohmann::json SegmentEntry::Serialize(TxnTimeStamp max_commit_ts) {
             if (block_entry->commit_ts_ <= max_commit_ts) {
                 block_entry->Flush(max_commit_ts);
                 json_res["block_entries"].emplace_back(block_entry->Serialize(max_commit_ts));
+                this->checkpoint_row_count_ += block_entry->checkpoint_row_count();
             }
         }
+
+        json_res["row_count"] = this->checkpoint_row_count_;
     }
     return json_res;
 }
@@ -455,6 +459,7 @@ SharedPtr<SegmentEntry> SegmentEntry::Deserialize(const nlohmann::json &segment_
     segment_entry->max_row_ts_ = segment_entry_json["max_row_ts"];
     segment_entry->row_count_ = segment_entry_json["row_count"];
     segment_entry->actual_row_count_ = segment_entry_json["actual_row_count"];
+    segment_entry->checkpoint_row_count_ = 0;
 
     segment_entry->commit_ts_ = segment_entry_json["commit_ts"];
     segment_entry->begin_ts_ = segment_entry_json["begin_ts"];

src/storage/meta/entry/segment_entry.cppm

@@ -151,6 +151,11 @@ public:
         return actual_row_count_;
     }
 
+    // only used in Serialize(), FullCheckpoint, and no concurrency
+    SizeT checkpoint_row_count() const {
+        return checkpoint_row_count_;
+    }
+
     int Room() const { return this->row_capacity_ - this->row_count(); }
 
     TxnTimeStamp min_row_ts() const { return min_row_ts_; }
@@ -207,6 +212,7 @@ private:
 
     SizeT row_count_{};
     SizeT actual_row_count_{}; // not deleted row count
+    SizeT checkpoint_row_count_{};
 
     TxnTimeStamp min_row_ts_{UNCOMMIT_TS}; // Indicate the commit_ts which create this SegmentEntry
     TxnTimeStamp max_row_ts_{0};

src/storage/meta/entry/table_entry.cpp

@@ -313,12 +313,10 @@ void TableEntry::Import(SharedPtr<SegmentEntry> segment_entry, Txn *txn) {
     {
         std::unique_lock lock(this->rw_locker_);
         SegmentID segment_id = segment_entry->segment_id();
-        SizeT row_count = segment_entry->row_count();
         auto [_, insert_ok] = this->segment_map_.emplace(segment_id, segment_entry);
         if (!insert_ok) {
             UnrecoverableError(fmt::format("Insert segment {} failed.", segment_id));
         }
-        this->row_count_ += row_count;
     }
     // Populate index entirely for the segment
     TxnTableStore *txn_table_store = txn->GetTxnTableStore(this);
@@ -369,6 +367,12 @@ void TableEntry::AppendData(TransactionID txn_id, void *txn_store, TxnTimeStamp
     AppendState *append_state_ptr = txn_store_ptr->append_state_.get();
     Txn *txn = txn_store_ptr->txn_;
     if (append_state_ptr->Finished()) {
+        // Import update row count
+        if (append_state_ptr->blocks_.empty()) {
+            for (auto &segment : txn_store_ptr->flushed_segments()) {
+                this->row_count_ += segment->row_count();
+            }
+        }
         LOG_TRACE("No append is done.");
         return;
     }
@@ -904,11 +908,11 @@ nlohmann::json TableEntry::Serialize(TxnTimeStamp max_commit_ts) {
     Vector<SegmentEntry *> segment_candidates;
     Vector<TableIndexMeta *> table_index_meta_candidates;
     Vector<String> table_index_name_candidates;
+    SizeT checkpoint_row_count = 0;
     {
         std::shared_lock<std::shared_mutex> lck(this->rw_locker_);
         json_res["table_name"] = *this->GetTableName();
         json_res["table_entry_type"] = this->table_entry_type_;
-        json_res["row_count"] = this->row_count_.load();
         json_res["begin_ts"] = this->begin_ts_;
         json_res["commit_ts"] = this->commit_ts_.load();
         json_res["txn_id"] = this->txn_id_.load();
@@ -952,7 +956,9 @@ nlohmann::json TableEntry::Serialize(TxnTimeStamp max_commit_ts) {
     // Serialize segments
     for (const auto &segment_entry : segment_candidates) {
         json_res["segments"].emplace_back(segment_entry->Serialize(max_commit_ts));
+        checkpoint_row_count += segment_entry->checkpoint_row_count();
     }
+    json_res["row_count"] = checkpoint_row_count;
     json_res["unsealed_id"] = unsealed_id_;
 
     // Serialize indexes

src/storage/txn/txn_store.cppm

@@ -134,6 +134,8 @@ public: // Getter
 
     const HashMap<SegmentID, TxnSegmentStore> &txn_segments() const { return txn_segments_store_; }
 
+    const Vector<SegmentEntry *> &flushed_segments() const { return flushed_segments_; }
+
 private:
     HashMap<SegmentID, TxnSegmentStore> txn_segments_store_{};
     Vector<SegmentEntry *> flushed_segments_{};