feat(graph): materialize projected graphs as arrow CSR at PROJECT_GRAPH time

src/function/table/project_native_graph.cpp

@@ -1,12 +1,18 @@
+#include <algorithm>
+
 #include "common/exception/binder.h"
 #include "common/types/value/nested.h"
 #include "function/gds/gds.h"
 #include "function/table/bind_data.h"
 #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 <format>
 
 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<main::ArrowQueryResult*>(result.get());
+        if (arrowResult != nullptr && arrowResult->isSuccess() && arrowResult->hasCSRMetadata()) {
+            entry.relCsrResults.push_back(std::shared_ptr<main::QueryResult>{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<ProjectGraphNativeBindData*>(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<ParsedNativeGraphEntry>(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;
 }

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<QueryResult> members destroy against a complete type.
+ParsedNativeGraphEntry::~ParsedNativeGraphEntry() = default;
+
 std::string GraphEntryTypeUtils::toString(GraphEntryType type) {
     switch (type) {
     case GraphEntryType::NATIVE:

src/include/graph/parsed_graph_entry.h

@@ -1,12 +1,16 @@
 #pragma once
 
 #include <cstdint>
+#include <memory>
 #include <string>
 #include <vector>
 
 #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<ParsedNativeGraphTableInfo> nodeInfos;
     std::vector<ParsedNativeGraphTableInfo> 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<std::shared_ptr<main::QueryResult>> relCsrResults;
 
     ParsedNativeGraphEntry(std::vector<ParsedNativeGraphTableInfo> nodeInfos,
         std::vector<ParsedNativeGraphTableInfo> relInfos)
         : ParsedGraphEntry{GraphEntryType::NATIVE}, nodeInfos{std::move(nodeInfos)},
           relInfos{std::move(relInfos)} {}
+    ~ParsedNativeGraphEntry() override;
 };
 
 struct LBUG_API ParsedCypherGraphEntry : ParsedGraphEntry {

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

test/api/project_graph_csr_test.cpp

@@ -0,0 +1,98 @@
+#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<ParsedNativeGraphEntry>();
+    }
+};
+
+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<ArrowQueryResult*>(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<int64_t>{0, 2, 3, 3}));
+    ASSERT_EQ(metadata.indices, (std::vector<int64_t>{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<ArrowQueryResult*>(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<ArrowQueryResult*>(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());
+}