reddb_server/runtime/
impl_graph.rs

1use super::*;
2
3impl RedDBRuntime {
4    pub fn graph_neighborhood(
5        &self,
6        node: &str,
7        direction: RuntimeGraphDirection,
8        max_depth: usize,
9        edge_labels: Option<Vec<String>>,
10        projection: Option<RuntimeGraphProjection>,
11    ) -> RedDBResult<RuntimeGraphNeighborhoodResult> {
12        let graph =
13            materialize_graph_with_projection(self.inner.db.store().as_ref(), projection.as_ref())?;
14        let node = resolve_graph_node_id(&graph, node)?;
15        let edge_filters = merge_edge_filters(edge_labels, projection.as_ref());
16
17        let mut visited: HashMap<String, usize> = HashMap::new();
18        let mut queue = VecDeque::new();
19        let mut nodes = Vec::new();
20        let mut edges = Vec::new();
21        let mut seen_edges = HashSet::new();
22
23        visited.insert(node.clone(), 0);
24        queue.push_back((node.clone(), 0usize));
25
26        while let Some((current, depth)) = queue.pop_front() {
27            if let Some(stored) = graph.get_node(&current) {
28                nodes.push(RuntimeGraphVisit {
29                    depth,
30                    node: stored_node_to_runtime(stored),
31                });
32            }
33
34            if depth >= max_depth {
35                continue;
36            }
37
38            let mut adjacent =
39                graph_adjacent_edges(&graph, &current, direction, edge_filters.as_ref());
40            adjacent.sort_by(|left, right| left.0.cmp(&right.0));
41
42            for (neighbor, edge) in adjacent {
43                push_runtime_edge(&mut edges, &mut seen_edges, edge);
44                // Single hash lookup: enqueue only on first visit.
45                if let std::collections::hash_map::Entry::Vacant(slot) = visited.entry(neighbor) {
46                    let next_depth = depth + 1;
47                    queue.push_back((slot.key().clone(), next_depth));
48                    slot.insert(next_depth);
49                }
50            }
51        }
52
53        Ok(RuntimeGraphNeighborhoodResult {
54            source: node,
55            direction,
56            max_depth,
57            nodes,
58            edges,
59        })
60    }
61
62    pub fn graph_traverse(
63        &self,
64        source: &str,
65        direction: RuntimeGraphDirection,
66        max_depth: usize,
67        strategy: RuntimeGraphTraversalStrategy,
68        edge_labels: Option<Vec<String>>,
69        projection: Option<RuntimeGraphProjection>,
70    ) -> RedDBResult<RuntimeGraphTraversalResult> {
71        let graph =
72            materialize_graph_with_projection(self.inner.db.store().as_ref(), projection.as_ref())?;
73        let source = resolve_graph_node_id(&graph, source)?;
74        let edge_filters = merge_edge_filters(edge_labels, projection.as_ref());
75
76        let mut visits = Vec::new();
77        let mut edges = Vec::new();
78        let mut seen_nodes = HashSet::new();
79        let mut seen_edges = HashSet::new();
80
81        match strategy {
82            RuntimeGraphTraversalStrategy::Bfs => {
83                let mut queue = VecDeque::new();
84                queue.push_back((source.clone(), 0usize));
85                seen_nodes.insert(source.clone());
86
87                while let Some((current, depth)) = queue.pop_front() {
88                    if let Some(stored) = graph.get_node(&current) {
89                        visits.push(RuntimeGraphVisit {
90                            depth,
91                            node: stored_node_to_runtime(stored),
92                        });
93                    }
94
95                    if depth >= max_depth {
96                        continue;
97                    }
98
99                    let mut adjacent =
100                        graph_adjacent_edges(&graph, &current, direction, edge_filters.as_ref());
101                    adjacent.sort_by(|left, right| left.0.cmp(&right.0));
102                    for (neighbor, edge) in adjacent {
103                        push_runtime_edge(&mut edges, &mut seen_edges, edge);
104                        if seen_nodes.insert(neighbor.clone()) {
105                            queue.push_back((neighbor, depth + 1));
106                        }
107                    }
108                }
109            }
110            RuntimeGraphTraversalStrategy::Dfs => {
111                let mut stack = vec![(source.clone(), 0usize)];
112                while let Some((current, depth)) = stack.pop() {
113                    if !seen_nodes.insert(current.clone()) {
114                        continue;
115                    }
116
117                    if let Some(stored) = graph.get_node(&current) {
118                        visits.push(RuntimeGraphVisit {
119                            depth,
120                            node: stored_node_to_runtime(stored),
121                        });
122                    }
123
124                    if depth >= max_depth {
125                        continue;
126                    }
127
128                    let mut adjacent =
129                        graph_adjacent_edges(&graph, &current, direction, edge_filters.as_ref());
130                    adjacent.sort_by(|left, right| right.0.cmp(&left.0));
131                    for (neighbor, edge) in adjacent {
132                        push_runtime_edge(&mut edges, &mut seen_edges, edge);
133                        if !seen_nodes.contains(&neighbor) {
134                            stack.push((neighbor, depth + 1));
135                        }
136                    }
137                }
138            }
139        }
140
141        Ok(RuntimeGraphTraversalResult {
142            source,
143            direction,
144            strategy,
145            max_depth,
146            visits,
147            edges,
148        })
149    }
150
151    pub fn graph_shortest_path(
152        &self,
153        source: &str,
154        target: &str,
155        direction: RuntimeGraphDirection,
156        algorithm: RuntimeGraphPathAlgorithm,
157        edge_labels: Option<Vec<String>>,
158        projection: Option<RuntimeGraphProjection>,
159    ) -> RedDBResult<RuntimeGraphPathResult> {
160        let graph =
161            materialize_graph_with_projection(self.inner.db.store().as_ref(), projection.as_ref())?;
162        let source_owned = resolve_graph_node_id(&graph, source)?;
163        let target_owned = resolve_graph_node_id(&graph, target)?;
164        let source = source_owned.as_str();
165        let target = target_owned.as_str();
166
167        let merged_edge_filters = merge_edge_filters(edge_labels, projection.as_ref());
168        let path = match (direction, merged_edge_filters.as_ref()) {
169            (RuntimeGraphDirection::Outgoing, None) => match algorithm {
170                RuntimeGraphPathAlgorithm::Bfs => {
171                    let result = BFS::shortest_path(&graph, source, target);
172                    RuntimeGraphPathResult {
173                        source: source.to_string(),
174                        target: target.to_string(),
175                        direction,
176                        algorithm,
177                        nodes_visited: result.nodes_visited,
178                        negative_cycle_detected: None,
179                        path: result.path.map(|path| path_to_runtime(&graph, &path)),
180                    }
181                }
182                RuntimeGraphPathAlgorithm::Dijkstra => {
183                    let result = Dijkstra::shortest_path(&graph, source, target);
184                    RuntimeGraphPathResult {
185                        source: source.to_string(),
186                        target: target.to_string(),
187                        direction,
188                        algorithm,
189                        nodes_visited: result.nodes_visited,
190                        negative_cycle_detected: None,
191                        path: result.path.map(|path| path_to_runtime(&graph, &path)),
192                    }
193                }
194                RuntimeGraphPathAlgorithm::AStar => {
195                    let result = AStar::shortest_path_no_heuristic(&graph, source, target);
196                    RuntimeGraphPathResult {
197                        source: source.to_string(),
198                        target: target.to_string(),
199                        direction,
200                        algorithm,
201                        nodes_visited: result.nodes_visited,
202                        negative_cycle_detected: None,
203                        path: result.path.map(|path| path_to_runtime(&graph, &path)),
204                    }
205                }
206                RuntimeGraphPathAlgorithm::BellmanFord => {
207                    let result = BellmanFord::shortest_path(&graph, source, target);
208                    RuntimeGraphPathResult {
209                        source: source.to_string(),
210                        target: target.to_string(),
211                        direction,
212                        algorithm,
213                        nodes_visited: result.nodes_visited,
214                        negative_cycle_detected: Some(result.has_negative_cycle),
215                        path: result.path.map(|path| path_to_runtime(&graph, &path)),
216                    }
217                }
218            },
219            _ => shortest_path_runtime(
220                &graph,
221                source,
222                target,
223                direction,
224                algorithm,
225                merged_edge_filters.as_ref(),
226            )?,
227        };
228
229        Ok(path)
230    }
231
232    pub fn graph_components(
233        &self,
234        mode: RuntimeGraphComponentsMode,
235        min_size: usize,
236        projection: Option<RuntimeGraphProjection>,
237    ) -> RedDBResult<RuntimeGraphComponentsResult> {
238        let graph =
239            materialize_graph_with_projection(self.inner.db.store().as_ref(), projection.as_ref())?;
240        let min_size = min_size.max(1);
241        let components = match mode {
242            RuntimeGraphComponentsMode::Connected => ConnectedComponents::find(&graph)
243                .components
244                .into_iter()
245                .filter(|component| component.size >= min_size)
246                .map(|component| RuntimeGraphComponent {
247                    id: component.id,
248                    size: component.size,
249                    nodes: component.nodes,
250                })
251                .collect::<Vec<_>>(),
252            RuntimeGraphComponentsMode::Weak => WeaklyConnectedComponents::find(&graph)
253                .components
254                .into_iter()
255                .filter(|component| component.len() >= min_size)
256                .enumerate()
257                .map(|(index, nodes)| RuntimeGraphComponent {
258                    id: format!("wcc:{index}"),
259                    size: nodes.len(),
260                    nodes,
261                })
262                .collect::<Vec<_>>(),
263            RuntimeGraphComponentsMode::Strong => StronglyConnectedComponents::find(&graph)
264                .components
265                .into_iter()
266                .filter(|component| component.len() >= min_size)
267                .enumerate()
268                .map(|(index, nodes)| RuntimeGraphComponent {
269                    id: format!("scc:{index}"),
270                    size: nodes.len(),
271                    nodes,
272                })
273                .collect::<Vec<_>>(),
274        };
275
276        Ok(RuntimeGraphComponentsResult {
277            mode,
278            count: components.len(),
279            components,
280        })
281    }
282
283    pub fn graph_centrality(
284        &self,
285        algorithm: RuntimeGraphCentralityAlgorithm,
286        top_k: usize,
287        normalize: bool,
288        max_iterations: Option<usize>,
289        epsilon: Option<f64>,
290        alpha: Option<f64>,
291        projection: Option<RuntimeGraphProjection>,
292    ) -> RedDBResult<RuntimeGraphCentralityResult> {
293        let graph =
294            materialize_graph_with_projection(self.inner.db.store().as_ref(), projection.as_ref())?;
295        let top_k = top_k.max(1);
296
297        match algorithm {
298            RuntimeGraphCentralityAlgorithm::Degree => {
299                let result = DegreeCentrality::compute(&graph);
300                let mut degree_scores = Vec::new();
301                let mut pairs: Vec<_> = result
302                    .total_degree
303                    .iter()
304                    .map(|(node_id, total_degree)| (node_id.clone(), *total_degree))
305                    .collect();
306                pairs
307                    .sort_by(|left, right| right.1.cmp(&left.1).then_with(|| left.0.cmp(&right.0)));
308                pairs.truncate(top_k);
309
310                for (node_id, total_degree) in pairs {
311                    if let Some(node) = graph.get_node(&node_id) {
312                        degree_scores.push(RuntimeGraphDegreeScore {
313                            node: stored_node_to_runtime(node),
314                            in_degree: result.in_degree.get(&node_id).copied().unwrap_or(0),
315                            out_degree: result.out_degree.get(&node_id).copied().unwrap_or(0),
316                            total_degree,
317                        });
318                    }
319                }
320
321                Ok(RuntimeGraphCentralityResult {
322                    algorithm,
323                    normalized: None,
324                    iterations: None,
325                    converged: None,
326                    scores: Vec::new(),
327                    degree_scores,
328                })
329            }
330            RuntimeGraphCentralityAlgorithm::Closeness => {
331                let result = ClosenessCentrality::compute(&graph);
332                Ok(RuntimeGraphCentralityResult {
333                    algorithm,
334                    normalized: None,
335                    iterations: None,
336                    converged: None,
337                    scores: top_runtime_scores(&graph, result.scores, top_k),
338                    degree_scores: Vec::new(),
339                })
340            }
341            RuntimeGraphCentralityAlgorithm::Betweenness => {
342                let result = BetweennessCentrality::compute(&graph, normalize);
343                Ok(RuntimeGraphCentralityResult {
344                    algorithm,
345                    normalized: Some(result.normalized),
346                    iterations: None,
347                    converged: None,
348                    scores: top_runtime_scores(&graph, result.scores, top_k),
349                    degree_scores: Vec::new(),
350                })
351            }
352            RuntimeGraphCentralityAlgorithm::Eigenvector => {
353                let mut runner = EigenvectorCentrality::new();
354                if let Some(max_iterations) = max_iterations {
355                    runner.max_iterations = max_iterations.max(1);
356                }
357                if let Some(epsilon) = epsilon {
358                    runner.epsilon = epsilon.max(0.0);
359                }
360                let result = runner.compute(&graph);
361                Ok(RuntimeGraphCentralityResult {
362                    algorithm,
363                    normalized: None,
364                    iterations: Some(result.iterations),
365                    converged: Some(result.converged),
366                    scores: top_runtime_scores(&graph, result.scores, top_k),
367                    degree_scores: Vec::new(),
368                })
369            }
370            RuntimeGraphCentralityAlgorithm::PageRank => {
371                let mut runner = PageRank::new();
372                if let Some(max_iterations) = max_iterations {
373                    runner = runner.max_iterations(max_iterations.max(1));
374                }
375                if let Some(alpha) = alpha {
376                    runner = runner.alpha(alpha);
377                }
378                if let Some(epsilon) = epsilon {
379                    runner = runner.epsilon(epsilon);
380                }
381                let result = runner.run(&graph);
382                Ok(RuntimeGraphCentralityResult {
383                    algorithm,
384                    normalized: None,
385                    iterations: Some(result.iterations),
386                    converged: Some(result.converged),
387                    scores: top_runtime_scores(&graph, result.scores, top_k),
388                    degree_scores: Vec::new(),
389                })
390            }
391        }
392    }
393
394    pub fn graph_communities(
395        &self,
396        algorithm: RuntimeGraphCommunityAlgorithm,
397        min_size: usize,
398        max_iterations: Option<usize>,
399        resolution: Option<f64>,
400        projection: Option<RuntimeGraphProjection>,
401    ) -> RedDBResult<RuntimeGraphCommunityResult> {
402        let graph =
403            materialize_graph_with_projection(self.inner.db.store().as_ref(), projection.as_ref())?;
404        let min_size = min_size.max(1);
405
406        match algorithm {
407            RuntimeGraphCommunityAlgorithm::LabelPropagation => {
408                let mut runner = LabelPropagation::new();
409                if let Some(max_iterations) = max_iterations {
410                    runner = runner.max_iterations(max_iterations.max(1));
411                }
412                let result = runner.run(&graph);
413                let communities = result
414                    .communities
415                    .into_iter()
416                    .filter(|community| community.size >= min_size)
417                    .map(|community| RuntimeGraphCommunity {
418                        id: community.label,
419                        size: community.size,
420                        nodes: community.nodes,
421                    })
422                    .collect::<Vec<_>>();
423                Ok(RuntimeGraphCommunityResult {
424                    algorithm,
425                    count: communities.len(),
426                    iterations: Some(result.iterations),
427                    converged: Some(result.converged),
428                    modularity: None,
429                    passes: None,
430                    communities,
431                })
432            }
433            RuntimeGraphCommunityAlgorithm::Louvain => {
434                let mut runner = Louvain::new();
435                if let Some(max_iterations) = max_iterations {
436                    runner = runner.max_iterations(max_iterations.max(1));
437                }
438                if let Some(resolution) = resolution {
439                    runner = runner.resolution(resolution.max(0.0));
440                }
441                let result = runner.run(&graph);
442                let mut communities = result
443                    .community_sizes()
444                    .into_iter()
445                    .filter(|(_, size)| *size >= min_size)
446                    .map(|(id, size)| RuntimeGraphCommunity {
447                        id: format!("community:{id}"),
448                        size,
449                        nodes: result.get_community(id),
450                    })
451                    .collect::<Vec<_>>();
452                communities.sort_by(|left, right| {
453                    right
454                        .size
455                        .cmp(&left.size)
456                        .then_with(|| left.id.cmp(&right.id))
457                });
458                Ok(RuntimeGraphCommunityResult {
459                    algorithm,
460                    count: communities.len(),
461                    iterations: None,
462                    converged: None,
463                    modularity: Some(result.modularity),
464                    passes: Some(result.passes),
465                    communities,
466                })
467            }
468        }
469    }
470
471    pub fn graph_clustering(
472        &self,
473        top_k: usize,
474        include_triangles: bool,
475        projection: Option<RuntimeGraphProjection>,
476    ) -> RedDBResult<RuntimeGraphClusteringResult> {
477        let graph =
478            materialize_graph_with_projection(self.inner.db.store().as_ref(), projection.as_ref())?;
479        let top_k = top_k.max(1);
480        let result = ClusteringCoefficient::compute(&graph);
481        let triangle_count = if include_triangles {
482            Some(crate::storage::engine::TriangleCounting::count(&graph).count)
483        } else {
484            None
485        };
486
487        Ok(RuntimeGraphClusteringResult {
488            global: result.global,
489            local: top_runtime_scores(&graph, result.local, top_k),
490            triangle_count,
491        })
492    }
493
494    pub fn graph_personalized_pagerank(
495        &self,
496        seeds: Vec<String>,
497        top_k: usize,
498        alpha: Option<f64>,
499        epsilon: Option<f64>,
500        max_iterations: Option<usize>,
501        projection: Option<RuntimeGraphProjection>,
502    ) -> RedDBResult<RuntimeGraphCentralityResult> {
503        let graph =
504            materialize_graph_with_projection(self.inner.db.store().as_ref(), projection.as_ref())?;
505        if seeds.is_empty() {
506            return Err(RedDBError::Query(
507                "personalized pagerank requires at least one seed".to_string(),
508            ));
509        }
510        for seed in &seeds {
511            ensure_graph_node(&graph, seed)?;
512        }
513
514        let mut runner = PersonalizedPageRank::new(seeds);
515        if let Some(alpha) = alpha {
516            runner = runner.alpha(alpha);
517        }
518        if let Some(epsilon) = epsilon {
519            runner = runner.epsilon(epsilon);
520        }
521        if let Some(max_iterations) = max_iterations {
522            runner = runner.max_iterations(max_iterations.max(1));
523        }
524        let result = runner.run(&graph);
525
526        Ok(RuntimeGraphCentralityResult {
527            algorithm: RuntimeGraphCentralityAlgorithm::PageRank,
528            normalized: None,
529            iterations: Some(result.iterations),
530            converged: Some(result.converged),
531            scores: top_runtime_scores(&graph, result.scores, top_k.max(1)),
532            degree_scores: Vec::new(),
533        })
534    }
535
536    pub fn graph_hits(
537        &self,
538        top_k: usize,
539        epsilon: Option<f64>,
540        max_iterations: Option<usize>,
541        projection: Option<RuntimeGraphProjection>,
542    ) -> RedDBResult<RuntimeGraphHitsResult> {
543        let graph =
544            materialize_graph_with_projection(self.inner.db.store().as_ref(), projection.as_ref())?;
545        let mut runner = HITS::new();
546        if let Some(epsilon) = epsilon {
547            runner.epsilon = epsilon.max(0.0);
548        }
549        if let Some(max_iterations) = max_iterations {
550            runner.max_iterations = max_iterations.max(1);
551        }
552        let result = runner.compute(&graph);
553
554        Ok(RuntimeGraphHitsResult {
555            iterations: result.iterations,
556            converged: result.converged,
557            hubs: top_runtime_scores(&graph, result.hub_scores, top_k.max(1)),
558            authorities: top_runtime_scores(&graph, result.authority_scores, top_k.max(1)),
559        })
560    }
561
562    pub fn graph_cycles(
563        &self,
564        max_length: usize,
565        max_cycles: usize,
566        projection: Option<RuntimeGraphProjection>,
567    ) -> RedDBResult<RuntimeGraphCyclesResult> {
568        let graph =
569            materialize_graph_with_projection(self.inner.db.store().as_ref(), projection.as_ref())?;
570        let result = CycleDetector::new()
571            .max_length(max_length.max(2))
572            .max_cycles(max_cycles.max(1))
573            .find(&graph);
574
575        Ok(RuntimeGraphCyclesResult {
576            limit_reached: result.limit_reached,
577            cycles: result
578                .cycles
579                .into_iter()
580                .map(|cycle| cycle_to_runtime(&graph, cycle))
581                .collect(),
582        })
583    }
584
585    pub fn graph_topological_sort(
586        &self,
587        projection: Option<RuntimeGraphProjection>,
588    ) -> RedDBResult<RuntimeGraphTopologicalSortResult> {
589        let graph =
590            materialize_graph_with_projection(self.inner.db.store().as_ref(), projection.as_ref())?;
591        let ordered_nodes = match DFS::topological_sort(&graph) {
592            Some(order) => order
593                .into_iter()
594                .filter_map(|id| graph.get_node(&id))
595                .map(stored_node_to_runtime)
596                .collect(),
597            None => Vec::new(),
598        };
599
600        Ok(RuntimeGraphTopologicalSortResult {
601            acyclic: !ordered_nodes.is_empty() || graph.node_count() == 0,
602            ordered_nodes,
603        })
604    }
605
606    pub fn graph_properties(
607        &self,
608        projection: Option<RuntimeGraphProjection>,
609    ) -> RedDBResult<RuntimeGraphPropertiesResult> {
610        let graph =
611            materialize_graph_with_projection(self.inner.db.store().as_ref(), projection.as_ref())?;
612        let node_count = graph.node_count() as usize;
613        let edges = graph.iter_all_edges();
614        let edge_count = edges.len();
615
616        let connected = ConnectedComponents::find(&graph);
617        let weak = WeaklyConnectedComponents::find(&graph);
618        let strong = StronglyConnectedComponents::find(&graph);
619        let cycle_result = CycleDetector::new()
620            .max_length(node_count.max(2))
621            .max_cycles(1)
622            .find(&graph);
623
624        let mut self_loop_count = 0usize;
625        let mut negative_edge_count = 0usize;
626        let mut directed_pairs = HashSet::new();
627        let mut undirected_pairs = HashSet::new();
628
629        for edge in &edges {
630            if edge.weight < 0.0 {
631                negative_edge_count += 1;
632            }
633            if edge.source_id == edge.target_id {
634                self_loop_count += 1;
635                continue;
636            }
637
638            directed_pairs.insert((edge.source_id.clone(), edge.target_id.clone()));
639            let (left, right) = if edge.source_id <= edge.target_id {
640                (edge.source_id.clone(), edge.target_id.clone())
641            } else {
642                (edge.target_id.clone(), edge.source_id.clone())
643            };
644            undirected_pairs.insert((left, right));
645        }
646
647        let expected_undirected_pairs = node_count.saturating_mul(node_count.saturating_sub(1)) / 2;
648        let expected_directed_pairs = node_count.saturating_mul(node_count.saturating_sub(1));
649        let density = if expected_undirected_pairs == 0 {
650            0.0
651        } else {
652            undirected_pairs.len() as f64 / expected_undirected_pairs as f64
653        };
654        let density_directed = if expected_directed_pairs == 0 {
655            0.0
656        } else {
657            directed_pairs.len() as f64 / expected_directed_pairs as f64
658        };
659
660        let is_empty = node_count == 0;
661        let is_connected = node_count <= 1 || connected.count == 1;
662        let is_weakly_connected = node_count <= 1 || weak.count == 1;
663        let is_strongly_connected = node_count <= 1 || strong.count == 1;
664        let is_cyclic = !cycle_result.cycles.is_empty();
665
666        Ok(RuntimeGraphPropertiesResult {
667            node_count,
668            edge_count,
669            self_loop_count,
670            negative_edge_count,
671            connected_component_count: connected.count,
672            weak_component_count: weak.count,
673            strong_component_count: strong.count,
674            is_empty,
675            is_connected,
676            is_weakly_connected,
677            is_strongly_connected,
678            is_complete: node_count <= 1 || undirected_pairs.len() == expected_undirected_pairs,
679            is_complete_directed: node_count <= 1
680                || directed_pairs.len() == expected_directed_pairs,
681            is_cyclic,
682            is_circular: is_cyclic,
683            is_acyclic: !is_cyclic,
684            is_tree: node_count > 0 && is_connected && undirected_pairs.len() + 1 == node_count,
685            density,
686            density_directed,
687        })
688    }
689}
reddb_server/runtime/impl_graph.rs

reddb_server/runtime/
impl_graph.rs
