refactor: skip no-op edge CSR compaction after COPY

include/neug/storages/graph/edge_table.h

@@ -180,6 +180,11 @@ class NEUG_API EdgeTable {
                           int32_t ie_offset, int32_t col_id,
                           const Value& new_prop, timestamp_t ts);
 
+  bool NeedsCompaction(
+      const std::optional<std::string>& sort_key_for_nbr) const noexcept {
+    return sort_key_for_nbr.has_value() || needs_csr_normalization_.load();
+  }
+
   void Compact(const std::optional<std::string>& sort_key_for_nbr);
 
   size_t PropTableSize() const;
@@ -208,6 +213,13 @@ class NEUG_API EdgeTable {
   std::unique_ptr<Table> table_;
   std::atomic<uint64_t> table_idx_{0};
   std::atomic<uint64_t> capacity_{0};
+  // DirtyTracker answers whether this table must be checkpointed; this bit
+  // answers whether its CSR contents need an O(E) normalization scan. The
+  // states are independent: timestamp-zero COPY is dirty but normalized, while
+  // an incremental checkpoint clears dirtiness without removing MVCC
+  // timestamps or tombstones. Keep this state EdgeTable-local and persist it
+  // across reopen so COPY can skip redundant compaction safely.
+  std::atomic<bool> needs_csr_normalization_{false};
 
   friend class PropertyGraph;
   friend class EdgeTableView;

include/neug/storages/graph/graph_view.h

@@ -14,6 +14,7 @@
  */
 #pragma once
 
+#include <atomic>
 #include <cassert>
 #include <memory>
 #include <string>
@@ -105,6 +106,7 @@ class EdgeTableView {
   CsrBase* out_csr_{nullptr};
   CsrBase* in_csr_{nullptr};
   std::atomic<uint64_t>* table_idx_{nullptr};
+  std::atomic<bool>* needs_csr_normalization_{nullptr};
 
   TableView view_;
 };

src/main/checkpoint_coordinator.cc

@@ -151,7 +151,8 @@ Status CheckpointCoordinator::CommitCowWrite(
 
     // Finalize only persistent COPY targets before checkpoint consumption,
     // while ordinary rollback remains safe. Vertex COPY has a timestamp-zero
-    // tail; edge COPY needs compaction only when it has a neighbor sort key.
+    // tail. An edge target needs compaction when it has a neighbor sort key or
+    // inherited ordinary writes left timestamps/tombstones to normalize.
     // Keeping the target sets transaction-local avoids compacting unrelated
     // dirty tables inherited by the private COW graph.
     workspace.FinalizeBulkTablesForCheckpoint();

src/storages/graph/edge_table.cc

@@ -473,6 +473,7 @@ EdgeTable::EdgeTable(EdgeTable&& edge_table)
   table_ = std::move(edge_table.table_);
   table_idx_ = edge_table.table_idx_.load();
   capacity_ = edge_table.capacity_.load();
+  needs_csr_normalization_ = edge_table.needs_csr_normalization_.load();
 }
 
 EdgeTable& EdgeTable::operator=(EdgeTable&& other) noexcept {
@@ -485,6 +486,7 @@ EdgeTable& EdgeTable::operator=(EdgeTable&& other) noexcept {
     table_ = std::move(other.table_);
     table_idx_ = other.table_idx_.load();
     capacity_ = other.capacity_.load();
+    needs_csr_normalization_ = other.needs_csr_normalization_.load();
   }
   return *this;
 }
@@ -502,6 +504,9 @@ void EdgeTable::Swap(EdgeTable& edge_table) {
   auto cap = capacity_.load();
   capacity_.store(edge_table.capacity_.load());
   edge_table.capacity_.store(cap);
+  auto needs_normalization = needs_csr_normalization_.load();
+  needs_csr_normalization_.store(edge_table.needs_csr_normalization_.load());
+  edge_table.needs_csr_normalization_.store(needs_normalization);
 }
 
 EdgeTable EdgeTable::Clone() const {
@@ -519,6 +524,7 @@ EdgeTable EdgeTable::Clone() const {
 
   cow_clone.table_idx_ = table_idx_.load();
   cow_clone.capacity_ = capacity_.load();
+  cow_clone.needs_csr_normalization_ = needs_csr_normalization_.load();
   return cow_clone;
 }
 
@@ -558,25 +564,35 @@ void EdgeTable::SortByEdgeData(timestamp_t ts) {
 
 void EdgeTable::BatchDeleteVertices(const std::set<vid_t>& src_set,
                                     const std::set<vid_t>& dst_set) {
+  if (!src_set.empty() || !dst_set.empty()) {
+    needs_csr_normalization_.store(true);
+  }
   out_csr_->batch_delete_vertices(src_set, dst_set);
   in_csr_->batch_delete_vertices(dst_set, src_set);
 }
 
 void EdgeTable::BatchDeleteEdges(const std::vector<vid_t>& src_list,
                                  const std::vector<vid_t>& dst_list) {
+  if (!src_list.empty() || !dst_list.empty()) {
+    needs_csr_normalization_.store(true);
+  }
   out_csr_->batch_delete_edges(src_list, dst_list);
   in_csr_->batch_delete_edges(dst_list, src_list);
 }
 
 void EdgeTable::BatchDeleteEdges(
     const std::vector<std::pair<vid_t, int32_t>>& oe_edges,
     const std::vector<std::pair<vid_t, int32_t>>& ie_edges) {
+  if (!oe_edges.empty() || !ie_edges.empty()) {
+    needs_csr_normalization_.store(true);
+  }
   out_csr_->batch_delete_edges(oe_edges);
   in_csr_->batch_delete_edges(ie_edges);
 }
 
 void EdgeTable::DeleteEdge(vid_t src_lid, vid_t dst_lid, int32_t oe_offset,
                            int32_t ie_offset, timestamp_t ts) {
+  needs_csr_normalization_.store(true);
   out_csr_->delete_edge(src_lid, oe_offset, ts);
   in_csr_->delete_edge(dst_lid, ie_offset, ts);
 }
@@ -653,6 +669,9 @@ void EdgeTable::UpdateEdgeProperty(vid_t src_lid, vid_t dst_lid,
                                    int32_t oe_offset, int32_t ie_offset,
                                    int32_t col_id, const Value& prop,
                                    timestamp_t ts) {
+  if (ts != 0) {
+    needs_csr_normalization_.store(true);
+  }
   auto accessor = get_edge_data_accessor(col_id);
   auto oe_edges = out_csr_->get_generic_view(ts).get_edges(src_lid);
   auto oe_iter = oe_edges.begin();
@@ -813,6 +832,9 @@ void EdgeTable::DeleteProperties(Checkpoint& ckp,
 std::pair<int32_t, const void*> EdgeTable::AddEdge(
     vid_t src_lid, vid_t dst_lid, const std::vector<Value>& edge_data,
     timestamp_t ts, Allocator& alloc, bool insert_safe) {
+  if (ts != 0) {
+    needs_csr_normalization_.store(true);
+  }
   return internal::insert_edge_into_csr_internal(
       *out_csr_, *in_csr_, *table_.get(), table_idx_, *meta_, src_lid, dst_lid,
       edge_data, ts, alloc, insert_safe);
@@ -999,6 +1021,7 @@ void EdgeTable::Compact(const std::optional<std::string>& sort_key_for_nbr) {
     out_csr_->batch_sort_by_edge_data(1);
     in_csr_->batch_sort_by_edge_data(1);
   }
+  needs_csr_normalization_.store(false);
 }
 
 size_t EdgeTable::PropTableSize() const {
@@ -1224,6 +1247,10 @@ EdgeTable EdgeTable::OpenFrom(std::shared_ptr<Checkpoint> ckp,
   et.SetCapacity(
       meta.GetScalarAs<uint64_t>(ScalarKey(src, edge, dst, "capacity"))
           .value_or(0));
+  et.needs_csr_normalization_.store(
+      meta.GetScalarAs<uint64_t>(
+              ScalarKey(src, edge, dst, "needs_csr_normalization"))
+          .value_or(meta.base_timestamp() == 0 ? 0 : 1) != 0);
   return et;
 }
 
@@ -1249,6 +1276,8 @@ void EdgeTable::DisassembleTo(ModuleBroker& store, CheckpointManifest& meta,
   }
   meta.SetScalar(ScalarKey(src, edge, dst, "capacity"),
                  std::to_string(GetCapacity()));
+  meta.SetScalar(ScalarKey(src, edge, dst, "needs_csr_normalization"),
+                 needs_csr_normalization_.load() ? "1" : "0");
 }
 
 void EdgeTable::ReuseCheckpointModules(Checkpoint& ckp,
@@ -1271,6 +1300,8 @@ void EdgeTable::ReuseCheckpointModules(Checkpoint& ckp,
     meta.CopyScalarFrom(prev, ScalarKey(src, edge, dst, "table_idx"));
   }
   meta.CopyScalarFrom(prev, ScalarKey(src, edge, dst, "capacity"));
+  meta.CopyScalarFrom(prev,
+                      ScalarKey(src, edge, dst, "needs_csr_normalization"));
 }
 
 }  // namespace neug

src/storages/graph/graph_view.cc

@@ -145,6 +145,7 @@ EdgeTableView::EdgeTableView(EdgeTable& table)
       out_csr_(table.out_csr_.get()),
       in_csr_(table.in_csr_.get()),
       table_idx_(&table.table_idx_),
+      needs_csr_normalization_(&table.needs_csr_normalization_),
       view_(*table.table()) {}
 
 CsrView EdgeTableView::GetOutgoingView(timestamp_t ts) const {
@@ -200,6 +201,9 @@ EdgeDataAccessor EdgeTableView::GetDataAccessor(
 std::pair<int32_t, const void*> EdgeTableView::AddEdge(
     vid_t src_lid, vid_t dst_lid, const std::vector<Value>& properties,
     timestamp_t ts, Allocator& alloc, bool insert_safe) {
+  if (ts != 0) {
+    needs_csr_normalization_->store(true);
+  }
   return internal::insert_edge_into_csr_internal(
       *out_csr_, *in_csr_, view_, *table_idx_, *meta_, src_lid, dst_lid,
       properties, ts, alloc, insert_safe);

src/storages/graph/property_graph.cc

@@ -1024,7 +1024,10 @@ void PropertyGraph::Compact() {
         }
         const auto& sort_key_for_nbr =
             schema_.get_sort_key_for_nbr(src_label_i, dst_label_i, e_label_i);
-        edge_tables_.at(index).Compact(sort_key_for_nbr);
+        auto& edge_table = edge_tables_.at(index);
+        if (edge_table.NeedsCompaction(sort_key_for_nbr)) {
+          edge_table.Compact(sort_key_for_nbr);
+        }
       }
     }
   }

src/transaction/cow_graph_workspace.cc

@@ -58,7 +58,11 @@ void CowGraphWorkspace::FinalizeBulkTablesForCheckpoint() {
         graph.schema().parse_edge_label(edge_triplet_id);
     const auto& sort_key =
         graph.schema().get_sort_key_for_nbr(src_label, dst_label, edge_label);
-    graph.get_edge_table_by_index(edge_triplet_id).Compact(sort_key);
+    auto& edge_table = graph.get_edge_table_by_index(edge_triplet_id);
+    if (!edge_table.NeedsCompaction(sort_key)) {
+      continue;
+    }
+    edge_table.Compact(sort_key);
   }
 }