diff --git a/src/function/table/project_native_graph.cpp b/src/function/table/project_native_graph.cpp index f9a3192e4..8122cb2c3 100644 --- a/src/function/table/project_native_graph.cpp +++ b/src/function/table/project_native_graph.cpp @@ -1,5 +1,7 @@ #include +#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 @@ -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(); + 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(result.get()); if (arrowResult != nullptr && arrowResult->isSuccess() && arrowResult->hasCSRMetadata()) { entry.relCsrResults.push_back(std::shared_ptr{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); } } } diff --git a/src/include/graph/parsed_graph_entry.h b/src/include/graph/parsed_graph_entry.h index 0a78c93ed..b33e5f501 100644 --- a/src/include/graph/parsed_graph_entry.h +++ b/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> 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 relCsrEpochs; ParsedNativeGraphEntry(std::vector nodeInfos, std::vector relInfos) diff --git a/test/api/project_graph_csr_test.cpp b/test/api/project_graph_csr_test.cpp index 77cd85f29..9765711fa 100644 --- a/test/api/project_graph_csr_test.cpp +++ b/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'})")