feat(algo): descriptor-template GDS layer — GDS_LEIDEN, GDS_LOUVAIN, GDS_PPR

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 gds_leiden.cpp gds_ppr.cpp)
 endif ()
 
 add_library(lbug_algo_function

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 <networkit/community/ParallelLeidenView.hpp>
+#include <networkit/structures/Partition.hpp>
+
+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<LeidenGamma> gamma;
+    OptionalParam<LeidenIterations> iterations;
+
+    explicit GDSLeidenOptionalParams(const expression_vector& optionalParams) {
+        for (auto& optionalParam : optionalParams) {
+            auto paramName = StringUtils::getLower(optionalParam->getAlias());
+            if (paramName == LeidenGamma::NAME) {
+                gamma = function::OptionalParam<LeidenGamma>(optionalParam);
+            } else if (paramName == LeidenIterations::NAME) {
+                iterations = function::OptionalParam<LeidenIterations>(optionalParam);
+            } else {
+                throw BinderException{"Unknown optional parameter: " + optionalParam->getAlias()};
+            }
+        }
+    }
+
+    GDSLeidenOptionalParams(OptionalParam<LeidenGamma> gamma,
+        OptionalParam<LeidenIterations> iterations)
+        : gamma{std::move(gamma)}, iterations{std::move(iterations)} {}
+
+    void evaluateParams(main::ClientContext* context) override {
+        gamma.evaluateParam(context);
+        iterations.evaluateParam(context);
+    }
+
+    std::unique_ptr<function::OptionalParams> copy() override {
+        return std::make_unique<GDSLeidenOptionalParams>(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<int64_t> run(const NetworKit::GraphR& g, offset_t numNodes,
+        const Params& params, const Extra::type&) {
+        NetworKit::ParallelLeidenView leiden(g, static_cast<int>(params.iterations.getParamVal()),
+            /*randomize=*/true, params.gamma.getParamVal());
+        leiden.run();
+        const auto& partition = leiden.getPartition();
+        std::vector<int64_t> communityIds(numNodes, -1);
+        for (offset_t i = 0; i < numNodes && i < partition.numberOfElements(); ++i) {
+            communityIds[i] = static_cast<int64_t>(partition[i]);
+        }
+        return communityIds;
+    }
+};
+
+function_set GDSLeidenFunction::getFunctionSet() {
+    return GDSPerNodeScalarAlgo<GDSLeidenDesc>::getFunctionSet();
+}
+
+} // namespace algo_extension
+} // namespace lbug

algo/src/function/gds_louvain.cpp

@@ -0,0 +1,54 @@
+// 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 <networkit/community/PLM.hpp>
+#include <networkit/structures/Partition.hpp>
+
+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<function::OptionalParams> copy() override {
+        return std::make_unique<GDSLouvainOptionalParams>(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<int64_t> 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<int64_t> communityIds(numNodes, -1);
+        for (offset_t i = 0; i < numNodes && i < partition.numberOfElements(); ++i) {
+            communityIds[i] = static_cast<int64_t>(partition[i]);
+        }
+        return communityIds;
+    }
+};
+
+function_set GDSLouvainFunction::getFunctionSet() {
+    return GDSPerNodeScalarAlgo<GDSLouvainDesc>::getFunctionSet();
+}
+
+} // namespace algo_extension
+} // namespace lbug

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 <arrow/api.h>
+#include "function/gds_algo_template.h"
 #include <networkit/centrality/PageRank.hpp>
-#include <networkit/graph/GraphR.hpp>
 
-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<Expression> nodeOutput,
-        std::unique_ptr<GDSPageRankOptionalParams> 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<TableFuncBindData> copy() const override {
-        return std::make_unique<GDSPageRankBindData>(*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<double>& 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<nodeID_t>(0, nodeID_t{i, tableID});
-            rankVector->setValue<double>(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<double> 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<VertexCompute> copy() override {
-        return std::make_unique<GDSPageRankResultVertexCompute>(mm, sharedState, scores);
-    }
-
-private:
-    const std::vector<double>& scores;
-    std::unique_ptr<ValueVector> nodeIDVector;
-    std::unique_ptr<ValueVector> 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<GDSFuncSharedState>();
-    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<GDSPageRankBindData>();
-    auto& config = bindData->optionalParams->constCast<GDSPageRankOptionalParams>();
-
-    // 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<double>& scores = pr.scores();
-
-    // 4. Stream scores back through the GDS result pipeline.
-    auto outputVC = std::make_unique<GDSPageRankResultVertexCompute>(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<TableFuncBindData> bindFunc(main::ClientContext* context,
-    const TableFuncBindInput* input) {
-    auto graphName = input->getLiteralVal<std::string>(0);
-    auto graphEntry = GDSFunction::bindGraphEntry(*context, graphName);
-    auto nodeOutput = GDSFunction::bindNodeOutput(*input, graphEntry.getNodeEntries());
-    expression_vector columns;
-    columns.push_back(nodeOutput->constCast<NodeExpression>().getInternalID());
-    columns.push_back(input->binder->createVariable(RANK_COLUMN_NAME, LogicalType::DOUBLE()));
-    return std::make_unique<GDSPageRankBindData>(std::move(columns), std::move(graphEntry),
-        nodeOutput, std::make_unique<GDSPageRankOptionalParams>(input->optionalParamsLegacy),
-        std::move(graphName));
-}
-
 function_set GDSPageRankFunction::getFunctionSet() {
-    function_set result;
-    auto func = std::make_unique<TableFunction>(GDSPageRankFunction::name,
-        std::vector<LogicalTypeID>{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<GDSPageRankDesc>::getFunctionSet();
 }
 
 } // namespace algo_extension