From 3f4d6cf8e252dd4e058fc14e392b2eed5e6133f7 Mon Sep 17 00:00:00 2001 From: Zach Winter Date: Sat, 8 Aug 2026 14:41:18 -0400 Subject: [PATCH 1/3] feat(algo): add GDS_LOUVAIN backed by icebug (NetworKit PLM) Louvain community detection via the icebug bridge: projected graph -> InMemGraph CSR -> Arrow UInt64 arrays (system pool, see previous commit) -> zero-copy NetworKit::GraphR -> PLM -> community assignments streamed back through the GDS result pipeline. Coexists with the hand-rolled LOUVAIN under the GDS_ prefix, same as GDS_PAGE_RANK. Test asserts community structure (two 4-cliques joined by a bridge edge), not raw IDs, since PLM is parallel and nondeterministic. Co-Authored-By: Claude Fable 5 --- algo/src/function/CMakeLists.txt | 7 +- algo/src/function/gds_louvain.cpp | 147 ++++++++++++++++++++++ algo/src/include/function/algo_function.h | 8 ++ algo/src/main/algo_extension.cpp | 1 + algo/test/test_files/gds_louvain.test | 46 +++++++ 5 files changed, 206 insertions(+), 3 deletions(-) create mode 100644 algo/src/function/gds_louvain.cpp create mode 100644 algo/test/test_files/gds_louvain.test diff --git a/algo/src/function/CMakeLists.txt b/algo/src/function/CMakeLists.txt index 06200b02..e72b942a 100644 --- a/algo/src/function/CMakeLists.txt +++ b/algo/src/function/CMakeLists.txt @@ -11,9 +11,10 @@ set(_algo_function_sources spanning_forest.cpp ) 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_csr_bridge.cpp gds_page_rank.cpp gds_node2vec.cpp) + # The gds_*.cpp sources include 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_csr_bridge.cpp gds_page_rank.cpp gds_node2vec.cpp + gds_louvain.cpp) endif () add_library(lbug_algo_function diff --git a/algo/src/function/gds_louvain.cpp b/algo/src/function/gds_louvain.cpp new file mode 100644 index 00000000..ba21d229 --- /dev/null +++ b/algo/src/function/gds_louvain.cpp @@ -0,0 +1,147 @@ +// GDS_LOUVAIN — Louvain community detection backed by icebug (a NetworKit fork with zero-copy +// Arrow CSR ingest). Same CALL surface as the hand-rolled LOUVAIN, but the compute is delegated +// to libnetworkit: we materialize the projected graph's adjacency as CSR, build a +// NetworKit::GraphR over Arrow buffers, run NetworKit::PLM (Parallel Louvain Method), and stream +// the resulting community assignments back through the GDS result pipeline. +// +// Part of the icebug bridge (adsharma-invited): the `algo` extension keeps its existing algos for +// one release cycle; the icebug-backed ones live alongside under GDS_* names. +#include "binder/binder.h" +#include "common/exception/binder.h" +#include "function/algo_function.h" +#include "function/gds/gds_utils.h" +#include "function/gds/gds_vertex_compute.h" +#include "function/gds_csr_bridge.h" +#include "function/table/bind_input.h" +#include "processor/execution_context.h" +#include "transaction/transaction.h" +#include +#include +#include +#include + +using namespace lbug::processor; +using namespace lbug::common; +using namespace lbug::binder; +using namespace lbug::storage; +using namespace lbug::graph; +using namespace lbug::function; + +namespace lbug { +namespace algo_extension { + +// Emits (node, community_id) rows, reading the icebug PLM partition by node offset. +class GDSLouvainResultVertexCompute : public GDSResultVertexCompute { +public: + GDSLouvainResultVertexCompute(storage::MemoryManager* mm, GDSFuncSharedState* sharedState, + const NetworKit::Partition& partition) + : GDSResultVertexCompute{mm, sharedState}, partition{partition} { + nodeIDVector = createVector(LogicalType::INTERNAL_ID()); + communityIDVector = createVector(LogicalType::INT64()); + } + + void beginOnTableInternal(table_id_t) override {} + + void vertexCompute(offset_t startOffset, offset_t endOffset, table_id_t tableID) override { + for (auto i = startOffset; i < endOffset; ++i) { + if (skip(i)) { + continue; + } + nodeIDVector->setValue(0, nodeID_t{i, tableID}); + communityIDVector->setValue(0, + i < partition.numberOfElements() ? static_cast(partition[i]) : -1); + localFT->append(vectors); + } + } + + std::unique_ptr copy() override { + return std::make_unique(mm, sharedState, partition); + } + +private: + const NetworKit::Partition& partition; + std::unique_ptr nodeIDVector; + std::unique_ptr communityIDVector; +}; + +// CSR construction lives in the shared bridge (gds_csr_bridge.cpp): zero-copy from the graph +// entry's materialized arrow CSR when available, storage-scan fallback otherwise. + +struct GDSLouvainBindData final : public GDSBindData { + // Projected graph name, for looking up the entry's materialized arrow CSR at run time. + std::string graphName; + + GDSLouvainBindData(expression_vector columns, graph::NativeGraphEntry graphEntry, + expression_vector output, std::string graphName) + : GDSBindData{std::move(columns), std::move(graphEntry), std::move(output)}, + graphName{std::move(graphName)} {} + + std::unique_ptr copy() const override { + return std::make_unique(*this); + } +}; + +static offset_t tableFunc(const TableFuncInput& input, TableFuncOutput&) { + auto clientContext = input.context->clientContext; + auto transaction = transaction::Transaction::Get(*clientContext); + auto sharedState = input.sharedState->ptrCast(); + auto graph = sharedState->graph.get(); + auto maxOffsetMap = graph->getMaxOffsetMap(transaction); + // MVP: single node table (the common case; multi-table is a follow-up). + if (maxOffsetMap.size() != 1) { + throw BinderException{"GDS_LOUVAIN currently supports single-node-table graphs only."}; + } + const auto tableID = maxOffsetMap.begin()->first; + const auto numNodes = maxOffsetMap.begin()->second; + auto mm = MemoryManager::Get(*clientContext); + auto bindData = input.bindData->constPtrCast(); + + // 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); + + // 2. icebug: zero-copy GraphR over the Arrow CSR, then PLM (Parallel Louvain Method). + NetworKit::GraphR g(numNodes, /*directed=*/false, csr.indices, csr.indptr); + NetworKit::PLM plm(g); + plm.run(); + const NetworKit::Partition& partition = plm.getPartition(); + + // 4. Stream community assignments back through the GDS result pipeline. + auto outputVC = std::make_unique(mm, sharedState, partition); + GDSUtils::runVertexCompute(input.context, GDSDensityState::DENSE, graph, *outputVC); + sharedState->factorizedTablePool.mergeLocalTables(); + return 0; +} + +static constexpr char COMMUNITY_ID_COLUMN_NAME[] = "community_id"; + +static std::unique_ptr bindFunc(main::ClientContext* context, + const TableFuncBindInput* input) { + auto graphName = input->getLiteralVal(0); + auto graphEntry = GDSFunction::bindGraphEntry(*context, graphName); + auto nodeOutput = GDSFunction::bindNodeOutput(*input, graphEntry.getNodeEntries()); + expression_vector columns; + columns.push_back(nodeOutput->constCast().getInternalID()); + columns.push_back( + input->binder->createVariable(COMMUNITY_ID_COLUMN_NAME, LogicalType::INT64())); + return std::make_unique(std::move(columns), std::move(graphEntry), + expression_vector{nodeOutput}, std::move(graphName)); +} + +function_set GDSLouvainFunction::getFunctionSet() { + function_set result; + auto func = std::make_unique(GDSLouvainFunction::name, + std::vector{LogicalTypeID::ANY}); + func->bindFunc = bindFunc; + func->tableFunc = tableFunc; + func->initSharedStateFunc = GDSFunction::initSharedState; + func->initLocalStateFunc = TableFunction::initEmptyLocalState; + func->canParallelFunc = [] { return false; }; + func->getLogicalPlanFunc = GDSFunction::getLogicalPlan; + func->getPhysicalPlanFunc = GDSFunction::getPhysicalPlan; + result.push_back(std::move(func)); + return result; +} + +} // namespace algo_extension +} // namespace lbug diff --git a/algo/src/include/function/algo_function.h b/algo/src/include/function/algo_function.h index 83d72933..84af4502 100644 --- a/algo/src/include/function/algo_function.h +++ b/algo/src/include/function/algo_function.h @@ -67,6 +67,14 @@ struct GDSNode2VecFunction { static function::function_set getFunctionSet(); }; +// icebug (NetworKit)-backed Louvain community detection. Coexists with LOUVAIN for one release +// cycle. +struct GDSLouvainFunction { + static constexpr const char* name = "GDS_LOUVAIN"; + + static function::function_set getFunctionSet(); +}; + struct KCoreDecompositionFunction { static constexpr const char* name = "K_CORE_DECOMPOSITION"; diff --git a/algo/src/main/algo_extension.cpp b/algo/src/main/algo_extension.cpp index 42bd9658..301f5f07 100644 --- a/algo/src/main/algo_extension.cpp +++ b/algo/src/main/algo_extension.cpp @@ -39,6 +39,7 @@ void AlgoExtension::load(main::ClientContext* context) { #if defined(ICEBUG_ENABLED) ExtensionUtils::addTableFunc(db); ExtensionUtils::addTableFunc(db); + ExtensionUtils::addTableFunc(db); #endif ExtensionUtils::addTableFunc(db); ExtensionUtils::addTableFuncAlias(db); diff --git a/algo/test/test_files/gds_louvain.test b/algo/test/test_files/gds_louvain.test new file mode 100644 index 00000000..e8818a45 --- /dev/null +++ b/algo/test/test_files/gds_louvain.test @@ -0,0 +1,46 @@ +-DATASET CSV empty + +-- + +-CASE GDSLouvainTwoCliques +-LOAD_DYNAMIC_EXTENSION algo +-STATEMENT CREATE NODE TABLE Node(id INT64 PRIMARY KEY); +---- ok +-STATEMENT CREATE REL TABLE Edge(FROM Node to Node); +---- ok +-STATEMENT CREATE (u0:Node {id: 0}), + (u1:Node {id: 1}), + (u2:Node {id: 2}), + (u3:Node {id: 3}), + (u4:Node {id: 4}), + (u5:Node {id: 5}), + (u6:Node {id: 6}), + (u7:Node {id: 7}), + (u0)-[:Edge]->(u1), + (u0)-[:Edge]->(u2), + (u1)-[:Edge]->(u2), + (u0)-[:Edge]->(u3), + (u1)-[:Edge]->(u3), + (u2)-[:Edge]->(u3), + (u4)-[:Edge]->(u5), + (u4)-[:Edge]->(u6), + (u5)-[:Edge]->(u6), + (u4)-[:Edge]->(u7), + (u5)-[:Edge]->(u7), + (u6)-[:Edge]->(u7), + (u3)-[:Edge]->(u4); +---- ok +-STATEMENT CALL PROJECT_GRAPH('Graph', ['Node'], ['Edge']) +---- ok +-LOG icebug PLM community IDs are arbitrary and PLM is parallel (nondeterministic), so assert structure not exact IDs: two communities, one per clique, joined by a bridge edge (u3-u4). +-STATEMENT CALL GDS_LOUVAIN('Graph') RETURN COUNT(DISTINCT community_id) +---- 1 +2 +-STATEMENT CALL GDS_LOUVAIN('Graph') WITH node, community_id WHERE node.id IN [0, 1, 2, 3] + RETURN COUNT(DISTINCT community_id) +---- 1 +1 +-STATEMENT CALL GDS_LOUVAIN('Graph') WITH node, community_id WHERE node.id IN [4, 5, 6, 7] + RETURN COUNT(DISTINCT community_id) +---- 1 +1 From d90630c95243c5e9158159b180624423a47785b4 Mon Sep 17 00:00:00 2001 From: Zach Winter Date: Thu, 13 Aug 2026 15:27:39 -0400 Subject: [PATCH 2/3] feat(algo): add GDS_PPR (icebug personalized PageRank via forSources) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Random walk with restart from caller-supplied source nodes: CALL GDS_PPR('G', [rowid, ...]) — teleportation is uniform over the sources (NetworKit::PageRank::forSources, memory-efficient: no n-sized personalization vector), so scores measure standing relative to the anchors rather than globally. Consumes the shared zero-copy CSR bridge. Optional params dampingFactor/tolerance as GDS_PAGE_RANK. Sources validated non-empty and non-negative at bind, range-checked against the projected node count at run time. Tests: star anchored at a leaf (asymmetric distribution vs the plain PageRank tie), and a two-clique trust-closure case — anchored in one clique, mass stays with the anchor's community (0.17-0.29 vs 0.03-0.07 across the bridge) — plus empty-list and out-of-range error cases. Co-Authored-By: Claude Fable 5 --- algo/src/function/CMakeLists.txt | 2 +- algo/src/function/gds_ppr.cpp | 214 ++++++++++++++++++++++ algo/src/include/function/algo_function.h | 8 + algo/src/main/algo_extension.cpp | 1 + algo/test/test_files/gds_ppr.test | 88 +++++++++ 5 files changed, 312 insertions(+), 1 deletion(-) create mode 100644 algo/src/function/gds_ppr.cpp create mode 100644 algo/test/test_files/gds_ppr.test diff --git a/algo/src/function/CMakeLists.txt b/algo/src/function/CMakeLists.txt index e72b942a..0d601eae 100644 --- a/algo/src/function/CMakeLists.txt +++ b/algo/src/function/CMakeLists.txt @@ -14,7 +14,7 @@ if (ICEBUG_ENABLED) # The gds_*.cpp sources include 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_csr_bridge.cpp gds_page_rank.cpp gds_node2vec.cpp - gds_louvain.cpp) + gds_louvain.cpp gds_ppr.cpp) endif () add_library(lbug_algo_function diff --git a/algo/src/function/gds_ppr.cpp b/algo/src/function/gds_ppr.cpp new file mode 100644 index 00000000..fc5e50da --- /dev/null +++ b/algo/src/function/gds_ppr.cpp @@ -0,0 +1,214 @@ +// GDS_PPR — personalized PageRank backed by icebug (NetworKit's PageRank::forSources). +// Random walk with restart: teleportation is restricted to the caller-supplied source nodes +// (uniform over the set), so scores measure standing *relative to those anchors* rather than +// globally. This is the trust-propagation primitive: run it from a viewpoint's attested keys +// and every node inherits a confidence score relative to that viewpoint. +// +// CALL GDS_PPR('G', [rowid, ...]) — sources are node rowids (offsets), the same identity the +// materialized CSR is built on. Optional params: dampingFactor, tolerance (as GDS_PAGE_RANK). +#include "binder/binder.h" +#include "common/exception/binder.h" +#include "common/string_utils.h" +#include "common/types/value/nested.h" +#include "function/algo_function.h" +#include "function/config/page_rank_config.h" +#include "function/gds/gds_utils.h" +#include "function/gds/gds_vertex_compute.h" +#include "function/gds_csr_bridge.h" +#include "function/table/bind_input.h" +#include "processor/execution_context.h" +#include "transaction/transaction.h" +#include +#include +#include + +using namespace lbug::processor; +using namespace lbug::common; +using namespace lbug::binder; +using namespace lbug::storage; +using namespace lbug::graph; +using namespace lbug::function; + +namespace lbug { +namespace algo_extension { + +struct GDSPprOptionalParams final : public function::OptionalParams { + OptionalParam dampingFactor; + OptionalParam tolerance; + + explicit GDSPprOptionalParams(const expression_vector& optionalParams) { + for (auto& optionalParam : optionalParams) { + auto paramName = StringUtils::getLower(optionalParam->getAlias()); + if (paramName == DampingFactor::NAME) { + dampingFactor = function::OptionalParam(optionalParam); + } else if (paramName == Tolerance::NAME) { + tolerance = function::OptionalParam(optionalParam); + } else { + throw BinderException{"Unknown optional parameter: " + optionalParam->getAlias()}; + } + } + } + + GDSPprOptionalParams(OptionalParam dampingFactor, + OptionalParam tolerance) + : dampingFactor{std::move(dampingFactor)}, tolerance{std::move(tolerance)} {} + + void evaluateParams(main::ClientContext* context) override { + dampingFactor.evaluateParam(context); + tolerance.evaluateParam(context); + } + + std::unique_ptr copy() override { + return std::make_unique(dampingFactor, tolerance); + } +}; + +struct GDSPprBindData final : public GDSBindData { + // Projected graph name, for looking up the entry's materialized arrow CSR at run time. + std::string graphName; + // Teleportation targets (node rowids), validated non-empty and non-negative at bind; + // range-checked against the projected node count at run time. + std::vector sources; + + GDSPprBindData(expression_vector columns, graph::NativeGraphEntry graphEntry, + expression_vector output, std::unique_ptr optionalParams, + std::string graphName, std::vector sources) + : GDSBindData{std::move(columns), std::move(graphEntry), std::move(output)}, + graphName{std::move(graphName)}, sources{std::move(sources)} { + this->optionalParams = std::move(optionalParams); + } + + std::unique_ptr copy() const override { + return std::make_unique(*this); + } +}; + +// Emits (node, rank) rows, reading the icebug PPR scores by node offset. +class GDSPprResultVertexCompute : public GDSResultVertexCompute { +public: + GDSPprResultVertexCompute(storage::MemoryManager* mm, GDSFuncSharedState* sharedState, + const std::vector& scores) + : GDSResultVertexCompute{mm, sharedState}, scores{scores} { + nodeIDVector = createVector(LogicalType::INTERNAL_ID()); + rankVector = createVector(LogicalType::DOUBLE()); + } + + void beginOnTableInternal(table_id_t) override {} + + void vertexCompute(offset_t startOffset, offset_t endOffset, table_id_t tableID) override { + for (auto i = startOffset; i < endOffset; ++i) { + if (skip(i)) { + continue; + } + nodeIDVector->setValue(0, nodeID_t{i, tableID}); + rankVector->setValue(0, i < scores.size() ? scores[i] : 0.0); + localFT->append(vectors); + } + } + + std::unique_ptr copy() override { + return std::make_unique(mm, sharedState, scores); + } + +private: + const std::vector& scores; + std::unique_ptr nodeIDVector; + std::unique_ptr rankVector; +}; + +static offset_t tableFunc(const TableFuncInput& input, TableFuncOutput&) { + auto clientContext = input.context->clientContext; + auto transaction = transaction::Transaction::Get(*clientContext); + auto sharedState = input.sharedState->ptrCast(); + auto graph = sharedState->graph.get(); + auto maxOffsetMap = graph->getMaxOffsetMap(transaction); + // MVP: single node table (the common case; multi-table is a follow-up). + if (maxOffsetMap.size() != 1) { + throw BinderException{"GDS_PPR currently supports single-node-table graphs only."}; + } + const auto tableID = maxOffsetMap.begin()->first; + const auto numNodes = maxOffsetMap.begin()->second; + auto mm = MemoryManager::Get(*clientContext); + auto bindData = input.bindData->constPtrCast(); + auto& config = bindData->optionalParams->constCast(); + for (const auto source : bindData->sources) { + if (source >= numNodes) { + throw BinderException{"GDS_PPR source node offset " + std::to_string(source) + + " is out of range: the projected graph has " + + std::to_string(numNodes) + " nodes."}; + } + } + + // 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); + + // 2. icebug: zero-copy GraphR, then k-source personalized PageRank (uniform teleportation + // over the sources; memory-efficient — no n-sized personalization vector). + NetworKit::GraphR g(numNodes, /*directed=*/false, csr.indices, csr.indptr); + const std::vector sources{bindData->sources.begin(), bindData->sources.end()}; + auto pr = NetworKit::PageRank::forSources(g, sources, config.dampingFactor.getParamVal(), + config.tolerance.getParamVal()); + pr.run(); + const std::vector& scores = pr.scores(); + + // 3. Stream scores back through the GDS result pipeline. + auto outputVC = std::make_unique(mm, sharedState, scores); + GDSUtils::runVertexCompute(input.context, GDSDensityState::DENSE, graph, *outputVC); + sharedState->factorizedTablePool.mergeLocalTables(); + return 0; +} + +static constexpr char RANK_COLUMN_NAME[] = "rank"; + +static std::vector extractSources(const Value& value) { + value.validateType(LogicalTypeID::LIST); + std::vector sources; + sources.reserve(NestedVal::getChildrenSize(&value)); + for (auto i = 0u; i < NestedVal::getChildrenSize(&value); ++i) { + const auto* child = NestedVal::getChildVal(&value, i); + child->validateType(LogicalTypeID::INT64); + const auto source = child->getValue(); + if (source < 0) { + throw BinderException{"GDS_PPR source node offsets must be non-negative."}; + } + sources.push_back(static_cast(source)); + } + if (sources.empty()) { + throw BinderException{"GDS_PPR requires at least one source node."}; + } + return sources; +} + +static std::unique_ptr bindFunc(main::ClientContext* context, + const TableFuncBindInput* input) { + auto graphName = input->getLiteralVal(0); + auto sources = extractSources(input->getValue(1)); + auto graphEntry = GDSFunction::bindGraphEntry(*context, graphName); + auto nodeOutput = GDSFunction::bindNodeOutput(*input, graphEntry.getNodeEntries()); + expression_vector columns; + 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), + expression_vector{nodeOutput}, + std::make_unique(input->optionalParamsLegacy), std::move(graphName), + std::move(sources)); +} + +function_set GDSPprFunction::getFunctionSet() { + function_set result; + auto func = std::make_unique(GDSPprFunction::name, + std::vector{LogicalTypeID::ANY, LogicalTypeID::ANY}); + func->bindFunc = bindFunc; + func->tableFunc = tableFunc; + func->initSharedStateFunc = GDSFunction::initSharedState; + func->initLocalStateFunc = TableFunction::initEmptyLocalState; + func->canParallelFunc = [] { return false; }; + func->getLogicalPlanFunc = GDSFunction::getLogicalPlan; + func->getPhysicalPlanFunc = GDSFunction::getPhysicalPlan; + result.push_back(std::move(func)); + return result; +} + +} // namespace algo_extension +} // namespace lbug diff --git a/algo/src/include/function/algo_function.h b/algo/src/include/function/algo_function.h index 84af4502..6a189ff5 100644 --- a/algo/src/include/function/algo_function.h +++ b/algo/src/include/function/algo_function.h @@ -75,6 +75,14 @@ struct GDSLouvainFunction { static function::function_set getFunctionSet(); }; +// icebug (NetworKit)-backed personalized PageRank (random walk with restart from caller-supplied +// source nodes) — scores measure standing relative to the sources, not globally. +struct GDSPprFunction { + static constexpr const char* name = "GDS_PPR"; + + static function::function_set getFunctionSet(); +}; + struct KCoreDecompositionFunction { static constexpr const char* name = "K_CORE_DECOMPOSITION"; diff --git a/algo/src/main/algo_extension.cpp b/algo/src/main/algo_extension.cpp index 301f5f07..be1f191f 100644 --- a/algo/src/main/algo_extension.cpp +++ b/algo/src/main/algo_extension.cpp @@ -40,6 +40,7 @@ void AlgoExtension::load(main::ClientContext* context) { ExtensionUtils::addTableFunc(db); ExtensionUtils::addTableFunc(db); ExtensionUtils::addTableFunc(db); + ExtensionUtils::addTableFunc(db); #endif ExtensionUtils::addTableFunc(db); ExtensionUtils::addTableFuncAlias(db); diff --git a/algo/test/test_files/gds_ppr.test b/algo/test/test_files/gds_ppr.test new file mode 100644 index 00000000..88154444 --- /dev/null +++ b/algo/test/test_files/gds_ppr.test @@ -0,0 +1,88 @@ +-DATASET CSV empty + +-- + +-CASE GDSPprStarFromLeaf +-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 MATCH (x:N{id:3}), (y:N{id:0}) CREATE (x)-[:E]->(y) +---- ok +-STATEMENT CALL PROJECT_GRAPH('G', ['N'], ['E']) +---- ok +-LOG Personalized from leaf 1: teleportation returns to the anchor only, so scores are relative +-LOG to it — the anchor and the hub dominate, the symmetric far leaves trail. (Contrast plain +-LOG GDS_PAGE_RANK, where all three leaves tie.) +-STATEMENT CALL GDS_PPR('G', [1]) RETURN node.id, rank ORDER BY node.id +---- 4 +0|0.459459 +1|0.280180 +2|0.130180 +3|0.130180 +-LOG Empty source list is a binder error. +-STATEMENT CALL GDS_PPR('G', []) RETURN node.id, rank +---- error +Binder exception: GDS_PPR requires at least one source node. +-LOG Out-of-range source is an error naming the offending offset. +-STATEMENT CALL GDS_PPR('G', [99]) RETURN node.id, rank +---- error +Binder exception: GDS_PPR source node offset 99 is out of range: the projected graph has 4 nodes. + +-CASE GDSPprTrustClosure +-LOAD_DYNAMIC_EXTENSION algo +-STATEMENT CREATE NODE TABLE M(id INT64 PRIMARY KEY) +---- ok +-STATEMENT CREATE REL TABLE F(FROM M TO M) +---- ok +-STATEMENT CREATE (:M{id:0}),(:M{id:1}),(:M{id:2}),(:M{id:3}),(:M{id:4}),(:M{id:5}),(:M{id:6}),(:M{id:7}) +---- ok +-STATEMENT MATCH (x:M{id:0}),(y:M{id:1}) CREATE (x)-[:F]->(y) +---- ok +-STATEMENT MATCH (x:M{id:0}),(y:M{id:2}) CREATE (x)-[:F]->(y) +---- ok +-STATEMENT MATCH (x:M{id:1}),(y:M{id:2}) CREATE (x)-[:F]->(y) +---- ok +-STATEMENT MATCH (x:M{id:0}),(y:M{id:3}) CREATE (x)-[:F]->(y) +---- ok +-STATEMENT MATCH (x:M{id:1}),(y:M{id:3}) CREATE (x)-[:F]->(y) +---- ok +-STATEMENT MATCH (x:M{id:2}),(y:M{id:3}) CREATE (x)-[:F]->(y) +---- ok +-STATEMENT MATCH (x:M{id:4}),(y:M{id:5}) CREATE (x)-[:F]->(y) +---- ok +-STATEMENT MATCH (x:M{id:4}),(y:M{id:6}) CREATE (x)-[:F]->(y) +---- ok +-STATEMENT MATCH (x:M{id:5}),(y:M{id:6}) CREATE (x)-[:F]->(y) +---- ok +-STATEMENT MATCH (x:M{id:4}),(y:M{id:7}) CREATE (x)-[:F]->(y) +---- ok +-STATEMENT MATCH (x:M{id:5}),(y:M{id:7}) CREATE (x)-[:F]->(y) +---- ok +-STATEMENT MATCH (x:M{id:6}),(y:M{id:7}) CREATE (x)-[:F]->(y) +---- ok +-STATEMENT MATCH (x:M{id:3}),(y:M{id:4}) CREATE (x)-[:F]->(y) +---- ok +-STATEMENT CALL PROJECT_GRAPH('GT', ['M'], ['F']) +---- ok +-LOG Two 4-cliques joined by a bridge (3-4). Anchored at node 0 in clique A, trust mass stays +-LOG in the anchor's community: clique A scores 0.17-0.29, clique B 0.03-0.07 — the closure +-LOG boundary is visible in the distribution. This is the confidence primitive for +-LOG viewpoint-based trust: scores are relative to the anchors, not global. +-STATEMENT CALL GDS_PPR('GT', [0]) RETURN node.id, rank ORDER BY node.id +---- 8 +0|0.288406 +1|0.171523 +2|0.171523 +3|0.193927 +4|0.070664 +5|0.034653 +6|0.034653 +7|0.034653 From d9ece899054c109ea8b1e32540b13b0eb99b8a83 Mon Sep 17 00:00:00 2001 From: Zach Winter Date: Thu, 13 Aug 2026 15:33:27 -0400 Subject: [PATCH 3/3] refactor(algo): descriptor templates for GDS functions + GDS_LEIDEN MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per review: the per-algorithm boilerplate is gone. GDSPerNodeScalarAlgo (gds_algo_template.h) supplies bind, the shared zero-copy CSR bridge, GraphR construction, and vertex-compute streaming for the whole per-node-scalar equivalence class; an algorithm is now a descriptor — name, output column, optional params, extra-positional-arg policy (NoExtraArgs / SourceListArg), and the NetworKit invocation. GDS_PAGE_RANK, GDS_LOUVAIN, and GDS_PPR are rewritten as descriptors (unchanged CALL surfaces, unchanged test expectations). GDS_LEIDEN is the first new algorithm to ride the template: ParallelLeidenView with gamma/iterations optional params, ~100 lines including both param structs. Per-node-vector outputs (GDS_NODE2VEC) stay bespoke until a second member of that class motivates the template. Co-Authored-By: Claude Fable 5 --- algo/src/function/CMakeLists.txt | 2 +- algo/src/function/gds_leiden.cpp | 100 ++++++++ algo/src/function/gds_louvain.cpp | 155 +++---------- algo/src/function/gds_page_rank.cpp | 144 ++---------- algo/src/function/gds_ppr.cpp | 176 ++------------ algo/src/include/function/algo_function.h | 8 + algo/src/include/function/gds_algo_template.h | 219 ++++++++++++++++++ algo/src/main/algo_extension.cpp | 1 + algo/test/test_files/gds_leiden.test | 54 +++++ 9 files changed, 451 insertions(+), 408 deletions(-) create mode 100644 algo/src/function/gds_leiden.cpp create mode 100644 algo/src/include/function/gds_algo_template.h create mode 100644 algo/test/test_files/gds_leiden.test diff --git a/algo/src/function/CMakeLists.txt b/algo/src/function/CMakeLists.txt index 0d601eae..7ce7bd83 100644 --- a/algo/src/function/CMakeLists.txt +++ b/algo/src/function/CMakeLists.txt @@ -14,7 +14,7 @@ if (ICEBUG_ENABLED) # The gds_*.cpp sources include 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_csr_bridge.cpp gds_page_rank.cpp gds_node2vec.cpp - gds_louvain.cpp gds_ppr.cpp) + gds_louvain.cpp gds_leiden.cpp gds_ppr.cpp) endif () add_library(lbug_algo_function diff --git a/algo/src/function/gds_leiden.cpp b/algo/src/function/gds_leiden.cpp new file mode 100644 index 00000000..063f01f3 --- /dev/null +++ b/algo/src/function/gds_leiden.cpp @@ -0,0 +1,100 @@ +// GDS_LEIDEN — Leiden community detection backed by icebug (NetworKit's ParallelLeidenView). +// Higher partition quality and performance than Louvain/PLM on most graphs — prefer this for +// new work. All bridge plumbing lives in GDSPerNodeScalarAlgo (gds_algo_template.h); this file +// is the descriptor: name, output column, optional params, and the NetworKit invocation. +#include "common/string_utils.h" +#include "function/algo_function.h" +#include "function/gds_algo_template.h" +#include +#include + +using namespace lbug::common; +using namespace lbug::binder; +using namespace lbug::function; + +namespace lbug { +namespace algo_extension { + +struct LeidenGamma { + // Resolution parameter: > 1 favors more/smaller communities, < 1 fewer/larger. + static constexpr const char* NAME = "gamma"; + static constexpr common::LogicalTypeID TYPE = common::LogicalTypeID::DOUBLE; + static constexpr double DEFAULT_VALUE = 1.0; + + static void validate(double gamma) { + if (gamma <= 0) { + throw BinderException{"Gamma must be positive."}; + } + } +}; + +struct LeidenIterations { + static constexpr const char* NAME = "iterations"; + static constexpr common::LogicalTypeID TYPE = common::LogicalTypeID::INT64; + static constexpr int64_t DEFAULT_VALUE = 3; + + static void validate(int64_t iterations) { + if (iterations <= 0) { + throw BinderException{"Iterations must be positive."}; + } + } +}; + +struct GDSLeidenOptionalParams final : public function::OptionalParams { + OptionalParam gamma; + OptionalParam iterations; + + explicit GDSLeidenOptionalParams(const expression_vector& optionalParams) { + for (auto& optionalParam : optionalParams) { + auto paramName = StringUtils::getLower(optionalParam->getAlias()); + if (paramName == LeidenGamma::NAME) { + gamma = function::OptionalParam(optionalParam); + } else if (paramName == LeidenIterations::NAME) { + iterations = function::OptionalParam(optionalParam); + } else { + throw BinderException{"Unknown optional parameter: " + optionalParam->getAlias()}; + } + } + } + + GDSLeidenOptionalParams(OptionalParam gamma, + OptionalParam iterations) + : gamma{std::move(gamma)}, iterations{std::move(iterations)} {} + + void evaluateParams(main::ClientContext* context) override { + gamma.evaluateParam(context); + iterations.evaluateParam(context); + } + + std::unique_ptr copy() override { + return std::make_unique(gamma, iterations); + } +}; + +struct GDSLeidenDesc { + static constexpr const char* name = GDSLeidenFunction::name; + static constexpr const char* OUTPUT_COLUMN = "community_id"; + using OutputType = int64_t; + using Params = GDSLeidenOptionalParams; + using Extra = NoExtraArgs; + + static std::vector run(const NetworKit::GraphR& g, offset_t numNodes, + const Params& params, const Extra::type&) { + NetworKit::ParallelLeidenView leiden(g, static_cast(params.iterations.getParamVal()), + /*randomize=*/true, params.gamma.getParamVal()); + leiden.run(); + const auto& partition = leiden.getPartition(); + std::vector communityIds(numNodes, -1); + for (offset_t i = 0; i < numNodes && i < partition.numberOfElements(); ++i) { + communityIds[i] = static_cast(partition[i]); + } + return communityIds; + } +}; + +function_set GDSLeidenFunction::getFunctionSet() { + return GDSPerNodeScalarAlgo::getFunctionSet(); +} + +} // namespace algo_extension +} // namespace lbug diff --git a/algo/src/function/gds_louvain.cpp b/algo/src/function/gds_louvain.cpp index ba21d229..4a7c0ee3 100644 --- a/algo/src/function/gds_louvain.cpp +++ b/algo/src/function/gds_louvain.cpp @@ -1,146 +1,53 @@ -// GDS_LOUVAIN — Louvain community detection backed by icebug (a NetworKit fork with zero-copy -// Arrow CSR ingest). Same CALL surface as the hand-rolled LOUVAIN, but the compute is delegated -// to libnetworkit: we materialize the projected graph's adjacency as CSR, build a -// NetworKit::GraphR over Arrow buffers, run NetworKit::PLM (Parallel Louvain Method), and stream -// the resulting community assignments back through the GDS result pipeline. -// -// Part of the icebug bridge (adsharma-invited): the `algo` extension keeps its existing algos for -// one release cycle; the icebug-backed ones live alongside under GDS_* names. -#include "binder/binder.h" -#include "common/exception/binder.h" +// GDS_LOUVAIN — Louvain community detection backed by icebug (NetworKit's PLM, Parallel +// Louvain Method). Coexists with the hand-rolled LOUVAIN for one release cycle. All bridge +// plumbing lives in GDSPerNodeScalarAlgo (gds_algo_template.h); this file is the descriptor. +// Prefer GDS_LEIDEN for new work — better partition quality on most graphs. #include "function/algo_function.h" -#include "function/gds/gds_utils.h" -#include "function/gds/gds_vertex_compute.h" -#include "function/gds_csr_bridge.h" -#include "function/table/bind_input.h" -#include "processor/execution_context.h" -#include "transaction/transaction.h" -#include +#include "function/gds_algo_template.h" #include -#include #include -using namespace lbug::processor; using namespace lbug::common; -using namespace lbug::binder; -using namespace lbug::storage; -using namespace lbug::graph; using namespace lbug::function; namespace lbug { namespace algo_extension { -// Emits (node, community_id) rows, reading the icebug PLM partition by node offset. -class GDSLouvainResultVertexCompute : public GDSResultVertexCompute { -public: - GDSLouvainResultVertexCompute(storage::MemoryManager* mm, GDSFuncSharedState* sharedState, - const NetworKit::Partition& partition) - : GDSResultVertexCompute{mm, sharedState}, partition{partition} { - nodeIDVector = createVector(LogicalType::INTERNAL_ID()); - communityIDVector = createVector(LogicalType::INT64()); - } - - void beginOnTableInternal(table_id_t) override {} - - void vertexCompute(offset_t startOffset, offset_t endOffset, table_id_t tableID) override { - for (auto i = startOffset; i < endOffset; ++i) { - if (skip(i)) { - continue; - } - nodeIDVector->setValue(0, nodeID_t{i, tableID}); - communityIDVector->setValue(0, - i < partition.numberOfElements() ? static_cast(partition[i]) : -1); - localFT->append(vectors); +// No optional params (PLM defaults). +struct GDSLouvainOptionalParams final : public function::OptionalParams { + explicit GDSLouvainOptionalParams(const binder::expression_vector& optionalParams) { + if (!optionalParams.empty()) { + throw BinderException{"Unknown optional parameter: " + optionalParams[0]->getAlias()}; } } - - std::unique_ptr copy() override { - return std::make_unique(mm, sharedState, partition); + void evaluateParams(main::ClientContext*) override {} + std::unique_ptr copy() override { + return std::make_unique(binder::expression_vector{}); } - -private: - const NetworKit::Partition& partition; - std::unique_ptr nodeIDVector; - std::unique_ptr communityIDVector; }; -// CSR construction lives in the shared bridge (gds_csr_bridge.cpp): zero-copy from the graph -// entry's materialized arrow CSR when available, storage-scan fallback otherwise. - -struct GDSLouvainBindData final : public GDSBindData { - // Projected graph name, for looking up the entry's materialized arrow CSR at run time. - std::string graphName; - - GDSLouvainBindData(expression_vector columns, graph::NativeGraphEntry graphEntry, - expression_vector output, std::string graphName) - : GDSBindData{std::move(columns), std::move(graphEntry), std::move(output)}, - graphName{std::move(graphName)} {} - - std::unique_ptr copy() const override { - return std::make_unique(*this); +struct GDSLouvainDesc { + static constexpr const char* name = GDSLouvainFunction::name; + static constexpr const char* OUTPUT_COLUMN = "community_id"; + using OutputType = int64_t; + using Params = GDSLouvainOptionalParams; + using Extra = NoExtraArgs; + + static std::vector run(const NetworKit::GraphR& g, offset_t numNodes, const Params&, + const Extra::type&) { + NetworKit::PLM plm(g); + plm.run(); + const auto& partition = plm.getPartition(); + std::vector communityIds(numNodes, -1); + for (offset_t i = 0; i < numNodes && i < partition.numberOfElements(); ++i) { + communityIds[i] = static_cast(partition[i]); + } + return communityIds; } }; -static offset_t tableFunc(const TableFuncInput& input, TableFuncOutput&) { - auto clientContext = input.context->clientContext; - auto transaction = transaction::Transaction::Get(*clientContext); - auto sharedState = input.sharedState->ptrCast(); - auto graph = sharedState->graph.get(); - auto maxOffsetMap = graph->getMaxOffsetMap(transaction); - // MVP: single node table (the common case; multi-table is a follow-up). - if (maxOffsetMap.size() != 1) { - throw BinderException{"GDS_LOUVAIN currently supports single-node-table graphs only."}; - } - const auto tableID = maxOffsetMap.begin()->first; - const auto numNodes = maxOffsetMap.begin()->second; - auto mm = MemoryManager::Get(*clientContext); - auto bindData = input.bindData->constPtrCast(); - - // 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); - - // 2. icebug: zero-copy GraphR over the Arrow CSR, then PLM (Parallel Louvain Method). - NetworKit::GraphR g(numNodes, /*directed=*/false, csr.indices, csr.indptr); - NetworKit::PLM plm(g); - plm.run(); - const NetworKit::Partition& partition = plm.getPartition(); - - // 4. Stream community assignments back through the GDS result pipeline. - auto outputVC = std::make_unique(mm, sharedState, partition); - GDSUtils::runVertexCompute(input.context, GDSDensityState::DENSE, graph, *outputVC); - sharedState->factorizedTablePool.mergeLocalTables(); - return 0; -} - -static constexpr char COMMUNITY_ID_COLUMN_NAME[] = "community_id"; - -static std::unique_ptr bindFunc(main::ClientContext* context, - const TableFuncBindInput* input) { - auto graphName = input->getLiteralVal(0); - auto graphEntry = GDSFunction::bindGraphEntry(*context, graphName); - auto nodeOutput = GDSFunction::bindNodeOutput(*input, graphEntry.getNodeEntries()); - expression_vector columns; - columns.push_back(nodeOutput->constCast().getInternalID()); - columns.push_back( - input->binder->createVariable(COMMUNITY_ID_COLUMN_NAME, LogicalType::INT64())); - return std::make_unique(std::move(columns), std::move(graphEntry), - expression_vector{nodeOutput}, std::move(graphName)); -} - function_set GDSLouvainFunction::getFunctionSet() { - function_set result; - auto func = std::make_unique(GDSLouvainFunction::name, - std::vector{LogicalTypeID::ANY}); - func->bindFunc = bindFunc; - func->tableFunc = tableFunc; - func->initSharedStateFunc = GDSFunction::initSharedState; - func->initLocalStateFunc = TableFunction::initEmptyLocalState; - func->canParallelFunc = [] { return false; }; - func->getLogicalPlanFunc = GDSFunction::getLogicalPlan; - func->getPhysicalPlanFunc = GDSFunction::getPhysicalPlan; - result.push_back(std::move(func)); - return result; + return GDSPerNodeScalarAlgo::getFunctionSet(); } } // namespace algo_extension diff --git a/algo/src/function/gds_page_rank.cpp b/algo/src/function/gds_page_rank.cpp index 32b84653..4fe7fd0e 100644 --- a/algo/src/function/gds_page_rank.cpp +++ b/algo/src/function/gds_page_rank.cpp @@ -1,31 +1,16 @@ // GDS_PAGE_RANK — PageRank backed by icebug (a NetworKit fork with zero-copy Arrow CSR ingest). -// Same CALL surface as the hand-rolled PAGE_RANK, but the compute is delegated to libnetworkit: -// we materialize the projected graph's adjacency as CSR, build a NetworKit::GraphR over Arrow -// buffers, run NetworKit::PageRank, and stream the scores back through the GDS result pipeline. -// -// Part of the icebug bridge (adsharma-invited): the `algo` extension keeps its existing algos for -// one release cycle; the icebug-backed ones live alongside under GDS_* names. -#include "binder/binder.h" -#include "common/exception/binder.h" +// Same CALL surface as the hand-rolled PAGE_RANK, but the compute is delegated to libnetworkit. +// All bridge plumbing lives in GDSPerNodeScalarAlgo (gds_algo_template.h); this file is the +// algorithm descriptor: name, output column, optional params, and the NetworKit invocation. #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" -#include "function/gds/gds_vertex_compute.h" -#include "function/table/bind_input.h" -#include "processor/execution_context.h" -#include "transaction/transaction.h" -#include +#include "function/gds_algo_template.h" #include -#include -using namespace lbug::processor; using namespace lbug::common; using namespace lbug::binder; -using namespace lbug::storage; -using namespace lbug::graph; using namespace lbug::function; namespace lbug { @@ -68,117 +53,24 @@ 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, std::string graphName) - : GDSBindData{std::move(columns), std::move(graphEntry), expression_vector{nodeOutput}}, - graphName{std::move(graphName)} { - this->optionalParams = std::move(optionalParams); - } - - std::unique_ptr copy() const override { - return std::make_unique(*this); - } -}; - -// Emits (node, rank) rows, reading the icebug PageRank scores by node offset. -class GDSPageRankResultVertexCompute : public GDSResultVertexCompute { -public: - GDSPageRankResultVertexCompute(storage::MemoryManager* mm, GDSFuncSharedState* sharedState, - const std::vector& scores) - : GDSResultVertexCompute{mm, sharedState}, scores{scores} { - nodeIDVector = createVector(LogicalType::INTERNAL_ID()); - rankVector = createVector(LogicalType::DOUBLE()); - } - - void beginOnTableInternal(table_id_t) override {} - - void vertexCompute(offset_t startOffset, offset_t endOffset, table_id_t tableID) override { - for (auto i = startOffset; i < endOffset; ++i) { - if (skip(i)) { - continue; - } - nodeIDVector->setValue(0, nodeID_t{i, tableID}); - rankVector->setValue(0, i < scores.size() ? scores[i] : 0.0); - localFT->append(vectors); - } +struct GDSPageRankDesc { + static constexpr const char* name = GDSPageRankFunction::name; + static constexpr const char* OUTPUT_COLUMN = "rank"; + using OutputType = double; + using Params = GDSPageRankOptionalParams; + using Extra = NoExtraArgs; + + static std::vector run(const NetworKit::GraphR& g, offset_t /*numNodes*/, + const Params& params, const Extra::type&) { + NetworKit::PageRank pr(g, params.dampingFactor.getParamVal(), + params.tolerance.getParamVal()); + pr.run(); + return pr.scores(); } - - std::unique_ptr copy() override { - return std::make_unique(mm, sharedState, scores); - } - -private: - const std::vector& scores; - std::unique_ptr nodeIDVector; - std::unique_ptr rankVector; }; -static offset_t tableFunc(const TableFuncInput& input, TableFuncOutput&) { - auto clientContext = input.context->clientContext; - auto transaction = transaction::Transaction::Get(*clientContext); - auto sharedState = input.sharedState->ptrCast(); - auto graph = sharedState->graph.get(); - auto maxOffsetMap = graph->getMaxOffsetMap(transaction); - // MVP: single node table (the common case; multi-table is a follow-up). - if (maxOffsetMap.size() != 1) { - throw BinderException{"GDS_PAGE_RANK currently supports single-node-table graphs only."}; - } - const auto tableID = maxOffsetMap.begin()->first; - const auto numNodes = maxOffsetMap.begin()->second; - auto mm = MemoryManager::Get(*clientContext); - auto bindData = input.bindData->constPtrCast(); - auto& config = bindData->optionalParams->constCast(); - - // 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); - - // 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(); - - // 4. Stream scores back through the GDS result pipeline. - auto outputVC = std::make_unique(mm, sharedState, scores); - GDSUtils::runVertexCompute(input.context, GDSDensityState::DENSE, graph, *outputVC); - sharedState->factorizedTablePool.mergeLocalTables(); - return 0; -} - -static constexpr char RANK_COLUMN_NAME[] = "rank"; - -static std::unique_ptr bindFunc(main::ClientContext* context, - const TableFuncBindInput* input) { - auto graphName = input->getLiteralVal(0); - auto graphEntry = GDSFunction::bindGraphEntry(*context, graphName); - auto nodeOutput = GDSFunction::bindNodeOutput(*input, graphEntry.getNodeEntries()); - expression_vector columns; - 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), - std::move(graphName)); -} - function_set GDSPageRankFunction::getFunctionSet() { - function_set result; - auto func = std::make_unique(GDSPageRankFunction::name, - std::vector{LogicalTypeID::ANY}); - func->bindFunc = bindFunc; - func->tableFunc = tableFunc; - func->initSharedStateFunc = GDSFunction::initSharedState; - func->initLocalStateFunc = TableFunction::initEmptyLocalState; - func->canParallelFunc = [] { return false; }; - func->getLogicalPlanFunc = GDSFunction::getLogicalPlan; - func->getPhysicalPlanFunc = GDSFunction::getPhysicalPlan; - result.push_back(std::move(func)); - return result; + return GDSPerNodeScalarAlgo::getFunctionSet(); } } // namespace algo_extension diff --git a/algo/src/function/gds_ppr.cpp b/algo/src/function/gds_ppr.cpp index fc5e50da..1e2ba493 100644 --- a/algo/src/function/gds_ppr.cpp +++ b/algo/src/function/gds_ppr.cpp @@ -4,29 +4,17 @@ // globally. This is the trust-propagation primitive: run it from a viewpoint's attested keys // and every node inherits a confidence score relative to that viewpoint. // -// CALL GDS_PPR('G', [rowid, ...]) — sources are node rowids (offsets), the same identity the -// materialized CSR is built on. Optional params: dampingFactor, tolerance (as GDS_PAGE_RANK). -#include "binder/binder.h" -#include "common/exception/binder.h" +// CALL GDS_PPR('G', [rowid, ...]) — sources are node rowids, the same identity the +// materialized CSR is built on. All bridge plumbing lives in GDSPerNodeScalarAlgo +// (gds_algo_template.h); this file is the descriptor. #include "common/string_utils.h" -#include "common/types/value/nested.h" #include "function/algo_function.h" #include "function/config/page_rank_config.h" -#include "function/gds/gds_utils.h" -#include "function/gds/gds_vertex_compute.h" -#include "function/gds_csr_bridge.h" -#include "function/table/bind_input.h" -#include "processor/execution_context.h" -#include "transaction/transaction.h" -#include +#include "function/gds_algo_template.h" #include -#include -using namespace lbug::processor; using namespace lbug::common; using namespace lbug::binder; -using namespace lbug::storage; -using namespace lbug::graph; using namespace lbug::function; namespace lbug { @@ -63,151 +51,25 @@ struct GDSPprOptionalParams final : public function::OptionalParams { } }; -struct GDSPprBindData final : public GDSBindData { - // Projected graph name, for looking up the entry's materialized arrow CSR at run time. - std::string graphName; - // Teleportation targets (node rowids), validated non-empty and non-negative at bind; - // range-checked against the projected node count at run time. - std::vector sources; - - GDSPprBindData(expression_vector columns, graph::NativeGraphEntry graphEntry, - expression_vector output, std::unique_ptr optionalParams, - std::string graphName, std::vector sources) - : GDSBindData{std::move(columns), std::move(graphEntry), std::move(output)}, - graphName{std::move(graphName)}, sources{std::move(sources)} { - this->optionalParams = std::move(optionalParams); - } - - std::unique_ptr copy() const override { - return std::make_unique(*this); +struct GDSPprDesc { + static constexpr const char* name = GDSPprFunction::name; + static constexpr const char* OUTPUT_COLUMN = "rank"; + using OutputType = double; + using Params = GDSPprOptionalParams; + using Extra = SourceListArg; + + static std::vector run(const NetworKit::GraphR& g, offset_t /*numNodes*/, + const Params& params, const Extra::type& sources) { + const std::vector nkSources{sources.begin(), sources.end()}; + auto pr = NetworKit::PageRank::forSources(g, nkSources, params.dampingFactor.getParamVal(), + params.tolerance.getParamVal()); + pr.run(); + return pr.scores(); } }; -// Emits (node, rank) rows, reading the icebug PPR scores by node offset. -class GDSPprResultVertexCompute : public GDSResultVertexCompute { -public: - GDSPprResultVertexCompute(storage::MemoryManager* mm, GDSFuncSharedState* sharedState, - const std::vector& scores) - : GDSResultVertexCompute{mm, sharedState}, scores{scores} { - nodeIDVector = createVector(LogicalType::INTERNAL_ID()); - rankVector = createVector(LogicalType::DOUBLE()); - } - - void beginOnTableInternal(table_id_t) override {} - - void vertexCompute(offset_t startOffset, offset_t endOffset, table_id_t tableID) override { - for (auto i = startOffset; i < endOffset; ++i) { - if (skip(i)) { - continue; - } - nodeIDVector->setValue(0, nodeID_t{i, tableID}); - rankVector->setValue(0, i < scores.size() ? scores[i] : 0.0); - localFT->append(vectors); - } - } - - std::unique_ptr copy() override { - return std::make_unique(mm, sharedState, scores); - } - -private: - const std::vector& scores; - std::unique_ptr nodeIDVector; - std::unique_ptr rankVector; -}; - -static offset_t tableFunc(const TableFuncInput& input, TableFuncOutput&) { - auto clientContext = input.context->clientContext; - auto transaction = transaction::Transaction::Get(*clientContext); - auto sharedState = input.sharedState->ptrCast(); - auto graph = sharedState->graph.get(); - auto maxOffsetMap = graph->getMaxOffsetMap(transaction); - // MVP: single node table (the common case; multi-table is a follow-up). - if (maxOffsetMap.size() != 1) { - throw BinderException{"GDS_PPR currently supports single-node-table graphs only."}; - } - const auto tableID = maxOffsetMap.begin()->first; - const auto numNodes = maxOffsetMap.begin()->second; - auto mm = MemoryManager::Get(*clientContext); - auto bindData = input.bindData->constPtrCast(); - auto& config = bindData->optionalParams->constCast(); - for (const auto source : bindData->sources) { - if (source >= numNodes) { - throw BinderException{"GDS_PPR source node offset " + std::to_string(source) + - " is out of range: the projected graph has " + - std::to_string(numNodes) + " nodes."}; - } - } - - // 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); - - // 2. icebug: zero-copy GraphR, then k-source personalized PageRank (uniform teleportation - // over the sources; memory-efficient — no n-sized personalization vector). - NetworKit::GraphR g(numNodes, /*directed=*/false, csr.indices, csr.indptr); - const std::vector sources{bindData->sources.begin(), bindData->sources.end()}; - auto pr = NetworKit::PageRank::forSources(g, sources, config.dampingFactor.getParamVal(), - config.tolerance.getParamVal()); - pr.run(); - const std::vector& scores = pr.scores(); - - // 3. Stream scores back through the GDS result pipeline. - auto outputVC = std::make_unique(mm, sharedState, scores); - GDSUtils::runVertexCompute(input.context, GDSDensityState::DENSE, graph, *outputVC); - sharedState->factorizedTablePool.mergeLocalTables(); - return 0; -} - -static constexpr char RANK_COLUMN_NAME[] = "rank"; - -static std::vector extractSources(const Value& value) { - value.validateType(LogicalTypeID::LIST); - std::vector sources; - sources.reserve(NestedVal::getChildrenSize(&value)); - for (auto i = 0u; i < NestedVal::getChildrenSize(&value); ++i) { - const auto* child = NestedVal::getChildVal(&value, i); - child->validateType(LogicalTypeID::INT64); - const auto source = child->getValue(); - if (source < 0) { - throw BinderException{"GDS_PPR source node offsets must be non-negative."}; - } - sources.push_back(static_cast(source)); - } - if (sources.empty()) { - throw BinderException{"GDS_PPR requires at least one source node."}; - } - return sources; -} - -static std::unique_ptr bindFunc(main::ClientContext* context, - const TableFuncBindInput* input) { - auto graphName = input->getLiteralVal(0); - auto sources = extractSources(input->getValue(1)); - auto graphEntry = GDSFunction::bindGraphEntry(*context, graphName); - auto nodeOutput = GDSFunction::bindNodeOutput(*input, graphEntry.getNodeEntries()); - expression_vector columns; - 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), - expression_vector{nodeOutput}, - std::make_unique(input->optionalParamsLegacy), std::move(graphName), - std::move(sources)); -} - function_set GDSPprFunction::getFunctionSet() { - function_set result; - auto func = std::make_unique(GDSPprFunction::name, - std::vector{LogicalTypeID::ANY, LogicalTypeID::ANY}); - func->bindFunc = bindFunc; - func->tableFunc = tableFunc; - func->initSharedStateFunc = GDSFunction::initSharedState; - func->initLocalStateFunc = TableFunction::initEmptyLocalState; - func->canParallelFunc = [] { return false; }; - func->getLogicalPlanFunc = GDSFunction::getLogicalPlan; - func->getPhysicalPlanFunc = GDSFunction::getPhysicalPlan; - result.push_back(std::move(func)); - return result; + return GDSPerNodeScalarAlgo::getFunctionSet(); } } // namespace algo_extension diff --git a/algo/src/include/function/algo_function.h b/algo/src/include/function/algo_function.h index 6a189ff5..790c7280 100644 --- a/algo/src/include/function/algo_function.h +++ b/algo/src/include/function/algo_function.h @@ -75,6 +75,14 @@ struct GDSLouvainFunction { static function::function_set getFunctionSet(); }; +// icebug (NetworKit)-backed Leiden community detection (ParallelLeidenView) — higher partition +// quality and performance than Louvain; prefer for new work. +struct GDSLeidenFunction { + static constexpr const char* name = "GDS_LEIDEN"; + + static function::function_set getFunctionSet(); +}; + // icebug (NetworKit)-backed personalized PageRank (random walk with restart from caller-supplied // source nodes) — scores measure standing relative to the sources, not globally. struct GDSPprFunction { diff --git a/algo/src/include/function/gds_algo_template.h b/algo/src/include/function/gds_algo_template.h new file mode 100644 index 00000000..d848cf18 --- /dev/null +++ b/algo/src/include/function/gds_algo_template.h @@ -0,0 +1,219 @@ +#pragma once + +// The GDS bridge factored by algorithm *shape*, so each new NetworKit algorithm is a small +// descriptor instead of a copied file. One class template per (input × output) equivalence +// class; an algorithm binds by supplying: +// +// static constexpr const char* name; // CALL surface, e.g. "GDS_PAGE_RANK" +// static constexpr const char* OUTPUT_COLUMN; // result column name, e.g. "rank" +// using OutputType = double | int64_t; // per-node scalar type +// using Params = ; // constructed from the optional param list +// using Extra = ; // NoExtraArgs, or e.g. SourceListArg +// static std::vector run(const NetworKit::GraphR& g, common::offset_t numNodes, +// const Params& params, const Extra::type& extra); +// +// The template supplies everything else: bind (graph entry, node output, columns, optional +// params, extra positional args), the shared zero-copy CSR bridge, GraphR construction, and +// result streaming through the GDS vertex-compute pipeline. Per-node *vector* outputs +// (embeddings) currently have a single member (GDS_NODE2VEC) and stay bespoke until a second +// member motivates a template. +#include "binder/binder.h" +#include "common/exception/binder.h" +#include "common/types/value/nested.h" +#include "function/gds/gds_utils.h" +#include "function/gds/gds_vertex_compute.h" +#include "function/gds_csr_bridge.h" +#include "function/table/bind_input.h" +#include "processor/execution_context.h" +#include "transaction/transaction.h" +#include +#include + +namespace lbug { +namespace algo_extension { + +template +struct GDSOutputTypeTraits; +template<> +struct GDSOutputTypeTraits { + static common::LogicalType logicalType() { return common::LogicalType::DOUBLE(); } +}; +template<> +struct GDSOutputTypeTraits { + static common::LogicalType logicalType() { return common::LogicalType::INT64(); } +}; + +// Extra-positional-argument policy: nothing beyond the graph name. +struct NoExtraArgs { + struct type {}; + static constexpr size_t NUM_ARGS = 0; + static type bind(const function::TableFuncBindInput*, const char* /*funcName*/) { return {}; } + static void validate(const type&, common::offset_t /*numNodes*/, const char* /*funcName*/) {} +}; + +// Extra-positional-argument policy: a non-empty LIST of node rowids (e.g. PPR sources). +struct SourceListArg { + using type = std::vector; + static constexpr size_t NUM_ARGS = 1; + + static type bind(const function::TableFuncBindInput* input, const char* funcName) { + const auto& value = input->getValue(1); + value.validateType(common::LogicalTypeID::LIST); + type sources; + sources.reserve(common::NestedVal::getChildrenSize(&value)); + for (auto i = 0u; i < common::NestedVal::getChildrenSize(&value); ++i) { + const auto* child = common::NestedVal::getChildVal(&value, i); + child->validateType(common::LogicalTypeID::INT64); + const auto source = child->getValue(); + if (source < 0) { + throw common::BinderException{ + std::string{funcName} + " source node offsets must be non-negative."}; + } + sources.push_back(static_cast(source)); + } + if (sources.empty()) { + throw common::BinderException{ + std::string{funcName} + " requires at least one source node."}; + } + return sources; + } + + static void validate(const type& sources, common::offset_t numNodes, const char* funcName) { + for (const auto source : sources) { + if (source >= numNodes) { + throw common::BinderException{std::string{funcName} + " source node offset " + + std::to_string(source) + + " is out of range: the projected graph has " + + std::to_string(numNodes) + " nodes."}; + } + } + } +}; + +// Per-node scalar output — the largest equivalence class (centrality, community detection). +template +struct GDSPerNodeScalarAlgo { + using OutputType = typename DESC::OutputType; + + struct BindData final : public function::GDSBindData { + std::string graphName; + typename DESC::Extra::type extra; + + BindData(binder::expression_vector columns, graph::NativeGraphEntry graphEntry, + binder::expression_vector output, std::unique_ptr params, + std::string graphName, typename DESC::Extra::type extra) + : function::GDSBindData{std::move(columns), std::move(graphEntry), std::move(output)}, + graphName{std::move(graphName)}, extra{std::move(extra)} { + this->optionalParams = std::move(params); + } + + std::unique_ptr copy() const override { + return std::make_unique(*this); + } + }; + + class ResultVertexCompute final : public function::GDSResultVertexCompute { + public: + ResultVertexCompute(storage::MemoryManager* mm, function::GDSFuncSharedState* sharedState, + const std::vector& values) + : GDSResultVertexCompute{mm, sharedState}, values{values} { + nodeIDVector = createVector(common::LogicalType::INTERNAL_ID()); + outputVector = createVector(GDSOutputTypeTraits::logicalType()); + } + + void beginOnTableInternal(common::table_id_t) override {} + + void vertexCompute(common::offset_t startOffset, common::offset_t endOffset, + common::table_id_t tableID) override { + for (auto i = startOffset; i < endOffset; ++i) { + if (skip(i)) { + continue; + } + nodeIDVector->setValue(0, common::nodeID_t{i, tableID}); + outputVector->setValue(0, i < values.size() ? values[i] : OutputType{}); + localFT->append(vectors); + } + } + + std::unique_ptr copy() override { + return std::make_unique(mm, sharedState, values); + } + + private: + const std::vector& values; + std::unique_ptr nodeIDVector; + std::unique_ptr outputVector; + }; + + static common::offset_t tableFunc(const function::TableFuncInput& input, + function::TableFuncOutput&) { + auto clientContext = input.context->clientContext; + auto transaction = transaction::Transaction::Get(*clientContext); + auto sharedState = input.sharedState->template ptrCast(); + auto graph = sharedState->graph.get(); + auto maxOffsetMap = graph->getMaxOffsetMap(transaction); + // MVP: single node table (the common case; multi-table is a follow-up). + if (maxOffsetMap.size() != 1) { + throw common::BinderException{ + std::string{DESC::name} + " currently supports single-node-table graphs only."}; + } + const auto tableID = maxOffsetMap.begin()->first; + const auto numNodes = maxOffsetMap.begin()->second; + auto mm = storage::MemoryManager::Get(*clientContext); + auto bindData = input.bindData->template constPtrCast(); + auto& params = bindData->optionalParams->template constCast(); + DESC::Extra::validate(bindData->extra, numNodes, DESC::name); + + // Undirected CSR — zero-copy from the projected graph's materialized arrow CSR when + // available, scan fallback otherwise (see gds_csr_bridge.cpp) — then a zero-copy + // GraphR and the descriptor's algorithm. + auto csr = + buildUndirectedCSR(clientContext, bindData->graphName, graph, tableID, numNodes, mm); + NetworKit::GraphR g(numNodes, /*directed=*/false, csr.indices, csr.indptr); + const auto values = DESC::run(g, numNodes, params, bindData->extra); + + auto outputVC = std::make_unique(mm, sharedState, values); + function::GDSUtils::runVertexCompute(input.context, function::GDSDensityState::DENSE, graph, + *outputVC); + sharedState->factorizedTablePool.mergeLocalTables(); + return 0; + } + + static std::unique_ptr bindFunc(main::ClientContext* context, + const function::TableFuncBindInput* input) { + auto graphName = input->getLiteralVal(0); + auto extra = DESC::Extra::bind(input, DESC::name); + auto graphEntry = function::GDSFunction::bindGraphEntry(*context, graphName); + auto nodeOutput = + function::GDSFunction::bindNodeOutput(*input, graphEntry.getNodeEntries()); + binder::expression_vector columns; + columns.push_back(nodeOutput->constCast().getInternalID()); + columns.push_back(input->binder->createVariable(DESC::OUTPUT_COLUMN, + GDSOutputTypeTraits::logicalType())); + return std::make_unique(std::move(columns), std::move(graphEntry), + binder::expression_vector{nodeOutput}, + std::make_unique(input->optionalParamsLegacy), + std::move(graphName), std::move(extra)); + } + + static function::function_set getFunctionSet() { + function::function_set result; + std::vector inputTypes{common::LogicalTypeID::ANY}; + for (auto i = 0u; i < DESC::Extra::NUM_ARGS; ++i) { + inputTypes.push_back(common::LogicalTypeID::ANY); + } + auto func = std::make_unique(std::string{DESC::name}, inputTypes); + func->bindFunc = bindFunc; + func->tableFunc = tableFunc; + func->initSharedStateFunc = function::GDSFunction::initSharedState; + func->initLocalStateFunc = function::TableFunction::initEmptyLocalState; + func->canParallelFunc = [] { return false; }; + func->getLogicalPlanFunc = function::GDSFunction::getLogicalPlan; + func->getPhysicalPlanFunc = function::GDSFunction::getPhysicalPlan; + result.push_back(std::move(func)); + return result; + } +}; + +} // namespace algo_extension +} // namespace lbug diff --git a/algo/src/main/algo_extension.cpp b/algo/src/main/algo_extension.cpp index be1f191f..45346a23 100644 --- a/algo/src/main/algo_extension.cpp +++ b/algo/src/main/algo_extension.cpp @@ -40,6 +40,7 @@ void AlgoExtension::load(main::ClientContext* context) { ExtensionUtils::addTableFunc(db); ExtensionUtils::addTableFunc(db); ExtensionUtils::addTableFunc(db); + ExtensionUtils::addTableFunc(db); ExtensionUtils::addTableFunc(db); #endif ExtensionUtils::addTableFunc(db); diff --git a/algo/test/test_files/gds_leiden.test b/algo/test/test_files/gds_leiden.test new file mode 100644 index 00000000..7236e325 --- /dev/null +++ b/algo/test/test_files/gds_leiden.test @@ -0,0 +1,54 @@ +-DATASET CSV empty + +-- + +-CASE GDSLeidenTwoCliques +-LOAD_DYNAMIC_EXTENSION algo +-STATEMENT CREATE NODE TABLE Node(id INT64 PRIMARY KEY); +---- ok +-STATEMENT CREATE REL TABLE Edge(FROM Node to Node); +---- ok +-STATEMENT CREATE (u0:Node {id: 0}), + (u1:Node {id: 1}), + (u2:Node {id: 2}), + (u3:Node {id: 3}), + (u4:Node {id: 4}), + (u5:Node {id: 5}), + (u6:Node {id: 6}), + (u7:Node {id: 7}), + (u0)-[:Edge]->(u1), + (u0)-[:Edge]->(u2), + (u1)-[:Edge]->(u2), + (u0)-[:Edge]->(u3), + (u1)-[:Edge]->(u3), + (u2)-[:Edge]->(u3), + (u4)-[:Edge]->(u5), + (u4)-[:Edge]->(u6), + (u5)-[:Edge]->(u6), + (u4)-[:Edge]->(u7), + (u5)-[:Edge]->(u7), + (u6)-[:Edge]->(u7), + (u3)-[:Edge]->(u4); +---- ok +-STATEMENT CALL PROJECT_GRAPH('Graph', ['Node'], ['Edge']) +---- ok +-LOG Leiden community IDs are arbitrary and the algorithm randomizes, so assert structure not +-LOG exact IDs: two communities, one per clique, joined by a bridge edge (u3-u4). +-STATEMENT CALL GDS_LEIDEN('Graph') RETURN COUNT(DISTINCT community_id) +---- 1 +2 +-STATEMENT CALL GDS_LEIDEN('Graph') WITH node, community_id WHERE node.id IN [0, 1, 2, 3] + RETURN COUNT(DISTINCT community_id) +---- 1 +1 +-STATEMENT CALL GDS_LEIDEN('Graph') WITH node, community_id WHERE node.id IN [4, 5, 6, 7] + RETURN COUNT(DISTINCT community_id) +---- 1 +1 +-LOG Gamma must be positive; unknown params rejected. +-STATEMENT CALL GDS_LEIDEN('Graph', gamma := -1.0) RETURN COUNT(DISTINCT community_id) +---- error +Binder exception: Gamma must be positive. +-STATEMENT CALL GDS_LEIDEN('Graph', bogus := 1) RETURN COUNT(DISTINCT community_id) +---- error +Binder exception: Unknown optional parameter: bogus