diff --git a/algo/src/function/CMakeLists.txt b/algo/src/function/CMakeLists.txt index afdee9ab..3bf896d0 100644 --- a/algo/src/function/CMakeLists.txt +++ b/algo/src/function/CMakeLists.txt @@ -13,7 +13,7 @@ set(_algo_function_sources if (ICEBUG_ENABLED) # gds_page_rank.cpp includes NetworKit + Arrow headers. Linking the imported targets to the # object library propagates their compile flags + include dirs to it. - list(APPEND _algo_function_sources gds_page_rank.cpp) + list(APPEND _algo_function_sources gds_csr_bridge.cpp gds_page_rank.cpp) endif () add_library(lbug_algo_function diff --git a/algo/src/function/gds_csr_bridge.cpp b/algo/src/function/gds_csr_bridge.cpp new file mode 100644 index 00000000..f1a82dce --- /dev/null +++ b/algo/src/function/gds_csr_bridge.cpp @@ -0,0 +1,161 @@ +// The shared projected-graph -> Arrow CSR bridge for icebug-backed GDS functions. +// +// Zero-copy path: PROJECT_GRAPH materializes each rel table's scan as arrow CSR and pins the +// ArrowQueryResult on the graph entry (ladybug#800). We fetch it here, symmetrize() it into the +// undirected view (ladybug#799 — per-row sorted, reciprocal pairs coalesced), import the C-ABI +// ArrowArrays into arrow::Arrays, and reinterpret the int64 buffers as uint64 in place. No edge +// is copied at any step. +// +// Fallback path: scan storage fwd + bwd into an InMemGraph and build the arrays with a builder — +// the original bridge path, kept for projections the pinned CSR can't represent (filters, +// multi-node-table graphs, manual-transaction projections, stale dimensions). +#include "function/gds_csr_bridge.h" + +#include + +#include "common/in_mem_graph.h" +#include "graph/graph.h" +#include "graph/graph_entry_set.h" +#include "graph/parsed_graph_entry.h" +#include "main/client_context.h" +#include "main/query_result/arrow_query_result.h" +#include "storage/storage_manager.h" +#include "storage/table/table.h" +#include + +using namespace lbug::common; +using namespace lbug::graph; + +namespace lbug { +namespace algo_extension { + +// Import a C-ABI ArrowArray (int64) and reinterpret it as uint64 without copying: rowids are +// non-negative, and the buffer layout is identical. The imported array's release callback keeps +// the underlying buffers alive for as long as the returned array (and anything constructed over +// its buffers, like GraphR) is held. +static std::shared_ptr importAsU64( + main::ArrowQueryResult::CSRArrowArray&& csrArray) { + auto imported = arrow::ImportArray(&csrArray.array, &csrArray.schema); + if (!imported.ok()) { + return nullptr; + } + auto data = (*imported)->data()->Copy(); // shallow: buffers are shared, not copied + data->type = arrow::uint64(); + return std::make_shared(std::move(data)); +} + +// The zero-copy path; returns {nullptr, nullptr} whenever any precondition fails so the caller +// falls back to the scan path. +static BridgeCSR fromMaterializedCsr(main::ClientContext* context, const std::string& graphName, + Graph* graph, table_id_t tableID, offset_t numNodes) { + if (graphName.empty()) { + return {}; + } + auto* entrySet = GraphEntrySet::Get(*context); + if (!entrySet->hasGraph(graphName)) { + return {}; + } + auto* parsed = entrySet->getEntry(graphName); + if (parsed->type != GraphEntryType::NATIVE) { + return {}; + } + auto& native = parsed->cast(); + // MVP mirrors the GDS functions: single rel table, first entry. + if (native.relCsrResults.empty() || native.relCsrResults[0] == nullptr) { + return {}; + } + auto* arrowResult = dynamic_cast(native.relCsrResults[0].get()); + if (arrowResult == nullptr || !arrowResult->hasCSRMetadata()) { + return {}; + } + // Staleness check: the CSR was pinned at projection time with the rel table's change epoch. + // Any mutation since (even one leaving node cardinality unchanged) bumps the epoch — serve + // live storage via the fallback instead of a stale snapshot. + if (native.relCsrEpochs.size() != native.relCsrResults.size()) { + return {}; + } + const auto nbrTables = graph->getRelInfos(tableID); + if (nbrTables.empty()) { + return {}; + } + const auto* relTable = + storage::StorageManager::Get(*context)->getTable(nbrTables[0].relTableID); + if (relTable == nullptr || relTable->getChangeEpoch() != native.relCsrEpochs[0]) { + return {}; + } + auto symmetric = arrowResult->getCSRArrowArrays().symmetrize(); + auto indptr = importAsU64(std::move(symmetric.indptr)); + auto indices = importAsU64(std::move(symmetric.indices)); + if (indptr == nullptr || indices == nullptr) { + return {}; + } + // Dimension check: the pinned CSR reflects projection-time cardinality. If the node table + // grew since, offsets would be out of range — fall back to a fresh scan. + if (static_cast(indptr->length()) != numNodes + 1) { + return {}; + } + return {std::move(indptr), std::move(indices)}; +} + +// The fallback: materialize the undirected adjacency by scanning storage. Per-row sort + unique +// coalesces parallel edges and reciprocal pairs so both paths build the same SIMPLE undirected +// graph — symmetrize()'s A + A.T semantics. (The pre-bridge scan kept multiplicity; that made +// degree-sensitive algorithms diverge between the two paths on multigraph data.) +static void scanCSR(table_id_t tableID, offset_t numNodes, Graph* graph, InMemGraph& inMem) { + const auto nbrTables = graph->getRelInfos(tableID); + const auto nbrInfo = nbrTables[0]; + const auto scanState = graph->prepareRelScan(*nbrInfo.relGroupEntry, nbrInfo.relTableID, + nbrInfo.dstTableID, {}, false /*randomLookup*/); + std::vector nbrs; + for (offset_t nodeId = 0; nodeId < numNodes; ++nodeId) { + nbrs.clear(); + const nodeID_t nid = {nodeId, tableID}; + for (auto chunk : graph->scanFwd(nid, *scanState)) { + chunk.forEach( + [&](auto neighbors, auto, auto i) { nbrs.push_back(neighbors[i].offset); }); + } + for (auto chunk : graph->scanBwd(nid, *scanState)) { + chunk.forEach([&](auto neighbors, auto, auto i) { + if (neighbors[i].offset != nodeId) { + nbrs.push_back(neighbors[i].offset); + } + }); + } + std::sort(nbrs.begin(), nbrs.end()); + nbrs.erase(std::unique(nbrs.begin(), nbrs.end()), nbrs.end()); + inMem.initNextNode(); + for (const auto nbr : nbrs) { + inMem.insertNbr(nbr); + } + } + inMem.initNextNode(); // trailing sentinel: csrOffsets[numNodes] == numEdges +} + +static std::shared_ptr toU64(const std::function& at, + offset_t count) { + arrow::UInt64Builder builder; + (void)builder.Reserve(count); + for (offset_t i = 0; i < count; ++i) { + (void)builder.Append(static_cast(at(i))); + } + std::shared_ptr arr; + (void)builder.Finish(&arr); + return std::static_pointer_cast(arr); +} + +BridgeCSR buildUndirectedCSR(main::ClientContext* context, const std::string& graphName, + Graph* graph, table_id_t tableID, offset_t numNodes, storage::MemoryManager* mm) { + auto materialized = fromMaterializedCsr(context, graphName, graph, tableID, numNodes); + if (materialized.indptr != nullptr && materialized.indices != nullptr) { + return materialized; + } + InMemGraph inMem(numNodes, mm); + scanCSR(tableID, numNodes, graph, inMem); + auto indptr = toU64([&](offset_t i) { return inMem.csrOffsets[i]; }, numNodes + 1); + auto indices = + toU64([&](offset_t i) { return inMem.csrEdges[i].neighbor; }, inMem.csrEdges.size()); + return {std::move(indptr), std::move(indices)}; +} + +} // namespace algo_extension +} // namespace lbug diff --git a/algo/src/function/gds_page_rank.cpp b/algo/src/function/gds_page_rank.cpp index fb645918..32b84653 100644 --- a/algo/src/function/gds_page_rank.cpp +++ b/algo/src/function/gds_page_rank.cpp @@ -7,9 +7,9 @@ // one release cycle; the icebug-backed ones live alongside under GDS_* names. #include "binder/binder.h" #include "common/exception/binder.h" -#include "common/in_mem_graph.h" #include "common/string_utils.h" #include "function/algo_function.h" +#include "function/gds_csr_bridge.h" #include "function/config/max_iterations_config.h" #include "function/config/page_rank_config.h" #include "function/gds/gds_utils.h" @@ -69,10 +69,14 @@ struct GDSPageRankOptionalParams final : public MaxIterationOptionalParams { }; struct GDSPageRankBindData final : public GDSBindData { + // Projected graph name, for looking up the entry's materialized arrow CSR at run time. + std::string graphName; + GDSPageRankBindData(expression_vector columns, graph::NativeGraphEntry graphEntry, std::shared_ptr nodeOutput, - std::unique_ptr optionalParams) - : GDSBindData{std::move(columns), std::move(graphEntry), expression_vector{nodeOutput}} { + std::unique_ptr optionalParams, std::string graphName) + : GDSBindData{std::move(columns), std::move(graphEntry), expression_vector{nodeOutput}}, + graphName{std::move(graphName)} { this->optionalParams = std::move(optionalParams); } @@ -114,43 +118,6 @@ class GDSPageRankResultVertexCompute : public GDSResultVertexCompute { std::unique_ptr rankVector; }; -// Materialize the projected graph's undirected adjacency as an InMemGraph CSR (fwd + bwd), -// exactly as the Louvain path does. -static void buildCSR(table_id_t tableID, offset_t numNodes, Graph* graph, InMemGraph& inMem) { - const auto nbrTables = graph->getRelInfos(tableID); - const auto nbrInfo = nbrTables[0]; - const auto scanState = graph->prepareRelScan(*nbrInfo.relGroupEntry, nbrInfo.relTableID, - nbrInfo.dstTableID, {}, false /*randomLookup*/); - for (offset_t nodeId = 0; nodeId < numNodes; ++nodeId) { - inMem.initNextNode(); - const nodeID_t nid = {nodeId, tableID}; - for (auto chunk : graph->scanFwd(nid, *scanState)) { - chunk.forEach( - [&](auto neighbors, auto, auto i) { inMem.insertNbr(neighbors[i].offset); }); - } - for (auto chunk : graph->scanBwd(nid, *scanState)) { - chunk.forEach([&](auto neighbors, auto, auto i) { - if (neighbors[i].offset != nodeId) { - inMem.insertNbr(neighbors[i].offset); - } - }); - } - } - inMem.initNextNode(); // trailing sentinel: csrOffsets[numNodes] == numEdges -} - -static std::shared_ptr toU64(const std::function& at, - offset_t count) { - arrow::UInt64Builder builder; - (void)builder.Reserve(count); - for (offset_t i = 0; i < count; ++i) { - (void)builder.Append(static_cast(at(i))); - } - std::shared_ptr arr; - (void)builder.Finish(&arr); - return std::static_pointer_cast(arr); -} - static offset_t tableFunc(const TableFuncInput& input, TableFuncOutput&) { auto clientContext = input.context->clientContext; auto transaction = transaction::Transaction::Get(*clientContext); @@ -167,17 +134,12 @@ static offset_t tableFunc(const TableFuncInput& input, TableFuncOutput&) { auto bindData = input.bindData->constPtrCast(); auto& config = bindData->optionalParams->constCast(); - // 1. Ladybug engine graph -> InMemGraph CSR. - InMemGraph inMem(numNodes, mm); - buildCSR(tableID, numNodes, graph, inMem); - - // 2. CSR -> Arrow UInt64 arrays (indptr length numNodes+1, indices length numEdges). - auto outIndptr = toU64([&](offset_t i) { return inMem.csrOffsets[i]; }, numNodes + 1); - auto outIndices = - toU64([&](offset_t i) { return inMem.csrEdges[i].neighbor; }, inMem.csrEdges.size()); + // 1. Undirected CSR — zero-copy from the projected graph's materialized arrow CSR when + // available, scan fallback otherwise (see gds_csr_bridge.cpp). + auto csr = buildUndirectedCSR(clientContext, bindData->graphName, graph, tableID, numNodes, mm); - // 3. icebug: zero-copy GraphR over the Arrow CSR, then PageRank. - NetworKit::GraphR g(numNodes, /*directed=*/false, outIndices, outIndptr); + // 2. icebug: zero-copy GraphR over the Arrow CSR, then PageRank. + NetworKit::GraphR g(numNodes, /*directed=*/false, csr.indices, csr.indptr); NetworKit::PageRank pr(g, config.dampingFactor.getParamVal(), config.tolerance.getParamVal()); pr.run(); const std::vector& scores = pr.scores(); @@ -200,7 +162,8 @@ static std::unique_ptr bindFunc(main::ClientContext* context, columns.push_back(nodeOutput->constCast().getInternalID()); columns.push_back(input->binder->createVariable(RANK_COLUMN_NAME, LogicalType::DOUBLE())); return std::make_unique(std::move(columns), std::move(graphEntry), - nodeOutput, std::make_unique(input->optionalParamsLegacy)); + nodeOutput, std::make_unique(input->optionalParamsLegacy), + std::move(graphName)); } function_set GDSPageRankFunction::getFunctionSet() { diff --git a/algo/src/include/function/gds_csr_bridge.h b/algo/src/include/function/gds_csr_bridge.h new file mode 100644 index 00000000..1cc40532 --- /dev/null +++ b/algo/src/include/function/gds_csr_bridge.h @@ -0,0 +1,44 @@ +#pragma once + +#include +#include + +#include "common/types/types.h" +#include + +namespace lbug { +namespace main { +class ClientContext; +} // namespace main +namespace graph { +class Graph; +} // namespace graph +namespace storage { +class MemoryManager; +} // namespace storage + +namespace algo_extension { + +// Symmetric (undirected) SIMPLE adjacency of the projected graph as Arrow CSR, ready to hand +// to NetworKit::GraphR. Shared by every icebug-backed GDS function. Both construction paths +// coalesce parallel edges and reciprocal pairs (symmetrize()'s A + A.T semantics): GDS +// operates on the simple undirected projection. +struct BridgeCSR { + std::shared_ptr indptr; // length numNodes + 1 + std::shared_ptr indices; // length numEdges (symmetric, per-row sorted + // on the zero-copy path) +}; + +// Build the undirected CSR for a single-node-table projected graph. +// +// Fast path (zero-copy): PROJECT_GRAPH pins each rel table's arrow CSR on the graph entry at +// projection time; we take that, symmetrize() it (A + A.T, per-row sorted), and reinterpret the +// int64 buffers as uint64 without copying. Falls back to scanning storage (fwd + bwd) into an +// InMemGraph whenever the pinned CSR is unavailable (filtered projection, multi-node-table +// graph, manual-transaction projection, stale dimensions). +BridgeCSR buildUndirectedCSR(main::ClientContext* context, const std::string& graphName, + graph::Graph* graph, common::table_id_t tableID, common::offset_t numNodes, + storage::MemoryManager* mm); + +} // namespace algo_extension +} // namespace lbug diff --git a/algo/test/test_files/gds_page_rank.test b/algo/test/test_files/gds_page_rank.test index fadb7800..b3587e5e 100644 --- a/algo/test/test_files/gds_page_rank.test +++ b/algo/test/test_files/gds_page_rank.test @@ -19,9 +19,98 @@ -STATEMENT CALL PROJECT_GRAPH('G', ['N'], ['E']) ---- ok -LOG icebug PageRank: the hub (id 0) ranks highest; scores are a sum-to-1 distribution. +-LOG Plain projection consumes the graph entry's materialized arrow CSR (zero-copy path). -STATEMENT CALL GDS_PAGE_RANK('G') RETURN node.id, rank ORDER BY rank DESC, node.id ---- 4 0|0.479730 1|0.173423 2|0.173423 3|0.173423 +-LOG Filtered projection skips materialization; the scan fallback must produce identical ranks. +-STATEMENT CALL PROJECT_GRAPH('GPred', ['N'], {E: 'r.rowid >= 0'}) +---- ok +-STATEMENT CALL GDS_PAGE_RANK('GPred') RETURN node.id, rank ORDER BY rank DESC, node.id +---- 4 +0|0.479730 +1|0.173423 +2|0.173423 +3|0.173423 + +-CASE GDSPageRankParallelEdges +-LOAD_DYNAMIC_EXTENSION algo +-STATEMENT CREATE NODE TABLE N(id INT64 PRIMARY KEY) +---- ok +-STATEMENT CREATE REL TABLE E(FROM N TO N) +---- ok +-STATEMENT CREATE (a:N{id:0}), (b:N{id:1}), (c:N{id:2}) +---- ok +-LOG Multigraph: (0,1) twice, plus the reciprocal (1,0), plus (1,2). +-STATEMENT MATCH (x:N{id:0}), (y:N{id:1}) CREATE (x)-[:E]->(y) +---- ok +-STATEMENT MATCH (x:N{id:0}), (y:N{id:1}) CREATE (x)-[:E]->(y) +---- ok +-STATEMENT MATCH (x:N{id:1}), (y:N{id:0}) CREATE (x)-[:E]->(y) +---- ok +-STATEMENT MATCH (x:N{id:1}), (y:N{id:2}) CREATE (x)-[:E]->(y) +---- ok +-LOG GDS operates on the SIMPLE undirected projection: parallel edges and reciprocal pairs +-LOG coalesce on BOTH paths, so the multigraph ranks equal the clean path graph 0-1-2 below. +-STATEMENT CALL PROJECT_GRAPH('GM', ['N'], ['E']) +---- ok +-STATEMENT CALL GDS_PAGE_RANK('GM') RETURN node.id, rank ORDER BY node.id +---- 3 +0|0.256757 +1|0.486486 +2|0.256757 +-LOG Same multigraph through the scan fallback (filtered projection): identical ranks. +-STATEMENT CALL PROJECT_GRAPH('GMPred', ['N'], {E: 'r.rowid >= 0'}) +---- ok +-STATEMENT CALL GDS_PAGE_RANK('GMPred') RETURN node.id, rank ORDER BY node.id +---- 3 +0|0.256757 +1|0.486486 +2|0.256757 +-LOG Cross-check: a clean simple path 0-1-2 produces the same distribution. +-STATEMENT CREATE NODE TABLE NC(id INT64 PRIMARY KEY) +---- ok +-STATEMENT CREATE REL TABLE EC(FROM NC TO NC) +---- ok +-STATEMENT CREATE (a:NC{id:0}), (b:NC{id:1}), (c:NC{id:2}) +---- ok +-STATEMENT MATCH (x:NC{id:0}), (y:NC{id:1}) CREATE (x)-[:EC]->(y) +---- ok +-STATEMENT MATCH (x:NC{id:1}), (y:NC{id:2}) CREATE (x)-[:EC]->(y) +---- ok +-STATEMENT CALL PROJECT_GRAPH('GC', ['NC'], ['EC']) +---- ok +-STATEMENT CALL GDS_PAGE_RANK('GC') RETURN node.id, rank ORDER BY node.id +---- 3 +0|0.256757 +1|0.486486 +2|0.256757 + +-CASE GDSPageRankStaleProjection +-LOAD_DYNAMIC_EXTENSION algo +-STATEMENT CREATE NODE TABLE N(id INT64 PRIMARY KEY) +---- ok +-STATEMENT CREATE REL TABLE E(FROM N TO N) +---- ok +-STATEMENT CREATE (a:N{id:0}), (b:N{id:1}), (c:N{id:2}), (d:N{id:3}) +---- ok +-STATEMENT MATCH (x:N{id:1}), (y:N{id:0}) CREATE (x)-[:E]->(y) +---- ok +-STATEMENT MATCH (x:N{id:2}), (y:N{id:0}) CREATE (x)-[:E]->(y) +---- ok +-STATEMENT CALL PROJECT_GRAPH('GS', ['N'], ['E']) +---- ok +-LOG Mutate the rel table AFTER projection (node cardinality unchanged): the pinned CSR is now +-LOG stale, the change-epoch check must reject it, and the fallback must serve LIVE storage — +-LOG so the ranks below are the full 3-leaf star, not the 2-edge graph that was projected. +-STATEMENT MATCH (x:N{id:3}), (y:N{id:0}) CREATE (x)-[:E]->(y) +---- ok +-STATEMENT CALL GDS_PAGE_RANK('GS') RETURN node.id, rank ORDER BY rank DESC, node.id +---- 4 +0|0.479730 +1|0.173423 +2|0.173423 +3|0.173423