feat(graph): record rel change epochs alongside materialized CSR

src/function/table/project_native_graph.cpp

@@ -1,5 +1,7 @@
 #include <algorithm>
 
+#include "catalog/catalog.h"
+#include "catalog/catalog_entry/rel_group_catalog_entry.h"
 #include "common/exception/binder.h"
 #include "common/types/value/nested.h"
 #include "function/gds/gds.h"
@@ -12,6 +14,8 @@
 #include "main/query_result/arrow_query_result.h"
 #include "parser/parser.h"
 #include "processor/execution_context.h"
+#include "storage/storage_manager.h"
+#include "storage/table/table.h"
 #include "transaction/transaction_context.h"
 #include <format>
 
@@ -64,16 +68,32 @@ static void materializeRelCsr(ParsedNativeGraphEntry& entry, main::ClientContext
     const auto& nodeTable = entry.nodeInfos[0].tableName;
     main::Connection conn{context->getDatabase()};
     entry.relCsrResults.reserve(entry.relInfos.size());
+    entry.relCsrEpochs.reserve(entry.relInfos.size());
     for (const auto& relInfo : entry.relInfos) {
+        // Capture the rel table's change epoch BEFORE the scan: a mutation racing the
+        // materialization bumps the epoch, so consumers see a mismatch and fall back.
+        uint64_t epoch = 0;
+        const auto* relEntry = catalog::Catalog::Get(*context)->getTableCatalogEntry(
+            transaction::Transaction::Get(*context), relInfo.tableName);
+        const auto& relGroup = relEntry->constCast<catalog::RelGroupCatalogEntry>();
+        if (!relGroup.getRelEntryInfos().empty()) {
+            const auto* relTable = storage::StorageManager::Get(*context)->getTable(
+                relGroup.getRelEntryInfos()[0].oid);
+            if (relTable != nullptr) {
+                epoch = relTable->getChangeEpoch();
+            }
+        }
         auto query = std::format("MATCH (a:`{}`)-[r:`{}`]->(b:`{}`) RETURN a.rowid, b.rowid",
             nodeTable, relInfo.tableName, nodeTable);
         auto result = conn.queryAsArrow(query, ARROW_CHUNK_SIZE);
         auto* arrowResult = dynamic_cast<main::ArrowQueryResult*>(result.get());
         if (arrowResult != nullptr && arrowResult->isSuccess() && arrowResult->hasCSRMetadata()) {
             entry.relCsrResults.push_back(std::shared_ptr<main::QueryResult>{std::move(result)});
+            entry.relCsrEpochs.push_back(epoch);
         } else {
             // Shape not tracked (or scan failed): this rel stays unmaterialized.
             entry.relCsrResults.push_back(nullptr);
+            entry.relCsrEpochs.push_back(0);
         }
     }
 }

src/include/graph/parsed_graph_entry.h

@@ -51,6 +51,12 @@ struct LBUG_API ParsedNativeGraphEntry : ParsedGraphEntry {
     // (multi-node-table graph, per-table predicate, manual transaction) — consumers must fall
     // back to scanning storage. Lifetime: the session's GraphEntrySet; freed on DROP.
     std::vector<std::shared_ptr<main::QueryResult>> relCsrResults;
+    // Each rel table's storage changeEpoch, captured immediately BEFORE the materializing scan
+    // (parallel to relCsrResults; meaningless where the result is null). Consumers compare it to
+    // the table's current epoch and treat any mismatch as staleness, falling back to scanning
+    // live storage. Capturing before the scan makes a concurrent mutation during materialization
+    // read as stale — conservative in the safe direction.
+    std::vector<uint64_t> relCsrEpochs;
 
     ParsedNativeGraphEntry(std::vector<ParsedNativeGraphTableInfo> nodeInfos,
         std::vector<ParsedNativeGraphTableInfo> relInfos)

test/api/project_graph_csr_test.cpp

@@ -58,6 +58,22 @@ TEST_F(ProjectGraphCsrTest, materializedCsrSurvivesConsumingQueries) {
     ASSERT_EQ(arrowResult->getCSRMetadata().indices.size(), 3u);
 }
 
+TEST_F(ProjectGraphCsrTest, recordsRelChangeEpochs) {
+    ASSERT_TRUE(conn->query("CALL PROJECT_GRAPH('CsrG', ['CsrNode'], ['CsrEdge'])")->isSuccess());
+    const auto& entry = getNativeEntry("CsrG");
+    ASSERT_EQ(entry.relCsrEpochs.size(), 1u);
+    const auto epochAtProjection = entry.relCsrEpochs[0];
+    // Mutating the rel table bumps its change epoch; a fresh projection must record a later one,
+    // which is what lets consumers detect that an old entry's pinned CSR is stale.
+    ASSERT_TRUE(conn->query("MATCH (a:CsrNode {id:2}), (b:CsrNode {id:0}) "
+                            "CREATE (a)-[:CsrEdge]->(b)")
+                    ->isSuccess());
+    ASSERT_TRUE(conn->query("CALL PROJECT_GRAPH('CsrG2', ['CsrNode'], ['CsrEdge'])")->isSuccess());
+    const auto& entry2 = getNativeEntry("CsrG2");
+    ASSERT_EQ(entry2.relCsrEpochs.size(), 1u);
+    ASSERT_GT(entry2.relCsrEpochs[0], epochAtProjection);
+}
+
 TEST_F(ProjectGraphCsrTest, skipsMaterializationWithPredicate) {
     ASSERT_TRUE(
         conn->query("CALL PROJECT_GRAPH('CsrGPred', ['CsrNode'], {CsrEdge: 'r.rowid >= 0'})")