feat(algo): add GDS_NODE2VEC (icebug/NetworKit node embeddings)

algo/src/function/CMakeLists.txt

@@ -8,6 +8,7 @@ add_library(lbug_algo_function
         weakly_connected_components.cpp
         page_rank.cpp
         gds_page_rank.cpp
+        gds_node2vec.cpp
         k_core_decomposition.cpp
         louvain.cpp
         spanning_forest.cpp

algo/src/function/gds_node2vec.cpp

@@ -0,0 +1,181 @@
+// 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 "common/in_mem_graph.h"
+#include "function/algo_function.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 <networkit/embedding/Node2Vec.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 {
+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
+
+// Materialize the projected graph's undirected adjacency as an InMemGraph CSR (same as Louvain).
+static void buildCSR(table_id_t tableID, offset_t numNodes, Graph* graph, InMemGraph& inMem) {
+    const auto nbrTables = graph->getRelInfos(tableID);
+    const auto nbrInfo = nbrTables[0];
+    const auto scanState = graph->prepareRelScan(*nbrInfo.relGroupEntry, nbrInfo.relTableID,
+        nbrInfo.dstTableID, {}, false /*randomLookup*/);
+    for (offset_t nodeId = 0; nodeId < numNodes; ++nodeId) {
+        inMem.initNextNode();
+        const nodeID_t nid = {nodeId, tableID};
+        for (auto chunk : graph->scanFwd(nid, *scanState)) {
+            chunk.forEach(
+                [&](auto neighbors, auto, auto i) { inMem.insertNbr(neighbors[i].offset); });
+        }
+        for (auto chunk : graph->scanBwd(nid, *scanState)) {
+            chunk.forEach([&](auto neighbors, auto, auto i) {
+                if (neighbors[i].offset != nodeId) {
+                    inMem.insertNbr(neighbors[i].offset);
+                }
+            });
+        }
+    }
+    inMem.initNextNode(); // trailing sentinel
+}
+
+static std::shared_ptr<arrow::UInt64Array> toU64(const std::function<offset_t(offset_t)>& at,
+    offset_t count) {
+    // Use the system (malloc) pool, not Arrow's mimalloc-backed default: mimalloc's per-thread
+    // init is not safe on threads that predate libarrow's dlopen (see gds_page_rank.cpp).
+    arrow::UInt64Builder builder(arrow::system_memory_pool());
+    (void)builder.Reserve(count);
+    for (offset_t i = 0; i < count; ++i) {
+        (void)builder.Append(static_cast<uint64_t>(at(i)));
+    }
+    std::shared_ptr<arrow::Array> arr;
+    (void)builder.Finish(&arr);
+    return std::static_pointer_cast<arrow::UInt64Array>(arr);
+}
+
+// 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<std::vector<float>>& 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<nodeID_t>(0, nodeID_t{i, tableID});
+            const auto& emb = i < features.size() ? features[i] : empty;
+            auto entry = ListVector::addList(embVector.get(), emb.size());
+            embVector->setValue<list_entry_t>(0, entry);
+            auto* dataVector = ListVector::getDataVector(embVector.get());
+            for (auto k = 0u; k < emb.size(); ++k) {
+                dataVector->setValue<float>(entry.offset + k, emb[k]);
+            }
+            localFT->append(vectors);
+        }
+    }
+
+    std::unique_ptr<VertexCompute> copy() override {
+        return std::make_unique<GDSNode2VecResultVertexCompute>(mm, sharedState, features);
+    }
+
+private:
+    const std::vector<std::vector<float>>& features;
+    const std::vector<float> empty;
+    std::unique_ptr<ValueVector> nodeIDVector;
+    std::unique_ptr<ValueVector> 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<GDSFuncSharedState>();
+    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);
+
+    // 1. Ladybug engine graph -> InMemGraph CSR -> Arrow UInt64 CSR.
+    InMemGraph inMem(numNodes, mm);
+    buildCSR(tableID, numNodes, graph, inMem);
+    auto outIndptr = toU64([&](offset_t i) { return inMem.csrOffsets[i]; }, numNodes + 1);
+    auto outIndices =
+        toU64([&](offset_t i) { return inMem.csrEdges[i].neighbor; }, inMem.csrEdges.size());
+
+    // 2. icebug: zero-copy GraphR, then Node2Vec.
+    NetworKit::GraphR g(numNodes, /*directed=*/false, outIndices, outIndptr);
+    NetworKit::Node2Vec n2v(g, N2V_P, N2V_Q, N2V_WALK_LEN, N2V_NUM_WALKS, N2V_DIM);
+    n2v.run();
+    const std::vector<std::vector<float>>& features = n2v.getFeatures();
+
+    // 3. Stream embeddings back through the GDS result pipeline.
+    auto outputVC = std::make_unique<GDSNode2VecResultVertexCompute>(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<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(EMBEDDING_COLUMN_NAME,
+        LogicalType::LIST(LogicalType::FLOAT())));
+    return std::make_unique<GDSBindData>(std::move(columns), std::move(graphEntry),
+        expression_vector{nodeOutput});
+}
+
+function_set GDSNode2VecFunction::getFunctionSet() {
+    function_set result;
+    auto func = std::make_unique<TableFunction>(GDSNode2VecFunction::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;
+}
+
+} // namespace algo_extension
+} // namespace lbug

algo/src/include/function/algo_function.h

@@ -60,6 +60,13 @@ struct GDSPageRankFunction {
     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();
+};
+
 struct KCoreDecompositionFunction {
     static constexpr const char* name = "K_CORE_DECOMPOSITION";
 

algo/src/main/algo_extension.cpp

@@ -19,6 +19,7 @@ void AlgoExtension::load(main::ClientContext* context) {
     ExtensionUtils::addTableFunc<PageRankFunction>(db);
     ExtensionUtils::addTableFuncAlias<PageRankAliasFunction>(db);
     ExtensionUtils::addTableFunc<GDSPageRankFunction>(db);
+    ExtensionUtils::addTableFunc<GDSNode2VecFunction>(db);
     ExtensionUtils::addTableFunc<KCoreDecompositionFunction>(db);
     ExtensionUtils::addTableFuncAlias<KCoreDecompositionAliasFunction>(db);
     ExtensionUtils::addTableFunc<LouvainFunction>(db);

algo/test/test_files/gds_node2vec.test

@@ -0,0 +1,29 @@
+-DATASET CSV empty
+
+--
+
+-CASE GDSNode2VecStar
+-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 Node2Vec is stochastic, so we assert the output shape (one 64-dim embedding per node)
+-LOG rather than exact values. The semantic property (structurally-similar nodes have
+-LOG higher cosine similarity) is validated separately.
+-STATEMENT CALL GDS_NODE2VEC('G') RETURN node.id, size(embedding) AS dim ORDER BY node.id
+---- 4
+0|64
+1|64
+2|64
+3|64