nodedb 0.0.0-beta.1

Local-first, real-time, edge-to-cloud hybrid database for multi-modal workloads
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

//! Graph algorithm dispatch handler.
//!
//! Routes `PhysicalPlan::GraphAlgo` to the appropriate algorithm
//! implementation in `engine::graph::algo::*`. Each algorithm runs
//! on the in-memory CSR index and returns JSON-serialized results.

use tracing::debug;

use crate::bridge::envelope::{ErrorCode, Response};
use crate::data::executor::core_loop::CoreLoop;
use crate::data::executor::task::ExecutionTask;
use crate::engine::graph::algo::params::{AlgoParams, GraphAlgorithm};

impl CoreLoop {
    pub(in crate::data::executor) fn execute_graph_algo(
        &self,
        task: &ExecutionTask,
        algorithm: &GraphAlgorithm,
        params: &AlgoParams,
    ) -> Response {
        debug!(
            core = self.core_id,
            algorithm = algorithm.name(),
            collection = %params.collection,
            "graph algorithm dispatch"
        );

        // Validate source_node for SSSP.
        if *algorithm == GraphAlgorithm::Sssp && params.source_node.is_none() {
            return self.response_error(
                task,
                ErrorCode::Internal {
                    detail: "SSSP requires FROM '<source_node>'".into(),
                },
            );
        }

        use crate::engine::graph::algo;

        let result: Result<Vec<u8>, crate::Error> = match algorithm {
            // ── Graphalytics (5 core algorithms) ──
            GraphAlgorithm::PageRank => algo::pagerank::run(&self.csr, params).to_json(),
            GraphAlgorithm::Wcc => algo::wcc::run(&self.csr).to_json(),
            GraphAlgorithm::LabelPropagation => {
                algo::label_propagation::run(&self.csr, params).to_json()
            }
            GraphAlgorithm::Lcc => algo::lcc::run(
                &self.csr,
                self.graph_tuning.lcc_high_degree_threshold,
                self.graph_tuning.lcc_sample_pairs,
            )
            .to_json(),
            GraphAlgorithm::Sssp => {
                algo::sssp::run(&self.csr, params).and_then(|batch| batch.to_json())
            }
            // ── Centrality suite ──
            GraphAlgorithm::Betweenness => algo::betweenness::run(&self.csr, params).to_json(),
            GraphAlgorithm::Closeness => algo::closeness::run(&self.csr).to_json(),
            GraphAlgorithm::Harmonic => algo::harmonic::run(&self.csr).to_json(),
            GraphAlgorithm::Degree => algo::degree::run(&self.csr, params).to_json(),
            // ── Extended algorithms ──
            GraphAlgorithm::Louvain => algo::louvain::run(&self.csr, params).to_json(),
            GraphAlgorithm::Triangles => algo::triangles::run(&self.csr, params).to_json(),
            GraphAlgorithm::Diameter => algo::diameter::run(&self.csr, params).to_json(),
            GraphAlgorithm::KCore => algo::kcore::run(&self.csr).to_json(),
        };

        match result {
            Ok(payload) => self.response_with_payload(task, payload),
            Err(e) => self.response_error(task, ErrorCode::from(e)),
        }
    }
}