From 6fc76273b266730466ea791847e405b10504541c Mon Sep 17 00:00:00 2001 From: Arun Sharma Date: Fri, 16 Oct 2026 08:47:05 -0700 Subject: [PATCH] algo: table-driven GDS registry for icebug algorithms Replace the one-file-per-algorithm GDS integration with a catalog table (gds_registry.cpp): each icebug algorithm is one row supplying its CALL name, output column, parameter signature (Specs pack), extra-arg kind, graph mode, and a NetworKit invocation lambda. Shared builders in gds_generic_algo.h cover three shapes (per-node double, per-node int64 partition, per-node embedding); GDSGenericParams reuses OptionalParam for parsing/validation so no params code is written per algorithm. Migrates the 5 existing functions with identical semantics and adds 11 rows (betweenness, degree, closeness, harmonic, katz, eigenvector, core, connected components, PLP, cut clustering with a REQUIRED alpha, directed single-source BFS distances). Also fixes GDS_PPR binding on main: LIST-typed extras now use inferInputTypes (binder cannot build a LogicalType from a bare LIST ID), which un-breaks the 2 gds_ppr e2e tests. New gds_registry.test covers the added rows. 55/55 algo e2e tests pass. --- algo/src/function/CMakeLists.txt | 12 +- algo/src/function/gds_csr_bridge.cpp | 53 ++ algo/src/function/gds_leiden.cpp | 100 ---- algo/src/function/gds_louvain.cpp | 54 -- algo/src/function/gds_node2vec.cpp | 159 ----- algo/src/function/gds_page_rank.cpp | 77 --- algo/src/function/gds_ppr.cpp | 76 --- algo/src/function/gds_registry.cpp | 276 +++++++++ algo/src/include/function/algo_function.h | 38 -- algo/src/include/function/gds_algo_template.h | 241 -------- algo/src/include/function/gds_csr_bridge.h | 16 + algo/src/include/function/gds_generic_algo.h | 553 ++++++++++++++++++ .../src/include/function/gds_generic_params.h | 83 +++ algo/src/include/function/gds_param_specs.h | 158 +++++ algo/src/include/function/gds_registry.h | 42 ++ algo/src/main/algo_extension.cpp | 11 +- algo/test/test_files/gds_registry.test | 167 ++++++ 17 files changed, 1360 insertions(+), 756 deletions(-) delete mode 100644 algo/src/function/gds_leiden.cpp delete mode 100644 algo/src/function/gds_louvain.cpp delete mode 100644 algo/src/function/gds_node2vec.cpp delete mode 100644 algo/src/function/gds_page_rank.cpp delete mode 100644 algo/src/function/gds_ppr.cpp create mode 100644 algo/src/function/gds_registry.cpp delete mode 100644 algo/src/include/function/gds_algo_template.h create mode 100644 algo/src/include/function/gds_generic_algo.h create mode 100644 algo/src/include/function/gds_generic_params.h create mode 100644 algo/src/include/function/gds_param_specs.h create mode 100644 algo/src/include/function/gds_registry.h create mode 100644 algo/test/test_files/gds_registry.test diff --git a/algo/src/function/CMakeLists.txt b/algo/src/function/CMakeLists.txt index 7ce7bd83..4b41e75c 100644 --- a/algo/src/function/CMakeLists.txt +++ b/algo/src/function/CMakeLists.txt @@ -11,10 +11,10 @@ set(_algo_function_sources spanning_forest.cpp ) 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_leiden.cpp gds_ppr.cpp) + # gds_registry.cpp is the only translation unit including NetworKit + Arrow headers: + # one row per algorithm in the catalog table, no per-algorithm files. Linking the + # imported targets to the object library propagates their compile flags + include dirs. + list(APPEND _algo_function_sources gds_csr_bridge.cpp gds_registry.cpp) endif () add_library(lbug_algo_function @@ -23,8 +23,8 @@ add_library(lbug_algo_function if (ICEBUG_ENABLED) target_link_libraries(lbug_algo_function PRIVATE Arrow::arrow_shared OpenMP::OpenMP_CXX) - # ICEBUG_ENABLED is read by gds_page_rank.cpp via #ifdef to skip its body when icebug is off - # (defensive: source isn't compiled in that case, but matches the header-side guard). + # ICEBUG_ENABLED guards the registry TU's NetworKit/Arrow includes (defensive: the TU + # isn't compiled when icebug is off, but matches the header-side guard). target_compile_definitions(lbug_algo_function PRIVATE ICEBUG_ENABLED) else () # Other sources still use for parallel regions, so keep the OpenMP link — just drop Arrow. diff --git a/algo/src/function/gds_csr_bridge.cpp b/algo/src/function/gds_csr_bridge.cpp index f1a82dce..df50bac5 100644 --- a/algo/src/function/gds_csr_bridge.cpp +++ b/algo/src/function/gds_csr_bridge.cpp @@ -157,5 +157,58 @@ BridgeCSR buildUndirectedCSR(main::ClientContext* context, const std::string& gr return {std::move(indptr), std::move(indices)}; } +// Scan one direction into CSR arrays: fwdneighbors -> out pair, bwd neighbors -> in pair. +// Per-row sort + unique coalesces parallel edges (the SIMPLE directed projection); self-loops +// are kept — they are real directed edges, unlike in the symmetrized undirected scan. +static std::pair, std::shared_ptr> +scanDirectedCSR(table_id_t tableID, offset_t numNodes, Graph* graph, bool fwd) { + 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> adj(numNodes); + std::vector nbrs; + for (offset_t nodeId = 0; nodeId < numNodes; ++nodeId) { + nbrs.clear(); + const nodeID_t nid = {nodeId, tableID}; + if (fwd) { + for (auto chunk : graph->scanFwd(nid, *scanState)) { + chunk.forEach( + [&](auto neighbors, auto, auto i) { nbrs.push_back(neighbors[i].offset); }); + } + } else { + for (auto chunk : graph->scanBwd(nid, *scanState)) { + chunk.forEach( + [&](auto neighbors, auto, auto i) { nbrs.push_back(neighbors[i].offset); }); + } + } + std::sort(nbrs.begin(), nbrs.end()); + nbrs.erase(std::unique(nbrs.begin(), nbrs.end()), nbrs.end()); + adj[nodeId].reserve(nbrs.size()); + for (const auto nbr : nbrs) { + adj[nodeId].push_back(static_cast(nbr)); + } + } + std::vector flat; + flat.reserve(1024); + std::vector indptr(numNodes + 1, 0); + for (offset_t i = 0; i < numNodes; ++i) { + flat.insert(flat.end(), adj[i].begin(), adj[i].end()); + indptr[i + 1] = flat.size(); + } + auto indptrArr = + toU64([&](offset_t i) { return static_cast(indptr[i]); }, numNodes + 1); + auto indicesArr = toU64([&](offset_t i) { return static_cast(flat[i]); }, + static_cast(flat.size())); + return {std::move(indptrArr), std::move(indicesArr)}; +} + +DirectedBridgeCSR buildDirectedCSR(graph::Graph* graph, common::table_id_t tableID, + common::offset_t numNodes) { + auto [outIndptr, outIndices] = scanDirectedCSR(tableID, numNodes, graph, /*fwd=*/true); + auto [inIndptr, inIndices] = scanDirectedCSR(tableID, numNodes, graph, /*fwd=*/false); + return {std::move(outIndptr), std::move(outIndices), std::move(inIndptr), std::move(inIndices)}; +} + } // namespace algo_extension } // namespace lbug diff --git a/algo/src/function/gds_leiden.cpp b/algo/src/function/gds_leiden.cpp deleted file mode 100644 index 063f01f3..00000000 --- a/algo/src/function/gds_leiden.cpp +++ /dev/null @@ -1,100 +0,0 @@ -// 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 deleted file mode 100644 index 4a7c0ee3..00000000 --- a/algo/src/function/gds_louvain.cpp +++ /dev/null @@ -1,54 +0,0 @@ -// 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_algo_template.h" -#include -#include - -using namespace lbug::common; -using namespace lbug::function; - -namespace lbug { -namespace algo_extension { - -// 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()}; - } - } - void evaluateParams(main::ClientContext*) override {} - std::unique_ptr copy() override { - return std::make_unique(binder::expression_vector{}); - } -}; - -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; - } -}; - -function_set GDSLouvainFunction::getFunctionSet() { - return GDSPerNodeScalarAlgo::getFunctionSet(); -} - -} // namespace algo_extension -} // namespace lbug diff --git a/algo/src/function/gds_node2vec.cpp b/algo/src/function/gds_node2vec.cpp deleted file mode 100644 index 3c7fe416..00000000 --- a/algo/src/function/gds_node2vec.cpp +++ /dev/null @@ -1,159 +0,0 @@ -// GDS_NODE2VEC — node embeddings backed by icebug (NetworKit's Node2Vec). -// Produces a structural embedding per node from graph topology alone (no features, no text): -// materialize the projected graph as CSR, build a NetworKit::GraphR, run NetworKit::Node2Vec, -// and stream each node's embedding back as a LIST(FLOAT). Cosine-near embeddings = structurally -// (and, for a citation graph, conceptually) related nodes — the graph-native "semantics" primitive. -// -// MVP: fixed hyperparameters and a LIST(FLOAT) output (castable to ARRAY(FLOAT, d) for the vector -// extension's HNSW index). Configurable params + ARRAY output are follow-ups. -#include "binder/binder.h" -#include "common/exception/binder.h" -#include "function/algo_function.h" -#include "function/gds_csr_bridge.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 -#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 { - -// Node2Vec hyperparameters (MVP: fixed). Modest walk length / count / dimension keep it tractable. -static constexpr double N2V_P = 1.0; // return parameter -static constexpr double N2V_Q = 1.0; // in-out parameter -static constexpr uint64_t N2V_WALK_LEN = 10; // walk length -static constexpr uint64_t N2V_NUM_WALKS = 5; // walks per node -static constexpr uint64_t N2V_DIM = 64; // embedding dimension - -// 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 GDSNode2VecBindData final : public GDSBindData { - // Projected graph name, for looking up the entry's materialized arrow CSR at run time. - std::string graphName; - - GDSNode2VecBindData(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); - } -}; - -// Emits (node, embedding) rows; embedding is a LIST(FLOAT) read from Node2Vec's features by offset. -class GDSNode2VecResultVertexCompute : public GDSResultVertexCompute { -public: - GDSNode2VecResultVertexCompute(storage::MemoryManager* mm, GDSFuncSharedState* sharedState, - const std::vector>& features) - : GDSResultVertexCompute{mm, sharedState}, features{features} { - nodeIDVector = createVector(LogicalType::INTERNAL_ID()); - embVector = createVector(LogicalType::LIST(LogicalType::FLOAT())); - } - - 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}); - const auto& emb = i < features.size() ? features[i] : empty; - auto entry = ListVector::addList(embVector.get(), emb.size()); - embVector->setValue(0, entry); - auto* dataVector = ListVector::getDataVector(embVector.get()); - for (auto k = 0u; k < emb.size(); ++k) { - dataVector->setValue(entry.offset + k, emb[k]); - } - localFT->append(vectors); - } - } - - std::unique_ptr copy() override { - return std::make_unique(mm, sharedState, features); - } - -private: - const std::vector>& features; - const std::vector empty; - std::unique_ptr nodeIDVector; - std::unique_ptr embVector; -}; - -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); - if (maxOffsetMap.size() != 1) { - throw BinderException{"GDS_NODE2VEC 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, then Node2Vec. - NetworKit::GraphR g(numNodes, /*directed=*/false, csr.indices, csr.indptr); - NetworKit::Node2Vec n2v(g, N2V_P, N2V_Q, N2V_WALK_LEN, N2V_NUM_WALKS, N2V_DIM); - n2v.run(); - const std::vector>& features = n2v.getFeatures(); - - // 3. Stream embeddings back through the GDS result pipeline. - auto outputVC = std::make_unique(mm, sharedState, features); - GDSUtils::runVertexCompute(input.context, GDSDensityState::DENSE, graph, *outputVC); - sharedState->factorizedTablePool.mergeLocalTables(); - return 0; -} - -static constexpr char EMBEDDING_COLUMN_NAME[] = "embedding"; - -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(EMBEDDING_COLUMN_NAME, - LogicalType::LIST(LogicalType::FLOAT()))); - return std::make_unique(std::move(columns), std::move(graphEntry), - expression_vector{nodeOutput}, std::move(graphName)); -} - -function_set GDSNode2VecFunction::getFunctionSet() { - function_set result; - auto func = std::make_unique(GDSNode2VecFunction::name, - std::vector{LogicalTypeID::STRING}); - func->parameterNames = {"graph_name"}; - 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/function/gds_page_rank.cpp b/algo/src/function/gds_page_rank.cpp deleted file mode 100644 index 4fe7fd0e..00000000 --- a/algo/src/function/gds_page_rank.cpp +++ /dev/null @@ -1,77 +0,0 @@ -// 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. -// 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/config/max_iterations_config.h" -#include "function/config/page_rank_config.h" -#include "function/gds_algo_template.h" -#include - -using namespace lbug::common; -using namespace lbug::binder; -using namespace lbug::function; - -namespace lbug { -namespace algo_extension { - -// Reuse the PAGE_RANK optional params (dampingFactor, tolerance, maxIterations, normalize). -struct GDSPageRankOptionalParams final : public MaxIterationOptionalParams { - OptionalParam dampingFactor; - OptionalParam tolerance; - - explicit GDSPageRankOptionalParams(const expression_vector& optionalParams) - : MaxIterationOptionalParams{constructMaxIterationParam(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 if (paramName == MaxIterations::NAME) { - continue; - } else { - throw BinderException{"Unknown optional parameter: " + optionalParam->getAlias()}; - } - } - } - - GDSPageRankOptionalParams(OptionalParam maxIterations, - OptionalParam dampingFactor, OptionalParam tolerance) - : MaxIterationOptionalParams{maxIterations}, dampingFactor{std::move(dampingFactor)}, - tolerance{std::move(tolerance)} {} - - void evaluateParams(main::ClientContext* context) override { - MaxIterationOptionalParams::evaluateParams(context); - dampingFactor.evaluateParam(context); - tolerance.evaluateParam(context); - } - - std::unique_ptr copy() override { - return std::make_unique(maxIterations, dampingFactor, tolerance); - } -}; - -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(); - } -}; - -function_set GDSPageRankFunction::getFunctionSet() { - return GDSPerNodeScalarAlgo::getFunctionSet(); -} - -} // namespace algo_extension -} // namespace lbug diff --git a/algo/src/function/gds_ppr.cpp b/algo/src/function/gds_ppr.cpp deleted file mode 100644 index 1e2ba493..00000000 --- a/algo/src/function/gds_ppr.cpp +++ /dev/null @@ -1,76 +0,0 @@ -// 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, 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 "function/algo_function.h" -#include "function/config/page_rank_config.h" -#include "function/gds_algo_template.h" -#include - -using namespace lbug::common; -using namespace lbug::binder; -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 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(); - } -}; - -function_set GDSPprFunction::getFunctionSet() { - return GDSPerNodeScalarAlgo::getFunctionSet(); -} - -} // namespace algo_extension -} // namespace lbug diff --git a/algo/src/function/gds_registry.cpp b/algo/src/function/gds_registry.cpp new file mode 100644 index 00000000..c5dbfe5e --- /dev/null +++ b/algo/src/function/gds_registry.cpp @@ -0,0 +1,276 @@ +// The GDS catalog: one row per icebug-backed table function. See gds_registry.h for the +// row checklist. Shapes and policies live in gds_generic_algo.h; knob specs in +// gds_param_specs.h (+ function::DampingFactor/Tolerance, MaxIterations). +// +// Coverage map (NetworKit module -> rows here): +// centrality PageRank, Betweenness, Degree, Closeness, Harmonic, Katz, Eigenvector, Core +// community PLM (Louvain), ParallelLeidenView (Leiden), PLP, CutClustering +// components ConnectedComponents +// distance BFS (single-source distances), PageRank::forSources (personalized) +// embedding Node2Vec (makeEmbeddingFunc also covers FastRP once the vendored prebuilt +// ships networkit/embedding/FastRP.hpp — the builder is double-instantiable) +// Follow-up shapes (global scalars, edge scores, link prediction, SCD, generators, ...) need +// new builders; rows for them do not fit make*Func and are rejected at design time, not with +// a half-working CALL. + +#include "function/gds_registry.h" + +#include "catalog/catalog.h" +#include "function/config/max_iterations_config.h" +#include "function/config/page_rank_config.h" +#include "function/gds_generic_algo.h" +#include "function/gds_param_specs.h" +#include "main/database.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace lbug { +namespace algo_extension { + +namespace { + +// --- Migrated rows (semantics identical to the retired per-algorithm gds_*.cpp files). --- + +function::function_set makeGDSPageRank() { + // MaxIterations is accepted + validated for CALL compatibility with PAGE_RANK; the + // NetworKit iteration does not take a cap, so the value is unused (as before). + return makeNodeScoreFunc( + "GDS_PAGE_RANK", "rank", GDSExtraKind::NONE, GDSGraphMode::UNDIRECTED, + [](const NetworKit::GraphR& g, common::offset_t, const auto& params, const auto&) { + NetworKit::PageRank pr(g, params.template get(), + params.template get()); + pr.run(); + return pr.scores(); + }); +} + +function::function_set makeGDSPpr() { + return makeNodeScoreFunc("GDS_PPR", "rank", + GDSExtraKind::SOURCE_LIST, GDSGraphMode::UNDIRECTED, + [](const NetworKit::GraphR& g, common::offset_t, const auto& params, const auto& extra) { + const std::vector sources{extra.sources.begin(), extra.sources.end()}; + auto pr = NetworKit::PageRank::forSources(g, sources, + params.template get(), + params.template get()); + pr.run(); + return pr.scores(); + }); +} + +function::function_set makeGDSLouvain() { + return makePartitionFunc<>("GDS_LOUVAIN", "community_id", GDSExtraKind::NONE, + GDSGraphMode::UNDIRECTED, + [](const NetworKit::GraphR& g, common::offset_t numNodes, const auto&, const auto&) { + NetworKit::PLM plm(g); + plm.run(); + return partitionToIds(plm.getPartition(), numNodes); + }); +} + +function::function_set makeGDSLeiden() { + return makePartitionFunc("GDS_LEIDEN", "community_id", + GDSExtraKind::NONE, GDSGraphMode::UNDIRECTED, + [](const NetworKit::GraphR& g, common::offset_t numNodes, const auto& params, const auto&) { + NetworKit::ParallelLeidenView leiden(g, + static_cast(params.template get()), /*randomize=*/true, + params.template get()); + leiden.run(); + return partitionToIds(leiden.getPartition(), numNodes); + }); +} + +function::function_set makeGDSNode2Vec() { + return makeEmbeddingFunc("GDS_NODE2VEC", GDSExtraKind::NONE, GDSGraphMode::UNDIRECTED, + [](const NetworKit::GraphR& g, common::offset_t, const auto&, const auto&) { + NetworKit::Node2Vec n2v(g, /*P=*/1.0, /*Q=*/1.0, /*L=*/10, /*N=*/5, /*D=*/64); + n2v.run(); + return n2v.getFeatures(); + }); +} + +// --- New rows: centrality (all share the scores() shape; only the signature differs). --- + +function::function_set makeGDSBetweenness() { + return makeNodeScoreFunc("GDS_BETWEENNESS", "score", GDSExtraKind::NONE, + GDSGraphMode::UNDIRECTED, + [](const NetworKit::GraphR& g, common::offset_t, const auto& params, const auto&) { + NetworKit::Betweenness algo(g, params.template get()); + algo.run(); + return algo.scores(); + }); +} + +function::function_set makeGDSDegree() { + return makeNodeScoreFunc("GDS_DEGREE", "score", GDSExtraKind::NONE, + GDSGraphMode::UNDIRECTED, + [](const NetworKit::GraphR& g, common::offset_t, const auto& params, const auto&) { + NetworKit::DegreeCentrality algo(g, params.template get()); + algo.run(); + return algo.scores(); + }); +} + +function::function_set makeGDSCloseness() { + return makeNodeScoreFunc("GDS_CLOSENESS", "score", GDSExtraKind::NONE, + GDSGraphMode::UNDIRECTED, + [](const NetworKit::GraphR& g, common::offset_t, const auto& params, const auto&) { + // Three-arg form: the two-arg call is ambiguous between the + // (normalized, variant) and the legacy (normalized, checkConnectedness) ctors. + NetworKit::Closeness algo(g, params.template get(), + /*checkConnectedness=*/true); + algo.run(); + return algo.scores(); + }); +} + +function::function_set makeGDSHarmonic() { + return makeNodeScoreFunc("GDS_HARMONIC", "score", GDSExtraKind::NONE, + GDSGraphMode::UNDIRECTED, + [](const NetworKit::GraphR& g, common::offset_t, const auto& params, const auto&) { + NetworKit::HarmonicCloseness algo(g, params.template get()); + algo.run(); + return algo.scores(); + }); +} + +function::function_set makeGDSKatz() { + return makeNodeScoreFunc("GDS_KATZ", "score", + GDSExtraKind::NONE, GDSGraphMode::UNDIRECTED, + [](const NetworKit::GraphR& g, common::offset_t, const auto& params, const auto&) { + NetworKit::KatzCentrality algo(g, params.template get(), + params.template get(), params.template get()); + algo.run(); + return algo.scores(); + }); +} + +function::function_set makeGDSEigenvector() { + return makeNodeScoreFunc("GDS_EIGENVECTOR", "score", GDSExtraKind::NONE, + GDSGraphMode::UNDIRECTED, + [](const NetworKit::GraphR& g, common::offset_t, const auto& params, const auto&) { + NetworKit::EigenvectorCentrality algo(g, params.template get()); + algo.run(); + return algo.scores(); + }); +} + +function::function_set makeGDSCore() { + return makeNodeScoreFunc("GDS_CORE_DECOMPOSITION", "core_number", + GDSExtraKind::NONE, GDSGraphMode::UNDIRECTED, + [](const NetworKit::GraphR& g, common::offset_t, const auto& params, const auto&) { + NetworKit::CoreDecomposition algo(g, params.template get()); + algo.run(); + return algo.scores(); + }); +} + +// --- New rows: community / components (Partition shape). --- + +function::function_set makeGDSConnectedComponents() { + return makePartitionFunc<>("GDS_CONNECTED_COMPONENTS", "component_id", GDSExtraKind::NONE, + GDSGraphMode::UNDIRECTED, + [](const NetworKit::GraphR& g, common::offset_t numNodes, const auto&, const auto&) { + NetworKit::ConnectedComponents algo(g); + algo.run(); + return partitionToIds(algo.getPartition(), numNodes); + }); +} + +function::function_set makeGDSPlp() { + return makePartitionFunc("GDS_PLP", "community_id", + GDSExtraKind::NONE, GDSGraphMode::UNDIRECTED, + [](const NetworKit::GraphR& g, common::offset_t numNodes, const auto& params, const auto&) { + const auto theta = params.template get() < 0 ? + NetworKit::none : + static_cast(params.template get()); + const auto maxIter = + params.template get() < 0 ? + NetworKit::none : + static_cast(params.template get()); + NetworKit::PLP algo(g, theta, maxIter); + algo.run(); + return partitionToIds(algo.getPartition(), numNodes); + }); +} + +function::function_set makeGDSCutClustering() { + // alpha is REQUIRED: CutClustering takes it as a constructor argument with no default. + return makePartitionFunc("GDS_CUT_CLUSTERING", "community_id", GDSExtraKind::NONE, + GDSGraphMode::UNDIRECTED, + [](const NetworKit::GraphR& g, common::offset_t numNodes, const auto& params, const auto&) { + NetworKit::CutClustering algo(g, params.template get()); + algo.run(); + return partitionToIds(algo.getPartition(), numNodes); + }); +} + +// --- New rows: single-source distances (directed) and embeddings. --- + +function::function_set makeGDSBfsDistance() { + return makeNodeScoreFunc<>("GDS_BFS_DISTANCE", "distance", GDSExtraKind::SINGLE_SOURCE, + GDSGraphMode::DIRECTED, + [](const NetworKit::GraphR& g, common::offset_t numNodes, const auto&, const auto& extra) { + NetworKit::BFS algo(g, extra.sources[0]); + algo.run(); + const auto& dist = algo.getDistances(); + std::vector out(numNodes, -1.0); + for (common::offset_t i = 0; i < numNodes && i < (common::offset_t)dist.size(); ++i) { + out[i] = normalizeDistance(dist[i]); + } + return out; + }); +} + +} // namespace + +const std::vector& getGDSCatalog() { + static const std::vector catalog = { + {"GDS_PAGE_RANK", makeGDSPageRank}, + {"GDS_PPR", makeGDSPpr}, + {"GDS_LOUVAIN", makeGDSLouvain}, + {"GDS_LEIDEN", makeGDSLeiden}, + {"GDS_NODE2VEC", makeGDSNode2Vec}, + {"GDS_BETWEENNESS", makeGDSBetweenness}, + {"GDS_DEGREE", makeGDSDegree}, + {"GDS_CLOSENESS", makeGDSCloseness}, + {"GDS_HARMONIC", makeGDSHarmonic}, + {"GDS_KATZ", makeGDSKatz}, + {"GDS_EIGENVECTOR", makeGDSEigenvector}, + {"GDS_CORE_DECOMPOSITION", makeGDSCore}, + {"GDS_CONNECTED_COMPONENTS", makeGDSConnectedComponents}, + {"GDS_PLP", makeGDSPlp}, + {"GDS_CUT_CLUSTERING", makeGDSCutClustering}, + {"GDS_BFS_DISTANCE", makeGDSBfsDistance}, + }; + return catalog; +} + +void registerGDSFunctions(main::Database& database) { + auto catalog = database.getCatalog(); + for (const auto& entry : getGDSCatalog()) { + if (catalog->containsFunction(&transaction::DUMMY_TRANSACTION, entry.name, + false /*isInternal*/)) { + continue; + } + catalog->addFunction(&transaction::DUMMY_TRANSACTION, + catalog::CatalogEntryType::TABLE_FUNCTION_ENTRY, entry.name, entry.makeFunctionSet(), + false /*isInternal*/); + } +} + +} // namespace algo_extension +} // namespace lbug diff --git a/algo/src/include/function/algo_function.h b/algo/src/include/function/algo_function.h index 790c7280..993d809c 100644 --- a/algo/src/include/function/algo_function.h +++ b/algo/src/include/function/algo_function.h @@ -53,44 +53,6 @@ struct PageRankAliasFunction { static constexpr const char* name = "PR"; }; -// icebug (NetworKit)-backed PageRank. Coexists with PAGE_RANK for one release cycle. -struct GDSPageRankFunction { - static constexpr const char* name = "GDS_PAGE_RANK"; - - static function::function_set getFunctionSet(); -}; - -// icebug (NetworKit)-backed Node2Vec structural embeddings. -struct GDSNode2VecFunction { - static constexpr const char* name = "GDS_NODE2VEC"; - - 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(); -}; - -// 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 { - 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/include/function/gds_algo_template.h b/algo/src/include/function/gds_algo_template.h deleted file mode 100644 index 897f6328..00000000 --- a/algo/src/include/function/gds_algo_template.h +++ /dev/null @@ -1,241 +0,0 @@ -#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::STRING}; - std::vector paramNames{"graph_name"}; - // LogicalType is explicitly-copyable only (private copy ctor): build via moves, - // not an initializer list (which would copy). - std::vector paramTypes; - paramTypes.reserve(1 + DESC::Extra::NUM_ARGS); - paramTypes.push_back(common::LogicalType::STRING()); - for (auto i = 0u; i < DESC::Extra::NUM_ARGS; ++i) { - // Extra args are lists of node rowids (see SourceListArg::bind). - inputTypes.push_back(common::LogicalTypeID::LIST); - paramNames.emplace_back("sources"); - paramTypes.push_back(common::LogicalType::LIST(common::LogicalType::INT64())); - } - auto func = std::make_unique(std::string{DESC::name}, inputTypes); - if constexpr (DESC::Extra::NUM_ARGS > 0) { - // Extra args are nested types (SOURCE_LIST's LIST(INT64)). The binder builds - // input LogicalTypes from the declared LogicalTypeIDs, and a bare LIST ID has - // no child info — infer the nested type from the CALL-site argument instead. - func->inferInputTypes = [](const binder::expression_vector& params) { - std::vector types; - types.push_back(common::LogicalType::STRING()); - types.push_back(params[1]->getDataType().copy()); - return types; - }; - } - func->parameterNames = std::move(paramNames); - func->parameterTypes = std::move(paramTypes); - 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/include/function/gds_csr_bridge.h b/algo/src/include/function/gds_csr_bridge.h index 1cc40532..ddaccdc0 100644 --- a/algo/src/include/function/gds_csr_bridge.h +++ b/algo/src/include/function/gds_csr_bridge.h @@ -29,6 +29,17 @@ struct BridgeCSR { // on the zero-copy path) }; +// Directed adjacency as separate out/in CSR pair (no symmetrization: parallel edges coalesce +// per direction, self-loops are kept). Both scans come from storage, so directed mode always +// takes the scan path even when a materialized (fwd-only) CSR is pinned on the graph entry — +// transposing it would risk serving a different snapshot for in-edges than out-edges. +struct DirectedBridgeCSR { + std::shared_ptr outIndptr; // length numNodes + 1 + std::shared_ptr outIndices; // per-row sorted, deduplicated + std::shared_ptr inIndptr; + std::shared_ptr inIndices; +}; + // 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 @@ -40,5 +51,10 @@ BridgeCSR buildUndirectedCSR(main::ClientContext* context, const std::string& gr graph::Graph* graph, common::table_id_t tableID, common::offset_t numNodes, storage::MemoryManager* mm); +// Build the directed CSR pair for a single-node-table projected graph by scanning storage +// (fwd for out-edges, bwd for in-edges). Always scans; see DirectedBridgeCSR. +DirectedBridgeCSR buildDirectedCSR(graph::Graph* graph, common::table_id_t tableID, + common::offset_t numNodes); + } // namespace algo_extension } // namespace lbug diff --git a/algo/src/include/function/gds_generic_algo.h b/algo/src/include/function/gds_generic_algo.h new file mode 100644 index 00000000..38bcffff --- /dev/null +++ b/algo/src/include/function/gds_generic_algo.h @@ -0,0 +1,553 @@ +#pragma once + +// Generic builders behind the GDS registry (gds_registry.cpp). +// +// A registry row is one call to makeNodeScoreFunc / makePartitionFunc / makeEmbeddingFunc: +// +// makeNodeScoreFunc( +// \"GDS_X\", \"score\", GDSExtraKind::NONE, GDSGraphMode::UNDIRECTED, +// [](const NetworKit::GraphR& g, common::offset_t, const auto& params, const auto&) { +// NetworKit::SomeCentrality algo(g, params.template get(), ...); +// algo.run(); +// return algo.scores(); +// }); +// +// The Specs... pack is the row's parameter signature (see gds_generic_params.h); the lambda +// body is the row's algorithm invocation. Everything else — CALL binding, the CSR bridge, +// GraphR construction, result streaming — is shared. Adding an algorithm is one table row: +// no new file, no CMake change, no extension-load change. +// +// Shapes: +// makeNodeScoreFunc — double per node (centrality scores, SSSP distances, core numbers). +// makePartitionFunc — int64 per node (community / component ids from a NetworKit Partition). +// makeEmbeddingFunc — fixed-width float/double vector per node (Node2Vec, FastRP). +// +// Algo-type dimensions expressed per row: +// GDSExtraKind — NONE (graph name only), SOURCE_LIST (PPR-style LIST of rowids), +// SINGLE_SOURCE (BFS-style single INT64 rowid). +// GDSGraphMode — UNDIRECTED (symmetrized simple projection, zero-copy when possible), +// DIRECTED (out/in CSR pair scanned from storage). + +#include +#include +#include + +#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/gds_generic_params.h" +#include "function/table/bind_input.h" +#include "processor/execution_context.h" +#include "transaction/transaction.h" +#include +#include +#include + +namespace lbug { +namespace algo_extension { + +enum class GDSExtraKind : uint8_t { NONE, SOURCE_LIST, SINGLE_SOURCE }; +enum class GDSGraphMode : uint8_t { UNDIRECTED, DIRECTED }; + +// Positional extra arguments after the graph name. SINGLE_SOURCE stores its one rowid in +// sources[0]; SOURCE_LIST stores the whole list. +struct GDSExtraArgs { + std::vector sources; +}; + +// Flatten a NetworKit partition to per-node int64 ids. Nodes outside the partition's range +// (should not happen on a dense ids graph) map to -1 rather than crashing. +inline std::vector partitionToIds(const NetworKit::Partition& partition, + common::offset_t numNodes) { + std::vector ids(numNodes, -1); + const auto count = std::min(numNodes, partition.numberOfElements()); + for (common::offset_t i = 0; i < count; ++i) { + ids[i] = static_cast(partition[i]); + } + return ids; +} + +// SSSP distances use huge sentinels for unreachable nodes; surface those as -1 so the column +// stays readable (reachable weighted distances are orders of magnitude smaller). +inline double normalizeDistance(double d) { + return d >= 1e100 ? -1.0 : d; +} + +namespace detail { + +// Appends the CALL-signature types for the extra positional args (used when building the +// TableFunction; no bind input is available there yet). +inline void appendExtraSignature(GDSExtraKind kind, std::vector& inputTypes, + std::vector& paramNames, std::vector& paramTypes) { + if (kind == GDSExtraKind::SOURCE_LIST) { + inputTypes.push_back(common::LogicalTypeID::LIST); + paramNames.emplace_back("sources"); + paramTypes.push_back(common::LogicalType::LIST(common::LogicalType::INT64())); + } else if (kind == GDSExtraKind::SINGLE_SOURCE) { + inputTypes.push_back(common::LogicalTypeID::INT64); + paramNames.emplace_back("source"); + paramTypes.push_back(common::LogicalType::INT64()); + } +} + +inline GDSExtraArgs bindExtraArgs(GDSExtraKind kind, const function::TableFuncBindInput* input, + const char* funcName) { + GDSExtraArgs extra; + if (kind == GDSExtraKind::SOURCE_LIST) { + const auto& value = input->getValue(1); + value.validateType(common::LogicalTypeID::LIST); + extra.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."}; + } + extra.sources.push_back(static_cast(source)); + } + if (extra.sources.empty()) { + throw common::BinderException{ + std::string{funcName} + " requires at least one source node."}; + } + } else if (kind == GDSExtraKind::SINGLE_SOURCE) { + const auto source = input->getLiteralVal(1); + if (source < 0) { + throw common::BinderException{ + std::string{funcName} + " source node offset must be non-negative."}; + } + extra.sources.push_back(static_cast(source)); + } + return extra; +} + +inline void validateExtraArgs(const GDSExtraArgs& extra, common::offset_t numNodes, + const char* funcName) { + for (const auto source : extra.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."}; + } + } +} + +// Functions whose extras include nested types (SOURCE_LIST's LIST(INT64)) cannot use the +// default input-type path: the binder builds LogicalTypes from the declared LogicalTypeIDs, +// and a bare LIST ID has no child info. Infer the nested type from the CALL-site argument +// instead (same mechanism as query_hnsw_index's ARRAY param). +inline void setInferInputTypes(GDSExtraKind kind, function::TableFunction* func) { + if (kind == GDSExtraKind::SOURCE_LIST) { + func->inferInputTypes = [](const binder::expression_vector& params) { + std::vector types; + types.push_back(common::LogicalType::STRING()); + types.push_back(params[1]->getDataType().copy()); + return types; + }; + } +} + +template +struct ScalarLogicalType; +template<> +struct ScalarLogicalType { + static common::LogicalType get() { return common::LogicalType::DOUBLE(); } +}; +template<> +struct ScalarLogicalType { + static common::LogicalType get() { return common::LogicalType::INT64(); } +}; + +template +struct GenericScalarBindData final : public function::GDSBindData { + using Runner = std::function(const NetworKit::GraphR&, common::offset_t, + const Params&, const GDSExtraArgs&)>; + + std::string funcName; + std::string graphName; + GDSExtraArgs extra; + Runner runner; + GDSGraphMode mode; + + GenericScalarBindData(binder::expression_vector columns, graph::NativeGraphEntry graphEntry, + binder::expression_vector output, std::unique_ptr params, + std::string funcName, std::string graphName, GDSExtraArgs extra, Runner runner, + GDSGraphMode mode) + : function::GDSBindData{std::move(columns), std::move(graphEntry), std::move(output)}, + funcName{std::move(funcName)}, graphName{std::move(graphName)}, extra{std::move(extra)}, + runner{std::move(runner)}, mode{mode} { + this->optionalParams = std::move(params); + } + + std::unique_ptr copy() const override { + return std::make_unique(*this); + } +}; + +template +class GenericScalarVertexCompute final : public function::GDSResultVertexCompute { +public: + GenericScalarVertexCompute(storage::MemoryManager* mm, + function::GDSFuncSharedState* sharedState, const std::vector& values) + : GDSResultVertexCompute{mm, sharedState}, values{values} { + nodeIDVector = createVector(common::LogicalType::INTERNAL_ID()); + outputVector = createVector(ScalarLogicalType::get()); + } + + 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; +}; + +template +struct EmbeddingLogicalType; +template<> +struct EmbeddingLogicalType { + static common::LogicalType get() { + return common::LogicalType::LIST(common::LogicalType::FLOAT()); + } +}; +template<> +struct EmbeddingLogicalType { + static common::LogicalType get() { + return common::LogicalType::LIST(common::LogicalType::DOUBLE()); + } +}; + +template +class GenericEmbeddingVertexCompute final : public function::GDSResultVertexCompute { +public: + GenericEmbeddingVertexCompute(storage::MemoryManager* mm, + function::GDSFuncSharedState* sharedState, const std::vector>& features) + : GDSResultVertexCompute{mm, sharedState}, features{features} { + nodeIDVector = createVector(common::LogicalType::INTERNAL_ID()); + embVector = createVector(EmbeddingLogicalType::get()); + } + + 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}); + const auto& emb = i < features.size() ? features[i] : empty; + auto entry = common::ListVector::addList(embVector.get(), emb.size()); + embVector->setValue(0, entry); + auto* dataVector = common::ListVector::getDataVector(embVector.get()); + for (auto k = 0u; k < emb.size(); ++k) { + if constexpr (std::is_same_v) { + dataVector->setValue(entry.offset + k, emb[k]); + } else { + dataVector->setValue(entry.offset + k, emb[k]); + } + } + localFT->append(vectors); + } + } + + std::unique_ptr copy() override { + return std::make_unique(mm, sharedState, features); + } + +private: + const std::vector>& features; + const std::vector empty; + std::unique_ptr nodeIDVector; + std::unique_ptr embVector; +}; + +} // namespace detail + +// Per-node double output (centrality scores, distances, core numbers, ...). +template +function::function_set makeNodeScoreFunc(std::string funcName, std::string outputColumn, + GDSExtraKind extraKind, GDSGraphMode mode, Runner&& runner) { + using Params = GDSGenericParams; + using BindData = detail::GenericScalarBindData; + typename BindData::Runner erasedRunner = std::forward(runner); + function::function_set result; + std::vector inputTypes{common::LogicalTypeID::STRING}; + std::vector paramNames{"graph_name"}; + std::vector paramTypes; + paramTypes.push_back(common::LogicalType::STRING()); + auto tableFunc = [erasedRunner](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); + if (maxOffsetMap.size() != 1) { + throw common::BinderException{"GDS functions currently support 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(); + detail::validateExtraArgs(bindData->extra, numNodes, bindData->funcName.c_str()); + std::vector values; + if (bindData->mode == GDSGraphMode::DIRECTED) { + auto csr = buildDirectedCSR(graph, tableID, numNodes); + NetworKit::GraphR g(numNodes, /*directed=*/true, csr.outIndices, csr.outIndptr, + csr.inIndices, csr.inIndptr); + values = bindData->runner(g, numNodes, params, bindData->extra); + } else { + auto csr = buildUndirectedCSR(clientContext, bindData->graphName, graph, tableID, + numNodes, mm); + NetworKit::GraphR g(numNodes, /*directed=*/false, csr.indices, csr.indptr); + values = bindData->runner(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; + }; + auto bindFunc = [funcName, outputColumn, extraKind, mode, erasedRunner]( + main::ClientContext* context, const function::TableFuncBindInput* input) { + auto graphName = input->getLiteralVal(0); + auto extra = detail::bindExtraArgs(extraKind, input, funcName.c_str()); + 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(outputColumn, common::LogicalType::DOUBLE())); + return std::unique_ptr(std::make_unique( + std::move(columns), std::move(graphEntry), binder::expression_vector{nodeOutput}, + std::make_unique(input->optionalParamsLegacy), funcName, std::move(graphName), + std::move(extra), erasedRunner, mode)); + }; + // The CALL signature's positional types must be fixed here (bind sees only values, not + // types): mirror the bind-time logic for the extra args. + detail::appendExtraSignature(extraKind, inputTypes, paramNames, paramTypes); + auto func = std::make_unique(funcName, inputTypes); + detail::setInferInputTypes(extraKind, func.get()); + func->parameterNames = std::move(paramNames); + func->parameterTypes = std::move(paramTypes); + func->bindFunc = std::move(bindFunc); + func->tableFunc = std::move(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; +} + +// Per-node int64 output (community / component ids). +template +function::function_set makePartitionFunc(std::string funcName, std::string outputColumn, + GDSExtraKind extraKind, GDSGraphMode mode, Runner&& runner) { + using Params = GDSGenericParams; + using BindData = detail::GenericScalarBindData; + typename BindData::Runner erasedRunner = std::forward(runner); + function::function_set result; + std::vector inputTypes{common::LogicalTypeID::STRING}; + std::vector paramNames{"graph_name"}; + std::vector paramTypes; + paramTypes.push_back(common::LogicalType::STRING()); + auto tableFunc = [erasedRunner](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); + if (maxOffsetMap.size() != 1) { + throw common::BinderException{"GDS functions currently support 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(); + detail::validateExtraArgs(bindData->extra, numNodes, bindData->funcName.c_str()); + std::vector values; + if (bindData->mode == GDSGraphMode::DIRECTED) { + auto csr = buildDirectedCSR(graph, tableID, numNodes); + NetworKit::GraphR g(numNodes, /*directed=*/true, csr.outIndices, csr.outIndptr, + csr.inIndices, csr.inIndptr); + values = bindData->runner(g, numNodes, params, bindData->extra); + } else { + auto csr = buildUndirectedCSR(clientContext, bindData->graphName, graph, tableID, + numNodes, mm); + NetworKit::GraphR g(numNodes, /*directed=*/false, csr.indices, csr.indptr); + values = bindData->runner(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; + }; + auto bindFunc = [funcName, outputColumn, extraKind, mode, erasedRunner]( + main::ClientContext* context, const function::TableFuncBindInput* input) { + auto graphName = input->getLiteralVal(0); + auto extra = detail::bindExtraArgs(extraKind, input, funcName.c_str()); + 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(outputColumn, common::LogicalType::INT64())); + return std::unique_ptr(std::make_unique( + std::move(columns), std::move(graphEntry), binder::expression_vector{nodeOutput}, + std::make_unique(input->optionalParamsLegacy), funcName, std::move(graphName), + std::move(extra), erasedRunner, mode)); + }; + detail::appendExtraSignature(extraKind, inputTypes, paramNames, paramTypes); + auto func = std::make_unique(funcName, inputTypes); + detail::setInferInputTypes(extraKind, func.get()); + func->parameterNames = std::move(paramNames); + func->parameterTypes = std::move(paramTypes); + func->bindFunc = std::move(bindFunc); + func->tableFunc = std::move(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; +} + +// Per-node embedding output (LIST(FLOAT) for float, LIST(DOUBLE) for double). +template +function::function_set makeEmbeddingFunc(std::string funcName, GDSExtraKind extraKind, + GDSGraphMode mode, Runner&& runner) { + using Params = GDSGenericParams; + using RunnerT = std::function>(const NetworKit::GraphR&, + common::offset_t, const Params&, const GDSExtraArgs&)>; + struct BindData final : public function::GDSBindData { + std::string funcName; + std::string graphName; + GDSExtraArgs extra; + RunnerT runner; + GDSGraphMode mode; + + BindData(binder::expression_vector columns, graph::NativeGraphEntry graphEntry, + binder::expression_vector output, std::unique_ptr params, + std::string funcName, std::string graphName, GDSExtraArgs extra, RunnerT runner, + GDSGraphMode mode) + : function::GDSBindData{std::move(columns), std::move(graphEntry), std::move(output)}, + funcName{std::move(funcName)}, graphName{std::move(graphName)}, + extra{std::move(extra)}, runner{std::move(runner)}, mode{mode} { + this->optionalParams = std::move(params); + } + + std::unique_ptr copy() const override { + return std::make_unique(*this); + } + }; + RunnerT erasedRunner = std::forward(runner); + function::function_set result; + std::vector inputTypes{common::LogicalTypeID::STRING}; + std::vector paramNames{"graph_name"}; + std::vector paramTypes; + paramTypes.push_back(common::LogicalType::STRING()); + auto tableFunc = [erasedRunner](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); + if (maxOffsetMap.size() != 1) { + throw common::BinderException{"GDS functions currently support 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(); + detail::validateExtraArgs(bindData->extra, numNodes, bindData->funcName.c_str()); + std::vector> features; + if (bindData->mode == GDSGraphMode::DIRECTED) { + auto csr = buildDirectedCSR(graph, tableID, numNodes); + NetworKit::GraphR g(numNodes, /*directed=*/true, csr.outIndices, csr.outIndptr, + csr.inIndices, csr.inIndptr); + features = bindData->runner(g, numNodes, params, bindData->extra); + } else { + auto csr = buildUndirectedCSR(clientContext, bindData->graphName, graph, tableID, + numNodes, mm); + NetworKit::GraphR g(numNodes, /*directed=*/false, csr.indices, csr.indptr); + features = bindData->runner(g, numNodes, params, bindData->extra); + } + auto outputVC = std::make_unique>(mm, + sharedState, features); + function::GDSUtils::runVertexCompute(input.context, function::GDSDensityState::DENSE, graph, + *outputVC); + sharedState->factorizedTablePool.mergeLocalTables(); + return 0; + }; + auto bindFunc = [funcName, extraKind, mode, erasedRunner](main::ClientContext* context, + const function::TableFuncBindInput* input) { + auto graphName = input->getLiteralVal(0); + auto extra = detail::bindExtraArgs(extraKind, input, funcName.c_str()); + 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("embedding", + detail::EmbeddingLogicalType::get())); + return std::unique_ptr(std::make_unique( + std::move(columns), std::move(graphEntry), binder::expression_vector{nodeOutput}, + std::make_unique(input->optionalParamsLegacy), funcName, std::move(graphName), + std::move(extra), erasedRunner, mode)); + }; + detail::appendExtraSignature(extraKind, inputTypes, paramNames, paramTypes); + auto func = std::make_unique(funcName, inputTypes); + detail::setInferInputTypes(extraKind, func.get()); + func->parameterNames = std::move(paramNames); + func->parameterTypes = std::move(paramTypes); + func->bindFunc = std::move(bindFunc); + func->tableFunc = std::move(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/include/function/gds_generic_params.h b/algo/src/include/function/gds_generic_params.h new file mode 100644 index 00000000..0ab7a264 --- /dev/null +++ b/algo/src/include/function/gds_generic_params.h @@ -0,0 +1,83 @@ +#pragma once + +// GDSGenericParams: the single OptionalParams implementation behind every row of +// the GDS registry. The Specs... pack *is* the row's parameter signature: each spec reuses +// function::OptionalParam, so parsing, literal evaluation, type coercion and validate() +// behave exactly like the hand-rolled Params structs it replaces. +// +// An empty pack means "no optional params" (any CALL-site optional param is an error) — +// this replaces the per-algorithm "throw on unknown param" boilerplate. + +#include + +#include "common/exception/binder.h" +#include "common/string_utils.h" +#include "function/table/optional_params.h" + +namespace lbug { +namespace algo_extension { + +template +static constexpr bool specIsRequired() { + if constexpr (requires { Spec::REQUIRED; }) { + return Spec::REQUIRED; + } else { + return false; + } +} + +template +struct GDSGenericParams final : public function::OptionalParams { + std::tuple...> params; + + explicit GDSGenericParams(const binder::expression_vector& optionalParams) { + for (auto& optionalParam : optionalParams) { + auto paramName = common::StringUtils::getLower(optionalParam->getAlias()); + const bool matched = (tryBind(paramName, optionalParam) || ... || false); + if (!matched) { + throw common::BinderException{ + "Unknown optional parameter: " + optionalParam->getAlias()}; + } + } + } + + // Copy shares the underlying expressions, matching the hand-rolled Params copy ctors. + GDSGenericParams(const GDSGenericParams&) = default; + + template + typename function::OptionalParam::T get() const { + return std::get>(params).getParamVal(); + } + + void evaluateParams(main::ClientContext* context) override { + std::apply([context](auto&... ps) { (ps.evaluateParam(context), ...); }, params); + (checkRequired(), ...); + } + + std::unique_ptr copy() override { + return std::make_unique(*this); + } + +private: + template + bool tryBind(const std::string& paramName, const std::shared_ptr& expr) { + if (paramName == Spec::NAME) { + std::get>(params) = function::OptionalParam(expr); + return true; + } + return false; + } + + template + void checkRequired() const { + if constexpr (specIsRequired()) { + if (!std::get>(params).isSet()) { + throw common::BinderException{ + std::string{"Missing required parameter: "} + Spec::NAME + "."}; + } + } + } +}; + +} // namespace algo_extension +} // namespace lbug diff --git a/algo/src/include/function/gds_param_specs.h b/algo/src/include/function/gds_param_specs.h new file mode 100644 index 00000000..54cba139 --- /dev/null +++ b/algo/src/include/function/gds_param_specs.h @@ -0,0 +1,158 @@ +#pragma once + +// Shared NetworKit knob specs for the table-driven GDS registry (gds_registry.cpp). +// +// Each spec is a tiny struct consumed by GDSGenericParams (gds_generic_params.h), +// which reuses function::OptionalParam for parsing, literal evaluation and validation. +// A spec needs only: +// static constexpr const char* NAME; // lower-case match against the CALL-site alias +// static constexpr LogicalTypeID TYPE; // DOUBLE, INT64 or BOOL +// static constexpr T DEFAULT_VALUE; +// optional: static void validate(T); // range check, run at bind time +// optional: static constexpr bool REQUIRED = true; // no usable default: CALL must pass it +// +// Reuse order: prefer the existing specs (function::DampingFactor / function::Tolerance from +// page_rank_config.h, MaxIterations from max_iterations_config.h) over adding near-duplicates +// here. Add a spec here only when no existing one matches the NetworKit default/semantics. + +#include "common/exception/binder.h" +#include "common/types/types.h" +#include "function/config/page_rank_config.h" + +namespace lbug { +namespace algo_extension { + +// Normalized flag shared by most centrality measures. NetworKit's default differs per +// algorithm (false for Betweenness/Degree, true for Closeness/Harmonic), so the default is a +// template parameter: spell Normalized (false) or NormalizedDefaultTrue (true) in the row to +// match the wrapped constructor's default. +template +struct NormalizedT { + static constexpr const char* NAME = "normalized"; + static constexpr common::LogicalTypeID TYPE = common::LogicalTypeID::BOOL; + static constexpr bool DEFAULT_VALUE = Default; +}; +using Normalized = NormalizedT; +using NormalizedDefaultTrue = NormalizedT; + +// Katz centrality knobs (NetworKit::KatzCentrality(G, alpha, beta, tol)). +struct KatzAlpha { + static constexpr const char* NAME = "alpha"; + static constexpr common::LogicalTypeID TYPE = common::LogicalTypeID::DOUBLE; + // 0 lets NetworKit pick a safe damping factor from the spectrum. + static constexpr double DEFAULT_VALUE = 0.0; + + static void validate(double alpha) { + if (alpha < 0) { + throw common::BinderException{"Alpha must be non-negative."}; + } + } +}; + +struct KatzBeta { + static constexpr const char* NAME = "beta"; + static constexpr common::LogicalTypeID TYPE = common::LogicalTypeID::DOUBLE; + static constexpr double DEFAULT_VALUE = 0.1; +}; + +// Leiden resolution (NetworKit::ParallelLeidenView). > 1 favors more/smaller communities. +struct LeidenGamma { + 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 common::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 common::BinderException{"Iterations must be positive."}; + } + } +}; + +// Label-propagation (NetworKit::PLP(G, theta, maxIterations)). NetworKit's default for both is +// `none` (no cap); -1 is the CALL-site sentinel for it. +struct PLPTheta { + static constexpr const char* NAME = "theta"; + static constexpr common::LogicalTypeID TYPE = common::LogicalTypeID::INT64; + static constexpr int64_t DEFAULT_VALUE = -1; + + static void validate(int64_t theta) { + if (theta < -1) { + throw common::BinderException{"Theta must be -1 (no cap) or non-negative."}; + } + } +}; + +struct PLPMaxIterations { + static constexpr const char* NAME = "maxiterations"; + static constexpr common::LogicalTypeID TYPE = common::LogicalTypeID::INT64; + static constexpr int64_t DEFAULT_VALUE = -1; + + static void validate(int64_t maxIterations) { + if (maxIterations < -1) { + throw common::BinderException{"Max iterations must be -1 (no cap) or non-negative."}; + } + } +}; + +// Cut clustering (NetworKit::CutClustering(G, alpha)) takes alpha as a *required* constructor +// argument, so the row declares it REQUIRED: CALL must pass alpha := .... DEFAULT_VALUE is +// only filler for OptionalParam's storage; evaluateParams() rejects an unset required param. +struct CutAlpha { + static constexpr const char* NAME = "alpha"; + static constexpr common::LogicalTypeID TYPE = common::LogicalTypeID::DOUBLE; + static constexpr double DEFAULT_VALUE = 0.0; + static constexpr bool REQUIRED = true; + + static void validate(double alpha) { + if (alpha < 0) { + throw common::BinderException{"Alpha must be non-negative."}; + } + } +}; + +// FastRP embedding knobs (NetworKit::FastRP). Weights stay fixed at the python default +// (1, 1, 1); dimension/seed/normalization are exposed. +struct FastRPDimension { + static constexpr const char* NAME = "dimension"; + static constexpr common::LogicalTypeID TYPE = common::LogicalTypeID::INT64; + static constexpr int64_t DEFAULT_VALUE = 128; + + static void validate(int64_t dimension) { + if (dimension <= 0) { + throw common::BinderException{"Dimension must be positive."}; + } + } +}; + +struct FastRPSeed { + static constexpr const char* NAME = "seed"; + static constexpr common::LogicalTypeID TYPE = common::LogicalTypeID::INT64; + static constexpr int64_t DEFAULT_VALUE = 42; + + static void validate(int64_t seed) { + if (seed < 0) { + throw common::BinderException{"Seed must be non-negative."}; + } + } +}; + +struct FastRPNormalization { + static constexpr const char* NAME = "normalization"; + static constexpr common::LogicalTypeID TYPE = common::LogicalTypeID::BOOL; + static constexpr bool DEFAULT_VALUE = true; +}; + +} // namespace algo_extension +} // namespace lbug diff --git a/algo/src/include/function/gds_registry.h b/algo/src/include/function/gds_registry.h new file mode 100644 index 00000000..59f9e424 --- /dev/null +++ b/algo/src/include/function/gds_registry.h @@ -0,0 +1,42 @@ +#pragma once + +// The GDS catalog: every icebug-backed table function in one place. +// +// Each row is {CALL name, builder}: the builder is a captureless factory returning the +// row's function_set (built by makeNodeScoreFunc / makePartitionFunc / makeEmbeddingFunc +// from gds_generic_algo.h). Wiring a new NetworKit algorithm is one row here — no new file, +// no CMake edit, no extension-load edit. +// +// Row checklist: +// - output column: "score" for scalar measures is not used; prefer the measure's name +// ("rank", "community_id", "component_id", "distance", "embedding", ...). +// - parameter signature: list the GDSGenericParams specs matching the NetworKit +// constructor's defaults (see gds_param_specs.h). An empty list means no optional params. +// - extra args: NONE (CALL f('G')), SOURCE_LIST (CALL f('G', [...])), SINGLE_SOURCE +// (CALL f('G', src)). +// - graph mode: UNDIRECTED unless the algorithm is direction-sensitive (BFS, ...). + +#include + +#include "function/function.h" + +namespace lbug { +namespace main { +class Database; +} // namespace main + +namespace algo_extension { + +struct GDSCatalogEntry { + const char* name; + function::function_set (*makeFunctionSet)(); +}; + +const std::vector& getGDSCatalog(); + +// Registers every catalog row on the database (mirrors ExtensionUtils::addFunc for +// runtime-built function sets). +void registerGDSFunctions(main::Database& database); + +} // namespace algo_extension +} // namespace lbug diff --git a/algo/src/main/algo_extension.cpp b/algo/src/main/algo_extension.cpp index 45346a23..f862ae74 100644 --- a/algo/src/main/algo_extension.cpp +++ b/algo/src/main/algo_extension.cpp @@ -4,6 +4,9 @@ #include "function/algo_function.h" #include "main/client_context.h" +#if defined(ICEBUG_ENABLED) +#include "function/gds_registry.h" +#endif namespace lbug { namespace algo_extension { @@ -37,11 +40,9 @@ void AlgoExtension::load(main::ClientContext* context) { ExtensionUtils::addTableFunc(db); ExtensionUtils::addTableFuncAlias(db); #if defined(ICEBUG_ENABLED) - ExtensionUtils::addTableFunc(db); - ExtensionUtils::addTableFunc(db); - ExtensionUtils::addTableFunc(db); - ExtensionUtils::addTableFunc(db); - ExtensionUtils::addTableFunc(db); + // Every icebug-backed GDS_* function comes from the registry table (gds_registry.cpp): + // adding an algorithm is one row there, nothing to wire up here. + registerGDSFunctions(db); #endif ExtensionUtils::addTableFunc(db); ExtensionUtils::addTableFuncAlias(db); diff --git a/algo/test/test_files/gds_registry.test b/algo/test/test_files/gds_registry.test new file mode 100644 index 00000000..1690c283 --- /dev/null +++ b/algo/test/test_files/gds_registry.test @@ -0,0 +1,167 @@ +-DATASET CSV empty + +-- + +-CASE GDSRegistryCentralityStar +-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 Every registry centrality row on a 3-leaf star: the hub dominates, the leaves tie. +-STATEMENT CALL GDS_BETWEENNESS('G') RETURN node.id, score ORDER BY node.id +---- 4 +0|6.000000 +1|0.000000 +2|0.000000 +3|0.000000 +-STATEMENT CALL GDS_DEGREE('G') RETURN node.id, score ORDER BY node.id +---- 4 +0|3.000000 +1|1.000000 +2|1.000000 +3|1.000000 +-STATEMENT CALL GDS_DEGREE('G', normalized := true) RETURN node.id, score ORDER BY node.id +---- 4 +0|0.750000 +1|0.250000 +2|0.250000 +3|0.250000 +-STATEMENT CALL GDS_CLOSENESS('G') RETURN node.id, score ORDER BY node.id +---- 4 +0|1.000000 +1|0.600000 +2|0.600000 +3|0.600000 +-STATEMENT CALL GDS_HARMONIC('G') RETURN node.id, score ORDER BY node.id +---- 4 +0|1.000000 +1|0.666667 +2|0.666667 +3|0.666667 +-STATEMENT CALL GDS_KATZ('G') RETURN node.id, score ORDER BY node.id +---- 4 +0|0.752296 +1|0.380373 +2|0.380373 +3|0.380373 +-STATEMENT CALL GDS_EIGENVECTOR('G') RETURN node.id, score ORDER BY node.id +---- 4 +0|0.866025 +1|0.288675 +2|0.288675 +3|0.288675 +-STATEMENT CALL GDS_CORE_DECOMPOSITION('G') RETURN node.id, core_number ORDER BY node.id +---- 4 +0|1.000000 +1|1.000000 +2|1.000000 +3|1.000000 +-STATEMENT CALL GDS_CONNECTED_COMPONENTS('G') RETURN node.id, component_id ORDER BY node.id +---- 4 +0|0 +1|0 +2|0 +3|0 +-LOG Directed BFS from leaf 1: only the hub is reachable downstream; the far leaves are -1. +-STATEMENT CALL GDS_BFS_DISTANCE('G', 1) RETURN node.id, distance ORDER BY node.id +---- 4 +0|1.000000 +1|0.000000 +2|-1.000000 +3|-1.000000 +-LOG Unknown optional params are rejected with the same message as the retired per-file code. +-STATEMENT CALL GDS_BETWEENNESS('G', bogus := 1) RETURN node.id, score +---- error +Binder exception: Unknown optional parameter: bogus +-LOG Single-source extras validate like the source list did: range then sign. +-STATEMENT CALL GDS_BFS_DISTANCE('G', 99) RETURN node.id, distance +---- error +Binder exception: GDS_BFS_DISTANCE source node offset 99 is out of range: the projected graph has 4 nodes. +-STATEMENT CALL GDS_BFS_DISTANCE('G', -1) RETURN node.id, distance +---- error +Binder exception: GDS_BFS_DISTANCE source node offset must be non-negative. + +-CASE GDSRegistryTwoCliques +-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 Partition rows assert structure, not exact ids: two communities, one per clique. +-STATEMENT CALL GDS_PLP('GT') RETURN COUNT(DISTINCT community_id) +---- 1 +2 +-STATEMENT CALL GDS_PLP('GT') WITH node, community_id WHERE node.id IN [0, 1, 2, 3] RETURN COUNT(DISTINCT community_id) +---- 1 +1 +-STATEMENT CALL GDS_PLP('GT') WITH node, community_id WHERE node.id IN [4, 5, 6, 7] RETURN COUNT(DISTINCT community_id) +---- 1 +1 +-STATEMENT CALL GDS_CUT_CLUSTERING('GT', alpha := 0.5) RETURN COUNT(DISTINCT community_id) +---- 1 +2 +-STATEMENT CALL GDS_CUT_CLUSTERING('GT', alpha := 0.5) WITH node, community_id WHERE node.id IN [0, 1, 2, 3] RETURN COUNT(DISTINCT community_id) +---- 1 +1 +-STATEMENT CALL GDS_CUT_CLUSTERING('GT', alpha := 0.5) WITH node, community_id WHERE node.id IN [4, 5, 6, 7] RETURN COUNT(DISTINCT community_id) +---- 1 +1 +-LOG CutClustering's alpha has no default: the row declares it REQUIRED. +-STATEMENT CALL GDS_CUT_CLUSTERING('GT') RETURN node.id, community_id +---- error +Binder exception: Missing required parameter: alpha. +-LOG One connected component across the bridge; directed BFS reaches the far clique in 2-3 hops. +-STATEMENT CALL GDS_CONNECTED_COMPONENTS('GT') RETURN COUNT(DISTINCT component_id) +---- 1 +1 +-STATEMENT CALL GDS_BFS_DISTANCE('GT', 0) RETURN node.id, distance ORDER BY node.id +---- 8 +0|0.000000 +1|1.000000 +2|1.000000 +3|1.000000 +4|2.000000 +5|3.000000 +6|3.000000 +7|3.000000