From 29fe93a67a79d7eac59959f90782a16bedafd91a Mon Sep 17 00:00:00 2001 From: Zach Winter Date: Tue, 11 Aug 2026 15:38:15 -0400 Subject: [PATCH 1/2] feat(graph): materialize projected graphs as arrow CSR at PROJECT_GRAPH time MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PROJECT_GRAPH now runs each projected rel table's scan through the arrow CSR collector (queryAsArrow on an internal connection; the MATCH..RETURN rowid shape tracks CSR with no row materialization, see #626) and pins the resulting ArrowQueryResult on the graph entry. GDS consumers can wrap the CSR indptr/indices zero-copy instead of re-scanning storage per algorithm. Fallback-first: materialization is skipped — and consumers fall back to scanning storage — whenever the CSR can't faithfully represent the projection yet (multiple node tables, per-table predicates) or when the internal read connection would not see the caller's writes (manual transactions). A rel whose result shape isn't CSR-tracked pins nullptr. Lifetime: the session's GraphEntrySet owns the pinned results; dropped with the graph. Tests: api suite ProjectGraphCsrTest — CSR content on a small graph, survival across later statements, empty rel table, and all three skip guards. Co-Authored-By: Claude Fable 5 --- src/function/table/project_native_graph.cpp | 51 +++++++++- src/graph/parsed_graph_entry.cpp | 5 + src/include/graph/parsed_graph_entry.h | 11 +++ test/api/CMakeLists.txt | 1 + test/api/project_graph_csr_test.cpp | 101 ++++++++++++++++++++ 5 files changed, 167 insertions(+), 2 deletions(-) create mode 100644 test/api/project_graph_csr_test.cpp diff --git a/src/function/table/project_native_graph.cpp b/src/function/table/project_native_graph.cpp index 19a24e2a7d..f9a3192e40 100644 --- a/src/function/table/project_native_graph.cpp +++ b/src/function/table/project_native_graph.cpp @@ -1,3 +1,5 @@ +#include + #include "common/exception/binder.h" #include "common/types/value/nested.h" #include "function/gds/gds.h" @@ -5,8 +7,12 @@ #include "function/table/bind_input.h" #include "function/table/standalone_call_function.h" #include "graph/graph_entry_set.h" +#include "main/connection.h" +#include "main/database.h" +#include "main/query_result/arrow_query_result.h" #include "parser/parser.h" #include "processor/execution_context.h" +#include "transaction/transaction_context.h" #include using namespace lbug::binder; @@ -33,13 +39,54 @@ struct ProjectGraphNativeBindData final : TableFuncBindData { } }; +// Materialize each projected rel table as arrow CSR by running the projection scan through the +// arrow CSR collector (queryAsArrow tracks CSR for the MATCH..RETURN rowid shape with no row +// materialization). The result is pinned on the entry for GDS consumers to wrap zero-copy. +// Fallback-first: any condition the CSR can't faithfully represent yet (multiple node tables, +// per-table predicates) or where an internal read connection would not see the caller's data +// (manual transaction: uncommitted writes are invisible to the inner connection) skips +// materialization; consumers fall back to scanning storage. +static void materializeRelCsr(ParsedNativeGraphEntry& entry, main::ClientContext* context) { + if (entry.nodeInfos.size() != 1) { + return; + } + const auto anyPredicate = [](const ParsedNativeGraphTableInfo& info) { + return !info.predicate.empty(); + }; + if (std::any_of(entry.nodeInfos.begin(), entry.nodeInfos.end(), anyPredicate) || + std::any_of(entry.relInfos.begin(), entry.relInfos.end(), anyPredicate)) { + return; + } + if (!transaction::TransactionContext::Get(*context)->isAutoTransaction()) { + return; + } + static constexpr int64_t ARROW_CHUNK_SIZE = 1 << 16; + const auto& nodeTable = entry.nodeInfos[0].tableName; + main::Connection conn{context->getDatabase()}; + entry.relCsrResults.reserve(entry.relInfos.size()); + for (const auto& relInfo : entry.relInfos) { + 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)}); + } else { + // Shape not tracked (or scan failed): this rel stays unmaterialized. + entry.relCsrResults.push_back(nullptr); + } + } +} + static offset_t tableFunc(const TableFuncInput& input, TableFuncOutput&) { const auto bindData = dynamic_cast_checked(input.bindData); - auto graphEntrySet = GraphEntrySet::Get(*input.context->clientContext); + auto clientContext = input.context->clientContext; + auto graphEntrySet = GraphEntrySet::Get(*clientContext); graphEntrySet->validateGraphNotExist(bindData->graphName); auto entry = std::make_unique(bindData->nodeInfos, bindData->relInfos); // bind graph entry to check if input is valid or not. Ignore bind result. - GDSFunction::bindGraphEntry(*input.context->clientContext, *entry); + GDSFunction::bindGraphEntry(*clientContext, *entry); + materializeRelCsr(*entry, clientContext); graphEntrySet->addGraph(bindData->graphName, std::move(entry)); return 0; } diff --git a/src/graph/parsed_graph_entry.cpp b/src/graph/parsed_graph_entry.cpp index 24b0730466..2759eddebd 100644 --- a/src/graph/parsed_graph_entry.cpp +++ b/src/graph/parsed_graph_entry.cpp @@ -1,10 +1,15 @@ #include "graph/parsed_graph_entry.h" +#include "main/query_result.h" + using namespace lbug::common; namespace lbug { namespace graph { +// Defined here so the shared_ptr members destroy against a complete type. +ParsedNativeGraphEntry::~ParsedNativeGraphEntry() = default; + std::string GraphEntryTypeUtils::toString(GraphEntryType type) { switch (type) { case GraphEntryType::NATIVE: diff --git a/src/include/graph/parsed_graph_entry.h b/src/include/graph/parsed_graph_entry.h index 46215aef30..0a78c93eda 100644 --- a/src/include/graph/parsed_graph_entry.h +++ b/src/include/graph/parsed_graph_entry.h @@ -1,12 +1,16 @@ #pragma once #include +#include #include #include #include "common/cast.h" namespace lbug { +namespace main { +class QueryResult; +} // namespace main namespace graph { enum class GraphEntryType : uint8_t { @@ -41,11 +45,18 @@ struct ParsedNativeGraphTableInfo { struct LBUG_API ParsedNativeGraphEntry : ParsedGraphEntry { std::vector nodeInfos; std::vector relInfos; + // Arrow CSR per relInfos[i] (same order), materialized at PROJECT_GRAPH time by running the + // projection scan through the arrow CSR collector; entries are ArrowQueryResults whose + // CSRMetadata GDS consumers wrap zero-copy. Null (or empty) when materialization was skipped + // (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; ParsedNativeGraphEntry(std::vector nodeInfos, std::vector relInfos) : ParsedGraphEntry{GraphEntryType::NATIVE}, nodeInfos{std::move(nodeInfos)}, relInfos{std::move(relInfos)} {} + ~ParsedNativeGraphEntry() override; }; struct LBUG_API ParsedCypherGraphEntry : ParsedGraphEntry { diff --git a/test/api/CMakeLists.txt b/test/api/CMakeLists.txt index b5ab7860d8..f636439dfc 100644 --- a/test/api/CMakeLists.txt +++ b/test/api/CMakeLists.txt @@ -12,6 +12,7 @@ add_lbug_api_test(api_test drop_index_test.cpp arrow_table_function_test.cpp prepare_test.cpp + project_graph_csr_test.cpp result_value_test.cpp storage_driver_test.cpp udf_test.cpp diff --git a/test/api/project_graph_csr_test.cpp b/test/api/project_graph_csr_test.cpp new file mode 100644 index 0000000000..930e6ad719 --- /dev/null +++ b/test/api/project_graph_csr_test.cpp @@ -0,0 +1,101 @@ +#include "api_test/api_test.h" +#include "graph/graph_entry_set.h" +#include "graph/parsed_graph_entry.h" +#include "main/query_result/arrow_query_result.h" + +using namespace lbug::common; +using namespace lbug::graph; +using namespace lbug::main; +using namespace lbug::testing; + +class ProjectGraphCsrTest : public ApiTest { +public: + void SetUp() override { + ApiTest::SetUp(); + ASSERT_TRUE(conn->query("CREATE NODE TABLE CsrNode(id INT64 PRIMARY KEY)")->isSuccess()); + ASSERT_TRUE(conn->query("CREATE REL TABLE CsrEdge(FROM CsrNode TO CsrNode)")->isSuccess()); + ASSERT_TRUE(conn->query("CREATE (:CsrNode {id:0}), (:CsrNode {id:1}), (:CsrNode {id:2})") + ->isSuccess()); + ASSERT_TRUE( + conn->query("MATCH (a:CsrNode {id:0}), (b:CsrNode {id:1}) " + "CREATE (a)-[:CsrEdge]->(b)") + ->isSuccess()); + ASSERT_TRUE( + conn->query("MATCH (a:CsrNode {id:0}), (b:CsrNode {id:2}) " + "CREATE (a)-[:CsrEdge]->(b)") + ->isSuccess()); + ASSERT_TRUE( + conn->query("MATCH (a:CsrNode {id:1}), (b:CsrNode {id:2}) " + "CREATE (a)-[:CsrEdge]->(b)") + ->isSuccess()); + } + + const ParsedNativeGraphEntry& getNativeEntry(const std::string& name) { + auto* set = GraphEntrySet::Get(*conn->getClientContext()); + EXPECT_TRUE(set->hasGraph(name)); + return set->getEntry(name)->cast(); + } +}; + +TEST_F(ProjectGraphCsrTest, materializesArrowCsr) { + ASSERT_TRUE(conn->query("CALL PROJECT_GRAPH('CsrG', ['CsrNode'], ['CsrEdge'])")->isSuccess()); + const auto& entry = getNativeEntry("CsrG"); + ASSERT_EQ(entry.relCsrResults.size(), 1u); + ASSERT_NE(entry.relCsrResults[0], nullptr); + auto* arrowResult = dynamic_cast(entry.relCsrResults[0].get()); + ASSERT_NE(arrowResult, nullptr); + ASSERT_TRUE(arrowResult->hasCSRMetadata()); + // Graph is 0->1, 0->2, 1->2 over rowids 0..2: indptr [0,2,3,3], indices [1,2,2]. + const auto& metadata = arrowResult->getCSRMetadata(); + ASSERT_EQ(metadata.indptr, (std::vector{0, 2, 3, 3})); + ASSERT_EQ(metadata.indices, (std::vector{1, 2, 2})); +} + +TEST_F(ProjectGraphCsrTest, materializedCsrSurvivesConsumingQueries) { + ASSERT_TRUE(conn->query("CALL PROJECT_GRAPH('CsrG', ['CsrNode'], ['CsrEdge'])")->isSuccess()); + // The pinned result must stay valid across later statements on the same connection. + ASSERT_TRUE(conn->query("MATCH (a:CsrNode) RETURN COUNT(*)")->isSuccess()); + const auto& entry = getNativeEntry("CsrG"); + auto* arrowResult = dynamic_cast(entry.relCsrResults[0].get()); + ASSERT_NE(arrowResult, nullptr); + ASSERT_EQ(arrowResult->getCSRMetadata().indices.size(), 3u); +} + +TEST_F(ProjectGraphCsrTest, skipsMaterializationWithPredicate) { + ASSERT_TRUE( + conn->query("CALL PROJECT_GRAPH('CsrGPred', ['CsrNode'], {CsrEdge: 'r.rowid >= 0'})") + ->isSuccess()); + const auto& entry = getNativeEntry("CsrGPred"); + ASSERT_TRUE(entry.relCsrResults.empty()); +} + +TEST_F(ProjectGraphCsrTest, skipsMaterializationWithMultipleNodeTables) { + ASSERT_TRUE(conn->query("CREATE NODE TABLE CsrNode2(id INT64 PRIMARY KEY)")->isSuccess()); + ASSERT_TRUE(conn->query("CALL PROJECT_GRAPH('CsrGMulti', ['CsrNode', 'CsrNode2'], ['CsrEdge'])") + ->isSuccess()); + const auto& entry = getNativeEntry("CsrGMulti"); + ASSERT_TRUE(entry.relCsrResults.empty()); +} + +TEST_F(ProjectGraphCsrTest, handlesEmptyRelTable) { + ASSERT_TRUE(conn->query("CREATE REL TABLE CsrEdgeEmpty(FROM CsrNode TO CsrNode)")->isSuccess()); + ASSERT_TRUE( + conn->query("CALL PROJECT_GRAPH('CsrGEmpty', ['CsrNode'], ['CsrEdgeEmpty'])")->isSuccess()); + const auto& entry = getNativeEntry("CsrGEmpty"); + ASSERT_EQ(entry.relCsrResults.size(), 1u); + // Zero edges: either unmaterialized (consumers fall back to scan) or a valid all-empty CSR. + if (entry.relCsrResults[0] != nullptr) { + auto* arrowResult = dynamic_cast(entry.relCsrResults[0].get()); + ASSERT_NE(arrowResult, nullptr); + ASSERT_EQ(arrowResult->getCSRMetadata().indices.size(), 0u); + } +} + +TEST_F(ProjectGraphCsrTest, skipsMaterializationInManualTransaction) { + ASSERT_TRUE(conn->query("BEGIN TRANSACTION")->isSuccess()); + ASSERT_TRUE( + conn->query("CALL PROJECT_GRAPH('CsrGTxn', ['CsrNode'], ['CsrEdge'])")->isSuccess()); + ASSERT_TRUE(conn->query("COMMIT")->isSuccess()); + const auto& entry = getNativeEntry("CsrGTxn"); + ASSERT_TRUE(entry.relCsrResults.empty()); +} From b8b1da7d9894ce2c5a2cad92578207981ff396a6 Mon Sep 17 00:00:00 2001 From: Zach Winter Date: Tue, 11 Aug 2026 15:47:29 -0400 Subject: [PATCH 2/2] style: clang-format-18 Co-Authored-By: Claude Fable 5 --- test/api/project_graph_csr_test.cpp | 25 +++++++++++-------------- 1 file changed, 11 insertions(+), 14 deletions(-) diff --git a/test/api/project_graph_csr_test.cpp b/test/api/project_graph_csr_test.cpp index 930e6ad719..77cd85f298 100644 --- a/test/api/project_graph_csr_test.cpp +++ b/test/api/project_graph_csr_test.cpp @@ -15,19 +15,16 @@ class ProjectGraphCsrTest : public ApiTest { ASSERT_TRUE(conn->query("CREATE NODE TABLE CsrNode(id INT64 PRIMARY KEY)")->isSuccess()); ASSERT_TRUE(conn->query("CREATE REL TABLE CsrEdge(FROM CsrNode TO CsrNode)")->isSuccess()); ASSERT_TRUE(conn->query("CREATE (:CsrNode {id:0}), (:CsrNode {id:1}), (:CsrNode {id:2})") - ->isSuccess()); - ASSERT_TRUE( - conn->query("MATCH (a:CsrNode {id:0}), (b:CsrNode {id:1}) " - "CREATE (a)-[:CsrEdge]->(b)") - ->isSuccess()); - ASSERT_TRUE( - conn->query("MATCH (a:CsrNode {id:0}), (b:CsrNode {id:2}) " - "CREATE (a)-[:CsrEdge]->(b)") - ->isSuccess()); - ASSERT_TRUE( - conn->query("MATCH (a:CsrNode {id:1}), (b:CsrNode {id:2}) " - "CREATE (a)-[:CsrEdge]->(b)") - ->isSuccess()); + ->isSuccess()); + ASSERT_TRUE(conn->query("MATCH (a:CsrNode {id:0}), (b:CsrNode {id:1}) " + "CREATE (a)-[:CsrEdge]->(b)") + ->isSuccess()); + ASSERT_TRUE(conn->query("MATCH (a:CsrNode {id:0}), (b:CsrNode {id:2}) " + "CREATE (a)-[:CsrEdge]->(b)") + ->isSuccess()); + ASSERT_TRUE(conn->query("MATCH (a:CsrNode {id:1}), (b:CsrNode {id:2}) " + "CREATE (a)-[:CsrEdge]->(b)") + ->isSuccess()); } const ParsedNativeGraphEntry& getNativeEntry(const std::string& name) { @@ -72,7 +69,7 @@ TEST_F(ProjectGraphCsrTest, skipsMaterializationWithPredicate) { TEST_F(ProjectGraphCsrTest, skipsMaterializationWithMultipleNodeTables) { ASSERT_TRUE(conn->query("CREATE NODE TABLE CsrNode2(id INT64 PRIMARY KEY)")->isSuccess()); ASSERT_TRUE(conn->query("CALL PROJECT_GRAPH('CsrGMulti', ['CsrNode', 'CsrNode2'], ['CsrEdge'])") - ->isSuccess()); + ->isSuccess()); const auto& entry = getNativeEntry("CsrGMulti"); ASSERT_TRUE(entry.relCsrResults.empty()); }