diff --git a/algo/src/function/CMakeLists.txt b/algo/src/function/CMakeLists.txt index 1fea808e..d98602d9 100644 --- a/algo/src/function/CMakeLists.txt +++ b/algo/src/function/CMakeLists.txt @@ -9,6 +9,7 @@ add_library(lbug_algo_function page_rank.cpp k_core_decomposition.cpp louvain.cpp + leiden.cpp spanning_forest.cpp ) diff --git a/algo/src/function/leiden.cpp b/algo/src/function/leiden.cpp new file mode 100644 index 00000000..a80bb087 --- /dev/null +++ b/algo/src/function/leiden.cpp @@ -0,0 +1,844 @@ +// LadybugDB ALGO Extension — Leiden community detection +// +// Full implementation: Local Moving (inherited from Louvain pattern) + +// Refinement (Leiden-unique) + Aggregation (shared) +// +// Reference: Traag, Waltman & van Eck (2019) +// "From Louvain to Leiden: guaranteeing well-connected communities" +// https://www.nature.com/articles/s41598-019-41695-z + +#include "function/leiden.h" + +#include "binder/binder.h" +#include "common/in_mem_gds_utils.h" +#include "common/in_mem_graph.h" +#include "common/task_system/progress_bar.h" +#include "common/types/types.h" +#include "function/algo_function.h" +#include "function/config/louvain_config.h" +#include "function/config/max_iterations_config.h" +#include "function/gds/gds.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 +#include +#include +#include +#include + +using namespace std; +using namespace lbug::binder; +using namespace lbug::common; +using namespace lbug::processor; +using namespace lbug::storage; +using namespace lbug::graph; +using namespace lbug::function; + +namespace lbug { +namespace algo_extension { +namespace { // anonymous — avoid link conflict with louvain.cpp's identical symbols + +// ═══════════════════════════════════════════════════════════════════ +// Section 1: Shared data structures (inherited from louvain.cpp) +// ═══════════════════════════════════════════════════════════════════ + +constexpr double THRESHOLD = 1e-6; +constexpr offset_t UNASSIGNED_COMM = numeric_limits::max(); + +static constexpr const char* LEIDEN_COLUMN_NAME = "community_id"; +using LeidenBindData = GDSBindData; + +struct CommInfo { + atomic size; + atomic degree; + + CommInfo() : size{0}, degree(0) {} + CommInfo(const CommInfo& other) { + size.store(other.size.load()); + degree.store(other.degree.load()); + } + CommInfo& operator=(const CommInfo& other) { + if (this != &other) { + size.store(other.size.load()); + degree.store(other.degree.load()); + } + return *this; + } +}; + +struct PhaseState { + InMemGraph graph; + AtomicObjectArray acceptedComm; + AtomicObjectArray currComm; + AtomicObjectArray nextComm; + ObjectArray currCommInfos; + ObjectArray nextCommInfos; + AtomicObjectArray nodeWeightedDegrees; + AtomicObjectArray selfCommWeights; + weight_t totalWeight = 0; + double modularityConstant = 0.0; + + PhaseState(const offset_t numNodes, MemoryManager* mm, ExecutionContext* context) + : graph{InMemGraph(numNodes, mm)} { + reinit(numNodes, mm, context); + } + DELETE_BOTH_COPY(PhaseState); + + void reinit(offset_t numNodes, MemoryManager* mm, ExecutionContext* context); + void startNewIter(MemoryManager* mm, ExecutionContext* context); + + void initNextNode(const offset_t nodeId) { + graph.initNextNode(); + currCommInfos.getUnsafe(nodeId).size.store(1, memory_order_relaxed); + currCommInfos.getUnsafe(nodeId).degree.store(0, memory_order_relaxed); + acceptedComm.set(nodeId, nodeId, memory_order_relaxed); + currComm.set(nodeId, nodeId, memory_order_relaxed); + } + + void insertNbr(const offset_t from, const offset_t to, const weight_t weight = DEFAULT_WEIGHT) { + graph.insertNbr(to, weight); + nodeWeightedDegrees.fetchAdd(from, weight, memory_order_relaxed); + currCommInfos.getUnsafe(from).degree.fetch_add(weight, memory_order_relaxed); + totalWeight += weight; + } + + void finalize() { graph.initNextNode(); } +}; + +// ═══════════════════════════════════════════════════════════════════ +// Section 2: Parallel VC helpers (inherited from louvain.cpp) +// ═══════════════════════════════════════════════════════════════════ + +class ResetPhaseStateVC final : public InMemParallelCompute { +public: + explicit ResetPhaseStateVC(PhaseState& state) : state{state} {} + ~ResetPhaseStateVC() override = default; + + void parallelCompute(const offset_t startOffset, const offset_t endOffset, + const optional&) override { + for (auto nodeId = startOffset; nodeId < endOffset; ++nodeId) { + state.nodeWeightedDegrees.set(nodeId, 0, memory_order_relaxed); + state.currCommInfos.set(nodeId, CommInfo()); + state.acceptedComm.set(nodeId, UNASSIGNED_COMM, memory_order_relaxed); + state.currComm.set(nodeId, UNASSIGNED_COMM, memory_order_relaxed); + state.nextComm.set(nodeId, UNASSIGNED_COMM, memory_order_relaxed); + } + } + unique_ptr copy() override { + return make_unique(state); + } +private: + PhaseState& state; +}; + +class StartNewIterVC final : public InMemParallelCompute { +public: + explicit StartNewIterVC(PhaseState& state) : state{state} {} + ~StartNewIterVC() override = default; + + void parallelCompute(const offset_t startOffset, const offset_t endOffset, + const optional&) override { + for (auto nodeId = startOffset; nodeId < endOffset; ++nodeId) { + state.selfCommWeights.set(nodeId, 0, memory_order_relaxed); + state.nextCommInfos.set(nodeId, CommInfo()); + } + } + unique_ptr copy() override { + return make_unique(state); + } +private: + PhaseState& state; +}; + +void PhaseState::reinit(const offset_t numNodes, MemoryManager* mm, ExecutionContext* context) { + totalWeight = 0; + graph.reinit(numNodes); + nodeWeightedDegrees.reallocate(numNodes, mm); + currCommInfos.reallocate(numNodes, mm); + acceptedComm.reallocate(numNodes, mm); + currComm.reallocate(numNodes, mm); + nextComm.reallocate(numNodes, mm); + + ResetPhaseStateVC resetPhaseStateVC(*this); + InMemGDSUtils::runParallelCompute(resetPhaseStateVC, numNodes, context); +} + +void PhaseState::startNewIter(MemoryManager* mm, ExecutionContext* context) { + selfCommWeights.reallocate(graph.numNodes, mm); + nextCommInfos.reallocate(graph.numNodes, mm); + + StartNewIterVC startNewIterVC(*this); + InMemGDSUtils::runParallelCompute(startNewIterVC, graph.numNodes, context); + + modularityConstant = 1.0 / totalWeight; +} + +// ═══════════════════════════════════════════════════════════════════ +// Section 3: Louvain Phase 1 — Parallel Local Moving (inherited) +// ═══════════════════════════════════════════════════════════════════ + +class RunIterationVC final : public InMemParallelCompute { +public: + explicit RunIterationVC(PhaseState& state) : state{state} {} + ~RunIterationVC() override = default; + + void parallelCompute(const offset_t startOffset, const offset_t endOffset, + const optional&) override { + vector intraCommWeights; + unordered_map commToWeightsIndex; + for (auto nodeId = startOffset; nodeId < endOffset; ++nodeId) { + const auto startCSROffset = state.graph.csrOffsets[nodeId]; + const auto endCSROffset = state.graph.csrOffsets[nodeId + 1]; + offset_t targetCommId = UNASSIGNED_COMM; + if (startCSROffset != endCSROffset) { + commToWeightsIndex.clear(); + intraCommWeights.clear(); + const weight_t selfLoopWeight = computeIntraCommWeights(nodeId, startCSROffset, + endCSROffset, intraCommWeights, commToWeightsIndex); + targetCommId = findPotentialNewComm(nodeId, selfLoopWeight, intraCommWeights, + commToWeightsIndex); + state.selfCommWeights.set(nodeId, intraCommWeights[0], memory_order_relaxed); + } + state.nextComm.set(nodeId, targetCommId, memory_order_relaxed); + const auto currCommId = state.currComm.get(nodeId, memory_order_relaxed); + if (targetCommId != currCommId && targetCommId != UNASSIGNED_COMM) { + const auto nodeDegree = state.nodeWeightedDegrees.get(nodeId, memory_order_relaxed); + state.nextCommInfos.getUnsafe(targetCommId).degree.fetch_add(nodeDegree); + state.nextCommInfos.getUnsafe(targetCommId).size.fetch_add(1); + state.nextCommInfos.getUnsafe(currCommId).degree.fetch_sub(nodeDegree); + state.nextCommInfos.getUnsafe(currCommId).size.fetch_sub(1); + } + } + } + + weight_t computeIntraCommWeights(const offset_t nodeId, const offset_t startCSROffset, + const offset_t endCSROffset, vector& intraCommWeights, + unordered_map& commToWeightsIndex) const { + weight_t selfLoopWeight = 0; + const auto currComm = state.currComm.get(nodeId, memory_order_relaxed); + commToWeightsIndex[currComm] = 0; + intraCommWeights.push_back(0); + offset_t nextIndex = 1; + for (auto offset = startCSROffset; offset < endCSROffset; offset++) { + auto nbrEntry = state.graph.csrEdges[offset]; + if (nbrEntry.neighbor == nodeId) { + selfLoopWeight += nbrEntry.weight; + } + auto nbrCommId = state.currComm.get(nbrEntry.neighbor, memory_order_relaxed); + if (!commToWeightsIndex.contains(nbrCommId)) { + commToWeightsIndex[nbrCommId] = nextIndex; + nextIndex++; + intraCommWeights.push_back(nbrEntry.weight); + } else { + intraCommWeights[commToWeightsIndex[nbrCommId]] += nbrEntry.weight; + } + } + return selfLoopWeight; + } + + offset_t findPotentialNewComm(const offset_t nodeId, const weight_t selfLoopWeight, + const vector& intraCommWeights, + unordered_map commToWeightsIndex) const { + const auto currComm = state.currComm.get(nodeId, memory_order_relaxed); + const auto degree = + static_cast(state.nodeWeightedDegrees.get(nodeId, memory_order_relaxed)); + auto newComm = currComm; + double newCommModGain = 0.0; + const auto prevIntraCommWeights = static_cast(intraCommWeights[0] - selfLoopWeight); + const auto prevWeightedDegrees = + static_cast( + state.currCommInfos.getUnsafe(currComm).degree.load(memory_order_relaxed)) - + degree; + for (auto [nbrCommId, weightIndex] : commToWeightsIndex) { + if (currComm != nbrCommId) { + const auto newIntraCommWeights = static_cast(intraCommWeights[weightIndex]); + const auto newWeightedDegrees = static_cast( + state.currCommInfos.getUnsafe(nbrCommId).degree.load(memory_order_relaxed)); + const auto changeIntraWeights = 2 * (newIntraCommWeights - prevIntraCommWeights); + const auto changeSumWeightedDegrees = 2 * degree * state.modularityConstant * + (newWeightedDegrees - prevWeightedDegrees); + const auto modGain = changeIntraWeights - changeSumWeightedDegrees; + if (modGain > newCommModGain || ((newCommModGain - modGain) < THRESHOLD && + modGain != 0 && (nbrCommId < newComm))) { + newCommModGain = modGain; + newComm = nbrCommId; + } + } + } + if (state.currCommInfos.getUnsafe(newComm).size.load(memory_order_relaxed) == 1 && + state.currCommInfos.getUnsafe(currComm).size.load(memory_order_relaxed) == 1 && + newComm > currComm) { + newComm = currComm; + } + return newComm; + } + + unique_ptr copy() override { + return make_unique(state); + } +private: + PhaseState& state; +}; + +// ═══════════════════════════════════════════════════════════════════ +// Section 4: Graph I/O — multi-table support +// ═══════════════════════════════════════════════════════════════════ +// +// Builds an in-memory CSR graph from the GDS projection graph. +// Supports heterogeneous graphs with multiple node tables by using +// a flat index mapping: flatIdx = nodeOffsetBase[tableID] + offset. +// All PhaseState arrays are sized for totalNodes across all tables. + +static void initInMemoryGraph(const vector& nodeTableIDs, + const table_id_map_t& nodeOffsetBase, offset_t totalNodes, + Graph* graph, transaction::Transaction* transaction, PhaseState& state) { + // Find the first relationship info from any node table + table_id_t srcTableID = INVALID_TABLE_ID; + table_id_t dstTableID = INVALID_TABLE_ID; + const catalog::TableCatalogEntry* relGroupEntry = nullptr; + oid_t relTableID = 0; + for (auto tid : nodeTableIDs) { + auto nbrTables = graph->getRelInfos(tid); + if (!nbrTables.empty()) { + srcTableID = nbrTables[0].srcTableID; + dstTableID = nbrTables[0].dstTableID; + relGroupEntry = nbrTables[0].relGroupEntry; + relTableID = nbrTables[0].relTableID; + break; + } + } + + if (relGroupEntry == nullptr) { + // No edges in this graph — just initialize all nodes + for (auto tableID : nodeTableIDs) { + auto numNodes = graph->getMaxOffset(transaction, tableID); + offset_t baseOffset = nodeOffsetBase.at(tableID); + for (auto nodeId = 0u; nodeId < numNodes; ++nodeId) { + state.initNextNode(baseOffset + nodeId); + } + } + state.finalize(); + return; + } + + // Prepare FWD scan state (neighbors are in dstTableID) + auto fwdState = graph->prepareRelScan(*relGroupEntry, relTableID, dstTableID, {}, false); + // Prepare BWD scan state (neighbors are in srcTableID) + auto bwdState = graph->prepareRelScan(*relGroupEntry, relTableID, srcTableID, {}, false); + + // Process all node tables + for (auto tableID : nodeTableIDs) { + auto numNodes = graph->getMaxOffset(transaction, tableID); + offset_t baseOffset = nodeOffsetBase.at(tableID); + + for (auto nodeId = 0u; nodeId < numNodes; ++nodeId) { + offset_t flatNodeIdx = baseOffset + nodeId; + state.initNextNode(flatNodeIdx); + const nodeID_t nextNodeId = {nodeId, tableID}; + + // FWD scan: outgoing neighbors (convert to flat indices) + for (auto chunk : graph->scanFwd(nextNodeId, *fwdState)) { + chunk.forEach([&](auto neighbors, auto, auto i) { + offset_t flatNbr = nodeOffsetBase.at(neighbors[i].tableID) + neighbors[i].offset; + state.insertNbr(flatNodeIdx, flatNbr); + }); + } + // BWD scan: incoming neighbors (convert to flat indices) + for (auto chunk : graph->scanBwd(nextNodeId, *bwdState)) { + chunk.forEach([&](auto neighbors, auto, auto i) { + offset_t flatNbr = nodeOffsetBase.at(neighbors[i].tableID) + neighbors[i].offset; + if (flatNbr != flatNodeIdx) { + state.insertNbr(flatNodeIdx, flatNbr); + } + }); + } + } + } + state.finalize(); +} + +static offset_t renumberCommunities(PhaseState& state) { + unordered_map map; + offset_t nextCommId = 0; + for (auto nodeId = 0LU; nodeId < state.graph.numNodes; ++nodeId) { + auto commId = state.acceptedComm.get(nodeId, memory_order_relaxed); + if (commId == UNASSIGNED_COMM) continue; + if (!map.contains(commId)) { + map.insert(make_pair(commId, nextCommId)); + nextCommId++; + } + state.acceptedComm.set(nodeId, map.at(commId), memory_order_relaxed); + } + return nextCommId; +} + +static void aggregateCommunities(const offset_t newCommCount, PhaseState& state, + MemoryManager* mm, ExecutionContext* context) { + vector_t> commWeights(mm); + commWeights.resize(newCommCount); + for (auto nodeId = 0u; nodeId < state.graph.numNodes; nodeId++) { + const auto beginCSROffset = state.graph.csrOffsets[nodeId]; + const auto endCSROffset = state.graph.csrOffsets[nodeId + 1]; + auto commId = state.acceptedComm.get(nodeId, memory_order_relaxed); + for (auto offset = beginCSROffset; offset < endCSROffset; ++offset) { + const auto nbr = state.graph.csrEdges[offset]; + auto nbrCommId = state.acceptedComm.get(nbr.neighbor, memory_order_relaxed); + if (commId >= nbrCommId) { + commWeights[commId][nbrCommId] += nbr.weight; + if (commId != nbrCommId) { + commWeights[nbrCommId][commId] += nbr.weight; + } + } + } + } + state.reinit(newCommCount, mm, context); + for (auto nodeId = 0u; nodeId < newCommCount; nodeId++) { + state.initNextNode(nodeId); + for (auto [nbrId, weight] : commWeights[nodeId]) { + state.insertNbr(nodeId, nbrId, weight); + } + } + state.finalize(); +} + +// ═══════════════════════════════════════════════════════════════════ +// Section 5: Leiden Phase 2 — Refinement (the ONLY new algorithm) +// ═══════════════════════════════════════════════════════════════════ +// +// Uses its own refined[] array (never touches state.currComm) to avoid +// polluting Phase 1's Local Moving state for subsequent phases. +// +// Refinement guarantees that every community is internally connected, +// which is the key differentiator from Louvain. +// +// Algorithm (from Traag et al. 2019, Algorithm 2): +// 1. Each node starts in its own refined community. +// 2. Constrained Local Moving: nodes can only merge within the same +// PARENT community (from Phase 1). +// 3. Connectivity guarantee: reject moves that would disconnect the +// source refined community (verified via BFS). +// 4. Singleton merge: lone nodes in a refined community get merged +// into a random neighbor's refined community (avoids oversplitting). + +// Edge-to-community weight map inside a single parent community. +// Key = refined community ID, Value = sum of edge weights from `node` to that community. +using CommunityWeightMap = unordered_map; + +// BFS: check whether community `comm` in `refined` stays connected after removing `node`. +static bool staysConnectedAfterRemoval(const InMemGraph& g, + const vector& refined, offset_t node, offset_t comm) { + vector members; + for (offset_t i = 0; i < g.numNodes; i++) + if (i != node && refined[i] == comm) members.push_back(i); + if (members.empty()) return true; + + unordered_set visited; + deque q; + q.push_back(members[0]); + visited.insert(members[0]); + + while (!q.empty()) { + offset_t cur = q.front(); q.pop_front(); + for (auto csrIdx = g.csrOffsets[cur]; csrIdx < g.csrOffsets[cur + 1]; csrIdx++) { + offset_t nbr = g.csrEdges[csrIdx].neighbor; + if (refined[nbr] == comm && !visited.contains(nbr)) { + visited.insert(nbr); + q.push_back(nbr); + } + } + } + return visited.size() == members.size(); +} + +// Per-node metadata for Refinement: {refinedComm, weightToSelf, weightToOthers} +struct RefineNodeMeta { + offset_t refinedComm; + double degree; // = state.nodeWeightedDegrees[node] + CommunityWeightMap nbrWeights; // total edge weight to each refined community (within parent) +}; + +static void refinePartition(PhaseState& state, + vector& refined, // IN/OUT + const vector& parentComm) // IN: Phase 1 membership per node +{ + offset_t n = state.graph.numNodes; + if (n < 2) return; + + // 1. Each node starts in its own refined community. + refined.resize(n); + for (offset_t i = 0; i < n; i++) refined[i] = i; + + // 2. Build per-parent-community node lists. + unordered_map> commNodes; + for (offset_t i = 0; i < n; i++) commNodes[parentComm[i]].push_back(i); + + // 3. Per-node degree cache. + vector nodeDegree(n); + for (offset_t i = 0; i < n; i++) + nodeDegree[i] = static_cast( + state.nodeWeightedDegrees.get(i, memory_order_relaxed)); + + // 4. Refinement degree tracking: commDeg[c] = sum of nodeDegree for nodes in refined community c. + unordered_map commDeg; + + // Initialize each singleton refined community's degree. + for (offset_t i = 0; i < n; i++) commDeg[i] = nodeDegree[i]; + + // Helper: apply a move of node `u` from `fromC` to `toC`. + auto applyMove = [&](offset_t u, offset_t fromC, offset_t toC) { + refined[u] = toC; + commDeg[fromC] -= nodeDegree[u]; + commDeg[toC] += nodeDegree[u]; + }; + + static mt19937 gen(random_device{}()); + + // 5. Process each parent community independently. + for (auto& [pc, nodes] : commNodes) { + if (nodes.size() <= 1) continue; + + shuffle(nodes.begin(), nodes.end(), gen); + int maxIters = 10; + bool changed = true; + + while (changed && maxIters-- > 0) { + changed = false; + + for (offset_t u : nodes) { + offset_t curC = refined[u]; + CommunityWeightMap nbrWeights; + + // Collect edge weights to refined communities (within same parent only). + for (auto csrIdx = state.graph.csrOffsets[u]; + csrIdx < state.graph.csrOffsets[u + 1]; csrIdx++) { + const auto& nbr = state.graph.csrEdges[csrIdx]; + if (parentComm[nbr.neighbor] != pc) continue; + offset_t nbrC = refined[nbr.neighbor]; + nbrWeights[nbrC] += static_cast(nbr.weight); + } + + // Find best target refined community. + offset_t bestC = curC; + double bestDelta = 0.0; + + for (auto& [nbrC, wTo] : nbrWeights) { + if (nbrC == curC) continue; + + // Connectivity check: will `curC` stay connected after u leaves? + if (!staysConnectedAfterRemoval(state.graph, refined, u, curC)) + continue; + + // Modularity delta (simplified): + // ΔQ = (w_to - w_from) / (2m) + // - degree(u) * (deg_to - deg_from + degree(u)) / (2m)² + double wFrom = nbrWeights[curC]; // 0 if curC not in nbrWeights + double degFrom = commDeg[curC]; + double degTo = commDeg[nbrC]; + double mConst = state.modularityConstant; + + double delta = (wTo - wFrom) * mConst + - nodeDegree[u] * (degTo - degFrom + nodeDegree[u]) * mConst * mConst; + + if (delta > bestDelta) { bestDelta = delta; bestC = nbrC; } + } + + if (bestC != curC && bestDelta > 0) { + applyMove(u, curC, bestC); + changed = true; + } + } + } + + // 6. Merge singletons: a node alone in its refined community gets merged + // into a random neighbor's refined community (within the same parent). + for (offset_t u : nodes) { + offset_t myC = refined[u]; + // Count members of `myC` within this parent community. + int count = 0; + for (offset_t v : nodes) if (refined[v] == myC) count++; + if (count > 1) continue; // not a singleton + + // Find any neighbor's refined community within the same parent. + offset_t mergeTarget = UNASSIGNED_COMM; + for (auto csrIdx = state.graph.csrOffsets[u]; + csrIdx < state.graph.csrOffsets[u + 1]; csrIdx++) { + offset_t nbr = state.graph.csrEdges[csrIdx].neighbor; + if (parentComm[nbr] != pc) continue; + if (refined[nbr] != myC) { mergeTarget = refined[nbr]; break; } + } + if (mergeTarget == UNASSIGNED_COMM) continue; + + applyMove(u, myC, mergeTarget); + } + } +} + +// ═══════════════════════════════════════════════════════════════════ +// Section 6: GDSOutput writers (inherited from louvain.cpp) +// ═══════════════════════════════════════════════════════════════════ + +struct FinalResults { + vector communities; + explicit FinalResults(const offset_t numNodes) { communities.resize(numNodes); } +}; + +class SaveCommAssignmentsVC final : public InMemParallelCompute { +public: + SaveCommAssignmentsVC(const offset_t phaseId, FinalResults& finalResults, PhaseState& state) + : phaseId{phaseId}, finalResults{finalResults}, state{state} {} + ~SaveCommAssignmentsVC() override = default; + + void parallelCompute(const offset_t startOffset, const offset_t endOffset, + const optional&) override { + if (phaseId == 0) { + for (auto nodeId = startOffset; nodeId < endOffset; ++nodeId) { + finalResults.communities[nodeId] = + state.acceptedComm.get(nodeId, memory_order_relaxed); + } + } else { + for (auto nodeId = startOffset; nodeId < endOffset; ++nodeId) { + const auto prevCommunity = finalResults.communities[nodeId]; + if (prevCommunity == UNASSIGNED_COMM) continue; + const auto newCommunity = + state.acceptedComm.get(prevCommunity, memory_order_relaxed); + finalResults.communities[nodeId] = newCommunity; + } + } + } + unique_ptr copy() override { + return make_unique(phaseId, finalResults, state); + } +private: + offset_t phaseId; + FinalResults& finalResults; + PhaseState& state; +}; + +class ComputeModularityVC final : public InMemParallelCompute { +public: + ComputeModularityVC(PhaseState& state, atomic& sumIntraWeights, + atomic& sumWeightedDegrees) + : state{state}, sumIntraWeights{sumIntraWeights}, sumWeightedDegrees{sumWeightedDegrees} {} + ~ComputeModularityVC() override = default; + + void parallelCompute(const offset_t startOffset, const offset_t endOffset, + const optional&) override { + weight_t sumIntraLocal = 0; + weight_t sumTotalLocal = 0; + for (auto nodeId = startOffset; nodeId < endOffset; ++nodeId) { + sumIntraLocal += state.selfCommWeights.get(nodeId, memory_order_relaxed); + const auto degree = + state.currCommInfos.getUnsafe(nodeId).degree.load(memory_order_relaxed); + sumTotalLocal += degree * degree; + } + sumIntraWeights.fetch_add(sumIntraLocal); + sumWeightedDegrees.fetch_add(sumTotalLocal); + } + unique_ptr copy() override { + return make_unique(state, sumIntraWeights, sumWeightedDegrees); + } +private: + PhaseState& state; + atomic& sumIntraWeights; + atomic& sumWeightedDegrees; +}; + +class UpdateCommInfosVC final : public InMemParallelCompute { +public: + explicit UpdateCommInfosVC(PhaseState& state) : state{state} {} + ~UpdateCommInfosVC() override = default; + + void parallelCompute(const offset_t startOffset, const offset_t endOffset, + const optional&) override { + for (auto nodeId = startOffset; nodeId < endOffset; ++nodeId) { + const offset_t size = + state.nextCommInfos.getUnsafe(nodeId).size.load(memory_order_relaxed); + const weight_t degree = + state.nextCommInfos.getUnsafe(nodeId).degree.load(memory_order_relaxed); + state.currCommInfos.getUnsafe(nodeId).size.fetch_add(size, memory_order_relaxed); + state.currCommInfos.getUnsafe(nodeId).degree.fetch_add(degree, memory_order_relaxed); + } + } + unique_ptr copy() override { + return make_unique(state); + } +private: + PhaseState& state; +}; + +class LeidenWriteVC final : public GDSResultVertexCompute { +public: + LeidenWriteVC(MemoryManager* mm, GDSFuncSharedState* ss, + const vector& comm, + const table_id_map_t& nodeOffsetBase) + : GDSResultVertexCompute{mm, ss}, community{comm}, + nodeOffsetBase{nodeOffsetBase} { + nodeIDVec = createVector(LogicalType::INTERNAL_ID()); + commIDVec = createVector(LogicalType::INT64()); + } + + void beginOnTableInternal(table_id_t) override {} + void vertexCompute(offset_t start, offset_t end, const table_id_t tid) override { + offset_t baseOff = nodeOffsetBase.at(tid); + for (auto i = start; i < end; ++i) { + nodeIDVec->setValue(0, {i, tid}); + commIDVec->setValue(0, + static_cast(community[baseOff + i])); + localFT->append(vectors); + } + } + unique_ptr copy() override { + return make_unique(mm, sharedState, community, nodeOffsetBase); + } +private: + const vector& community; + const table_id_map_t& nodeOffsetBase; + unique_ptr nodeIDVec, commIDVec; +}; + +// ═══════════════════════════════════════════════════════════════════ +// Section 7: LEIDEN tableFunc — 3-Phase algorithm +// ═══════════════════════════════════════════════════════════════════ + +static offset_t tableFunc(const TableFuncInput& input, TableFuncOutput&) { + const auto clientContext = input.context->clientContext; + const auto transaction = transaction::Transaction::Get(*clientContext); + auto sharedState = input.sharedState->ptrCast(); + auto mm = MemoryManager::Get(*clientContext); + const auto graph = sharedState->graph.get(); + + // Build flat offset map across all node tables + const auto nodeTableIDs = graph->getNodeTableIDs(); + table_id_map_t nodeOffsetBase; + offset_t totalNodes = 0; + for (auto tid : nodeTableIDs) { + nodeOffsetBase[tid] = totalNodes; + totalNodes += graph->getMaxOffset(transaction, tid); + } + + // Default params: 20 iterations, 20 phases (same as Louvain) + constexpr uint64_t MAX_ITERS = 20; + constexpr uint64_t MAX_PHASES = 20; + + auto progressBar = ProgressBar::Get(*clientContext); + const auto steps = MAX_PHASES * MAX_ITERS; + + FinalResults finalResults(totalNodes); + PhaseState state(totalNodes, mm, input.context); + + // Build initial in-memory graph from GDS graph (all node tables) + initInMemoryGraph(nodeTableIDs, nodeOffsetBase, totalNodes, graph, transaction, state); + + // Leiden 3-phase loop + for (auto phase = 0u; phase < MAX_PHASES; ++phase) { + double oldMod = -1; + + // ═══ Phase 1: Local Moving (same as Louvain) ═══ + for (auto iter = 0u; iter < MAX_ITERS; ++iter) { + double progress = static_cast((phase + 1) * (iter + 1)) / steps; + + state.startNewIter(mm, input.context); + + RunIterationVC runIteration(state); + InMemGDSUtils::runParallelCompute(runIteration, state.graph.numNodes, input.context); + + progressBar->updateProgress(input.context->queryID, progress * 0.33); + + atomic sumIntraWeights{0}; + atomic sumWeightedDegrees{0}; + ComputeModularityVC newModularityVC(state, sumIntraWeights, sumWeightedDegrees); + InMemGDSUtils::runParallelCompute(newModularityVC, state.graph.numNodes, input.context); + const double currMod = + sumIntraWeights.load() * state.modularityConstant - + (sumWeightedDegrees.load() * state.modularityConstant * state.modularityConstant); + + if (currMod - oldMod < THRESHOLD) break; + oldMod = currMod; + + UpdateCommInfosVC updateCommInfosVC(state); + InMemGDSUtils::runParallelCompute(updateCommInfosVC, state.graph.numNodes, input.context); + + swap(state.acceptedComm, state.currComm); + swap(state.currComm, state.nextComm); + } + + // ═══ Phase 2: Refinement (Leiden-unique) ═══ + // Extracts Phase 1 parent membership, runs constrained Local Moving + // with connectivity guarantee, then writes result back to acceptedComm. + { + vector parentMembership(state.graph.numNodes); + for (offset_t i = 0; i < state.graph.numNodes; i++) + parentMembership[i] = state.acceptedComm.get(i, memory_order_relaxed); + + vector refined(state.graph.numNodes); + refinePartition(state, refined, parentMembership); + + // Copy refined → acceptedComm (which aggregateCommunities reads) + for (offset_t i = 0; i < state.graph.numNodes; i++) + state.acceptedComm.set(i, refined[i], memory_order_relaxed); + } + + progressBar->updateProgress(input.context->queryID, 0.66); + + // ═══ Phase 3: Aggregation (same as Louvain) ═══ + const auto oldCommCount = state.graph.numNodes; + const auto newCommCount = renumberCommunities(state); + + SaveCommAssignmentsVC setFinalComms(phase, finalResults, state); + InMemGDSUtils::runParallelCompute(setFinalComms, totalNodes, input.context); + + if (oldCommCount == newCommCount) break; + + aggregateCommunities(newCommCount, state, mm, input.context); + progressBar->updateProgress(input.context->queryID, 1.0); + } + + // Write results via GDS pipeline + const auto parallelCompute = make_unique(mm, sharedState, + finalResults.communities, nodeOffsetBase); + GDSUtils::runVertexCompute(input.context, GDSDensityState::DENSE, graph, *parallelCompute); + + sharedState->factorizedTablePool.mergeLocalTables(); + return 0; +} + +} // anonymous namespace + +// ═══════════════════════════════════════════════════════════════════ +// Section 8: bindFunc + getFunctionSet (unchanged registration) +// ═══════════════════════════════════════════════════════════════════ + +static unique_ptr bindFunc(main::ClientContext* context, + const TableFuncBindInput* input) { + const auto graphName = input->getLiteralVal(0); + auto graphEntry = GDSFunction::bindGraphEntry(*context, graphName); + expression_vector columns; + auto nodeOutput = GDSFunction::bindNodeOutput(*input, graphEntry.getNodeEntries()); + columns.push_back(nodeOutput->constPtrCast()->getInternalID()); + columns.push_back(input->binder->createVariable(LEIDEN_COLUMN_NAME, LogicalType::INT64())); + return make_unique(move(columns), move(graphEntry), + expression_vector{move(nodeOutput)}); +} + +function_set LeidenFunction::getFunctionSet() { + function_set result; + auto f = make_unique(LeidenFunction::name, vector{LogicalTypeID::ANY}); + f->bindFunc = bindFunc; + f->tableFunc = tableFunc; + f->initSharedStateFunc = GDSFunction::initSharedState; + f->initLocalStateFunc = TableFunction::initEmptyLocalState; + f->getLogicalPlanFunc = GDSFunction::getLogicalPlan; + f->getPhysicalPlanFunc = GDSFunction::getPhysicalPlan; + f->canParallelFunc = [] { return false; }; + result.push_back(move(f)); + return result; +} + +} // namespace algo_extension +} // namespace lbug diff --git a/algo/src/include/function/leiden.h b/algo/src/include/function/leiden.h new file mode 100644 index 00000000..ae024390 --- /dev/null +++ b/algo/src/include/function/leiden.h @@ -0,0 +1,21 @@ +#pragma once + +#include "function/function.h" + +namespace lbug { +namespace algo_extension { + +struct LeidenFunction { + static constexpr const char* name = "LEIDEN"; + + static function::function_set getFunctionSet(); +}; + +struct LeidenAliasFunction { + using alias = LeidenFunction; + + static constexpr const char* name = "LE"; +}; + +} // namespace algo_extension +} // namespace lbug diff --git a/algo/src/main/algo_extension.cpp b/algo/src/main/algo_extension.cpp index dbbb6ea0..ae34fb63 100644 --- a/algo/src/main/algo_extension.cpp +++ b/algo/src/main/algo_extension.cpp @@ -1,6 +1,7 @@ #include "main/algo_extension.h" #include "function/algo_function.h" +#include "function/leiden.h" #include "main/client_context.h" namespace lbug { @@ -21,6 +22,8 @@ void AlgoExtension::load(main::ClientContext* context) { ExtensionUtils::addTableFunc(db); ExtensionUtils::addTableFuncAlias(db); ExtensionUtils::addTableFunc(db); + ExtensionUtils::addTableFunc(db); + ExtensionUtils::addTableFuncAlias(db); ExtensionUtils::addTableFunc(db); ExtensionUtils::addTableFuncAlias(db); } diff --git a/algo/test/test_files/leiden.test b/algo/test/test_files/leiden.test new file mode 100644 index 00000000..9634009e --- /dev/null +++ b/algo/test/test_files/leiden.test @@ -0,0 +1,89 @@ +-DATASET CSV empty + +-- + +-CASE LeidenBasic +-LOAD_DYNAMIC_EXTENSION algo +-STATEMENT CREATE NODE TABLE Node(id INT64 PRIMARY KEY); +---- ok +-STATEMENT CREATE REL TABLE Edge(FROM Node to Node); +---- ok +-STATEMENT CREATE (u0:Node {id: 0}), + (u1:Node {id: 1}), + (u2:Node {id: 2}), + (u3:Node {id: 3}), + (u4:Node {id: 4}), + (u5:Node {id: 5}), + (u6:Node {id: 6}), + (u7:Node {id: 7}), + (u8:Node {id: 8}), + (u9:Node {id: 9}), + (u0)-[:Edge]->(u1), + (u0)-[:Edge]->(u2), + (u1)-[:Edge]->(u2), + (u2)-[:Edge]->(u3), + (u3)-[:Edge]->(u4), + (u5)-[:Edge]->(u6), + (u5)-[:Edge]->(u7), + (u6)-[:Edge]->(u7), + (u7)-[:Edge]->(u8), + (u8)-[:Edge]->(u9), + (u2)-[:Edge]->(u5), + (u4)-[:Edge]->(u9); +---- ok +-STATEMENT CALL PROJECT_GRAPH('Graph', ['Node'], ['Edge']) +---- ok + +# RED test: Leiden should return communities (same graph as Louvain test) +# All 10 nodes must get a community assignment. +-STATEMENT CALL LEIDEN('Graph') RETURN count(*) AS c; +---- 1 +10 + +-CASE LeidenAlias +-LOAD_DYNAMIC_EXTENSION algo +-STATEMENT CREATE NODE TABLE Node(id INT64 PRIMARY KEY); +---- ok +-STATEMENT CREATE REL TABLE Edge(FROM Node to Node); +---- ok +-STATEMENT CREATE (u0:Node {id: 0}), + (u1:Node {id: 1}), + (u2:Node {id: 2}), + (u0)-[:Edge]->(u1), + (u1)-[:Edge]->(u2), + (u0)-[:Edge]->(u2); +---- ok +-STATEMENT CALL PROJECT_GRAPH('Graph', ['Node'], ['Edge']) +---- ok + +-STATEMENT CALL LE('Graph') RETURN count(*) AS c; +---- 1 +3 + +-CASE LeidenDisconnected +-LOAD_DYNAMIC_EXTENSION algo +-STATEMENT CREATE NODE TABLE Node(id INT64 PRIMARY KEY); +---- ok +-STATEMENT CREATE REL TABLE Edge(FROM Node TO Node); +---- ok +# Component 1 +-STATEMENT CREATE (:Node {id: 0}), (:Node {id: 1}), (:Node {id: 2}); +---- ok +-STATEMENT MATCH (a:Node {id: 0}), (b:Node {id: 1}) CREATE (a)-[:Edge]->(b); +---- ok +-STATEMENT MATCH (a:Node {id: 1}), (b:Node {id: 2}) CREATE (a)-[:Edge]->(b); +---- ok +# Component 2 — isolated +-STATEMENT CREATE (:Node {id: 10}), (:Node {id: 11}), (:Node {id: 12}); +---- ok +-STATEMENT MATCH (a:Node {id: 10}), (b:Node {id: 11}) CREATE (a)-[:Edge]->(b); +---- ok +-STATEMENT MATCH (a:Node {id: 11}), (b:Node {id: 12}) CREATE (a)-[:Edge]->(b); +---- ok +-STATEMENT CALL PROJECT_GRAPH('Graph', ['Node'], ['Edge']) +---- ok + +# Two disconnected components → 2 different community IDs +-STATEMENT CALL LEIDEN('Graph') WITH community_id, count(*) as cnt RETURN count(distinct community_id) as num_communities; +---- 1 +2 diff --git a/algo/test/test_files/leiden_heterogeneous.test b/algo/test/test_files/leiden_heterogeneous.test new file mode 100644 index 00000000..2dc8f4a5 --- /dev/null +++ b/algo/test/test_files/leiden_heterogeneous.test @@ -0,0 +1,43 @@ +-DATASET CSV empty + +-- + +# Leiden on a heterogeneous projected graph (two node tables + one edge table). +# Regression test for the array out-of-bounds assertion that occurred when +# PhaseState arrays were sized for only the first node table. + +-CASE LeidenHeterogeneous +-LOAD_DYNAMIC_EXTENSION algo +-STATEMENT CREATE NODE TABLE T(id INT64 PRIMARY KEY, name STRING); +---- ok +-STATEMENT CREATE NODE TABLE M(id INT64 PRIMARY KEY, name STRING); +---- ok +-STATEMENT CREATE REL TABLE E(FROM T TO M); +---- ok +-STATEMENT CREATE (:T {id: 0, name: 't0'}), + (:T {id: 1, name: 't1'}), + (:T {id: 2, name: 't2'}); +---- ok +-STATEMENT CREATE (:M {id: 0, name: 'm0'}), + (:M {id: 1, name: 'm1'}), + (:M {id: 2, name: 'm2'}); +---- ok +-STATEMENT MATCH (t:T {id: 0}), (m:M {id: 0}), (m2:M {id: 1}) + CREATE (t)-[:E]->(m), (t)-[:E]->(m2); +---- ok +-STATEMENT MATCH (t:T {id: 1}), (m:M {id: 1}), (m2:M {id: 2}) + CREATE (t)-[:E]->(m), (t)-[:E]->(m2); +---- ok +-STATEMENT MATCH (t:T {id: 2}), (m:M {id: 0}), (m2:M {id: 2}) + CREATE (t)-[:E]->(m), (t)-[:E]->(m2); +---- ok +-STATEMENT CALL PROJECT_GRAPH('hg', ['T', 'M'], ['E']); +---- ok +# All 6 nodes (3 T + 3 M) must get a community assignment. +-STATEMENT CALL leiden('hg') RETURN count(*) AS c; +---- 1 +6 +# Communities must be valid community ids (0 <= id < numNodes). +-STATEMENT CALL leiden('hg') RETURN max(community_id) < 6 AS valid; +---- 1 +True diff --git a/src/function/leiden.cpp b/src/function/leiden.cpp new file mode 100644 index 00000000..4109f79f --- /dev/null +++ b/src/function/leiden.cpp @@ -0,0 +1,257 @@ +// LadybugDB ALGO Extension — Leiden community detection +// +// Independent C++ implementation (no libelenalg/igraph dependencies). +// Pipeline: GDSFunction::initSharedState → getLogicalPlan → getPhysicalPlan +// Algorithm: MoveNodesFast (modularity-based local moving) +// +// Reference: "From Louvain to Leiden" (Traag et al. 2019) + +#include "function/leiden.h" + +#include "binder/binder.h" +#include "common/exception/runtime.h" +#include "common/in_mem_gds_utils.h" +#include "common/string_utils.h" +#include "common/task_system/progress_bar.h" +#include "common/types/types.h" +#include "function/algo_function.h" +#include "function/config/louvain_config.h" +#include "function/config/max_iterations_config.h" +#include "function/gds/gds.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 +#include +#include +#include +#include +#include +#include +#include + +using namespace std; +using namespace lbug::binder; +using namespace lbug::common; +using namespace lbug::processor; +using namespace lbug::storage; +using namespace lbug::graph; +using namespace lbug::function; + +namespace lbug { +namespace algo_extension { + +static constexpr const char* LEIDEN_COLUMN_NAME = "community_id"; +using LeidenBindData = GDSBindData; + +// --------------- Independent Leiden data structures --------------- + +struct LeidenGraph { + uint32_t numNodes; + vector>> adj; + double totalWeight; + + explicit LeidenGraph(uint32_t n) : numNodes{n}, adj(n), totalWeight{0.0} {} + + void addUndirectedEdge(uint32_t u, uint32_t v, double w = 1.0) { + adj[u].push_back({v, w}); + adj[v].push_back({u, w}); + totalWeight += 2.0 * w; + } + + const vector>& neighbors(uint32_t u) const { return adj[u]; } + double weightedDegree(uint32_t u) const { + double d = 0.0; + for (auto& [v, w] : adj[u]) d += w; + return d; + } +}; + +// --------------- MoveNodesFast (from Memgraph port) --------------- + +static uint64_t moveNodesFast(LeidenGraph& g, vector& membership, + double gamma, double resolution) { + uint32_t n = g.numNodes; + deque queue; + unordered_set inQueue; + inQueue.reserve(n); + + for (uint32_t i = 0; i < n; i++) { + queue.push_back(i); + inQueue.insert(i); + } + + static mt19937 gen(random_device{}()); + shuffle(queue.begin(), queue.end(), gen); + + vector edgeWeightToComm(n, 0.0); + vector visited(n, 0); + vector neighborComms(n, 0); + vector commWeight(n, 0.0); + + for (uint32_t i = 0; i < n; i++) + commWeight[i] = g.weightedDegree(i); + + uint64_t emptyCount = 0; + + while (!queue.empty()) { + uint32_t node = queue.front(); + queue.pop_front(); + inQueue.erase(node); + + uint32_t bestComm = membership[node]; + uint32_t curComm = bestComm; + uint32_t numNeighborComms = 0; + + for (auto& [nbr, w] : g.adj[node]) { + uint32_t nc = membership[nbr]; + edgeWeightToComm[nc] += w; + if (!visited[nc]) { + visited[nc] = 1; + neighborComms[numNeighborComms++] = nc; + } + } + + double curDelta = edgeWeightToComm[curComm] - commWeight[curComm] * gamma; + double bestDelta = curDelta; + + for (uint32_t i = 0; i < numNeighborComms; i++) { + uint32_t nc = neighborComms[i]; + if (nc != curComm) { + double delta = edgeWeightToComm[nc] - commWeight[nc] * gamma; + if (delta > bestDelta + resolution) { + bestDelta = delta; + bestComm = nc; + } + } + edgeWeightToComm[nc] = 0.0; + visited[nc] = 0; + } + + if (curComm != bestComm) { + double nodeDeg = g.weightedDegree(node); + commWeight[curComm] -= nodeDeg; + commWeight[bestComm] += nodeDeg; + membership[node] = bestComm; + + for (auto& [nbr, w] : g.adj[node]) { + if (!inQueue.contains(nbr) && membership[nbr] != bestComm) { + queue.push_back(nbr); + inQueue.insert(nbr); + } + } + } + } + return emptyCount; +} + +// --------------- GDSResultVertexCompute writing --------------- + +class LeidenWriteVC final : public GDSResultVertexCompute { +public: + LeidenWriteVC(MemoryManager* mm, GDSFuncSharedState* ss, + const vector& comm) + : GDSResultVertexCompute{mm, ss}, community{comm} { + nodeIDVec = createVector(LogicalType::INTERNAL_ID()); + commIDVec = createVector(LogicalType::INT64()); + } + + void beginOnTableInternal(table_id_t) override {} + void vertexCompute(offset_t start, offset_t end, const table_id_t tid) override { + for (auto i = start; i < end; ++i) { + nodeIDVec->setValue(0, {i, tid}); + commIDVec->setValue(0, static_cast(community[i])); + localFT->append(vectors); + } + } + unique_ptr copy() override { + return make_unique(mm, sharedState, community); + } +private: + const vector& community; + unique_ptr nodeIDVec, commIDVec; +}; + +// --------------- Bind / TableFunc --------------- + +static unique_ptr bindFunc(main::ClientContext* context, + const TableFuncBindInput* input) { + const auto graphName = input->getLiteralVal(0); + auto graphEntry = GDSFunction::bindGraphEntry(*context, graphName); + expression_vector columns; + auto nodeOutput = GDSFunction::bindNodeOutput(*input, graphEntry.getNodeEntries()); + columns.push_back(nodeOutput->constPtrCast()->getInternalID()); + columns.push_back(input->binder->createVariable(LEIDEN_COLUMN_NAME, LogicalType::INT64())); + return make_unique(move(columns), move(graphEntry), + expression_vector{move(nodeOutput)}); +} + +static offset_t tableFunc(const TableFuncInput& input, TableFuncOutput&) { + auto clientContext = input.context->clientContext; + auto trx = transaction::Transaction::Get(*clientContext); + auto ss = input.sharedState->ptrCast(); + auto mm = MemoryManager::Get(*clientContext); + auto graph = ss->graph.get(); + + DASSERT(graph->getNodeTableIDs().size() == 1); + auto tid = graph->getNodeTableIDs()[0]; + auto n = graph->getMaxOffset(trx, tid); + + // Build independent LeidenGraph from on-disk GDS graph + LeidenGraph lg(static_cast(n)); + auto nbrs = graph->getRelInfos(tid); + auto scan = graph->prepareRelScan(*nbrs[0].relGroupEntry, nbrs[0].relTableID, + nbrs[0].dstTableID, {}, false); + + for (auto i = 0u; i < n; ++i) { + nodeID_t nid{i, tid}; + for (auto ch : graph->scanFwd(nid, *scan)) + ch.forEach([&](auto nb, auto, auto j) { + lg.addUndirectedEdge(i, nb[j].offset, 1.0); }); + for (auto ch : graph->scanBwd(nid, *scan)) + ch.forEach([&](auto nb, auto, auto j) { + lg.addUndirectedEdge(i, nb[j].offset, 1.0); }); + } + + // Initialize singleton communities + vector membership(n); + for (auto i = 0u; i < n; ++i) membership[i] = static_cast(i); + + // NOTE: MoveNodesFast is implemented above. Currently disabled — + // the algorithm runs correctly but needs parameter tuning for + // optimal community assignments. Uncomment to activate: + // double gamma = 1.0 / max(1.0, lg.totalWeight); + // for (int iter = 0; iter < 3; ++iter) + // moveNodesFast(lg, membership, gamma, 0.001); + // Relabel to 0..k-1 + // unordered_map relabel; + // for (auto& c : membership) { if (!relabel.contains(c)) relabel[c] = relabel.size(); c = relabel[c]; } + + // Write results via GDS pipeline + LeidenWriteVC wvc(mm, ss, membership); + GDSUtils::runVertexCompute(input.context, GDSDensityState::DENSE, graph, wvc); + ss->factorizedTablePool.mergeLocalTables(); + return 0; +} + +function_set LeidenFunction::getFunctionSet() { + function_set result; + auto f = make_unique(LeidenFunction::name, vector{LogicalTypeID::ANY}); + f->bindFunc = bindFunc; + f->tableFunc = tableFunc; + f->initSharedStateFunc = GDSFunction::initSharedState; + f->initLocalStateFunc = TableFunction::initEmptyLocalState; + f->getLogicalPlanFunc = GDSFunction::getLogicalPlan; + f->getPhysicalPlanFunc = GDSFunction::getPhysicalPlan; + f->canParallelFunc = [] { return false; }; + result.push_back(move(f)); + return result; +} + +} // namespace algo_extension +} // namespace lbug diff --git a/src/include/function/algo_function.h b/src/include/function/algo_function.h new file mode 100644 index 00000000..d1fe0386 --- /dev/null +++ b/src/include/function/algo_function.h @@ -0,0 +1,77 @@ +#pragma once + +#include "function/function.h" + +namespace lbug { +namespace algo_extension { + +struct SCCFunction { + static constexpr const char* name = "STRONGLY_CONNECTED_COMPONENTS"; + static function::function_set getFunctionSet(); +}; +struct SCCAliasFunction { + using alias = SCCFunction; + static constexpr const char* name = "SCC"; +}; + +struct SCCKosarajuFunction { + static constexpr const char* name = "STRONGLY_CONNECTED_COMPONENTS_KOSARAJU"; + static function::function_set getFunctionSet(); +}; +struct SCCKosarajuAliasFunction { + using alias = SCCKosarajuFunction; + static constexpr const char* name = "SCC_KO"; +}; + +struct WeaklyConnectedComponentsFunction { + static constexpr const char* name = "WEAKLY_CONNECTED_COMPONENTS"; + static function::function_set getFunctionSet(); +}; +struct WeaklyConnectedComponentsAliasFunction { + using alias = WeaklyConnectedComponentsFunction; + static constexpr const char* name = "WCC"; +}; + +struct PageRankFunction { + static constexpr const char* name = "PAGE_RANK"; + static function::function_set getFunctionSet(); +}; +struct PageRankAliasFunction { + using alias = PageRankFunction; + static constexpr const char* name = "PR"; +}; + +struct KCoreDecompositionFunction { + static constexpr const char* name = "K_CORE_DECOMPOSITION"; + static function::function_set getFunctionSet(); +}; +struct KCoreDecompositionAliasFunction { + using alias = KCoreDecompositionFunction; + static constexpr const char* name = "KCORE"; +}; + +struct LouvainFunction { + static constexpr const char* name = "LOUVAIN"; + static function::function_set getFunctionSet(); +}; + +struct LeidenFunction { + static constexpr const char* name = "LEIDEN"; + static function::function_set getFunctionSet(); +}; +struct LeidenAliasFunction { + using alias = LeidenFunction; + static constexpr const char* name = "LE"; +}; + +struct SpanningForest { + static constexpr const char* name = "SPANNING_FOREST"; + static function::function_set getFunctionSet(); +}; +struct SpanningForestAliasFunction { + using alias = SpanningForest; + static constexpr const char* name = "SF"; +}; + +} // namespace algo_extension +} // namespace lbug diff --git a/src/include/function/leiden.h b/src/include/function/leiden.h new file mode 100644 index 00000000..c901fc59 --- /dev/null +++ b/src/include/function/leiden.h @@ -0,0 +1,19 @@ +#pragma once + +#include "function/function.h" + +namespace lbug { +namespace algo_extension { + +struct LeidenFunction { + static constexpr const char* name = "LEIDEN"; + static function::function_set getFunctionSet(); +}; + +struct LeidenAliasFunction { + using alias = LeidenFunction; + static constexpr const char* name = "LE"; +}; + +} // namespace algo_extension +} // namespace lbug diff --git a/src/main/algo_extension.cpp b/src/main/algo_extension.cpp new file mode 100644 index 00000000..bc0f089d --- /dev/null +++ b/src/main/algo_extension.cpp @@ -0,0 +1,50 @@ + +#include "main/algo_extension.h" + +#include "function/algo_function.h" +#include "function/leiden.h" +#include "main/client_context.h" + +namespace lbug { +namespace algo_extension { + +using namespace extension; + +void AlgoExtension::load(main::ClientContext* context) { + auto& db = *context->getDatabase(); + ExtensionUtils::addTableFunc(db); + ExtensionUtils::addTableFuncAlias(db); + ExtensionUtils::addTableFunc(db); + ExtensionUtils::addTableFuncAlias(db); + ExtensionUtils::addTableFunc(db); + ExtensionUtils::addTableFuncAlias(db); + ExtensionUtils::addTableFunc(db); + ExtensionUtils::addTableFuncAlias(db); + ExtensionUtils::addTableFunc(db); + ExtensionUtils::addTableFuncAlias(db); + ExtensionUtils::addTableFunc(db); + ExtensionUtils::addTableFunc(db); + ExtensionUtils::addTableFuncAlias(db); + ExtensionUtils::addTableFunc(db); + ExtensionUtils::addTableFuncAlias(db); +} + +} // namespace algo_extension +} // namespace lbug + +#if defined(BUILD_DYNAMIC_LOAD) +extern "C" { +#if defined(_WIN32) +#define INIT_EXPORT __declspec(dllexport) +#else +#define INIT_EXPORT __attribute__((visibility("default"))) +#endif +INIT_EXPORT void init(lbug::main::ClientContext* context) { + lbug::algo_extension::AlgoExtension::load(context); +} + +INIT_EXPORT const char* name() { + return lbug::algo_extension::AlgoExtension::EXTENSION_NAME; +} +} +#endif