apsp

Cargo.toml

@@ -33,6 +33,7 @@ itertools = "0.10.3"
 regex = "1.6.0"
 pest = "2.2.1"
 pest_derive = "2.2.1"
+rayon = "1.5.3"
 cozorocks = { path = "cozorocks" }
 
 #[target.'cfg(not(target_env = "msvc"))'.dependencies]

README.md

@@ -21,4 +21,6 @@
   * [x] connected components
   * [ ] label propagation
   * [ ] louvain modularity
-* [x] direct loading of data
+* [x] direct loading of data
+* [ ] serial agg function
+* [ ] random function

src/algo/all_pairs_shortest_path.rs

@@ -0,0 +1,37 @@
+use std::cmp::Reverse;
+
+use ordered_float::OrderedFloat;
+use priority_queue::PriorityQueue;
+use rayon::prelude::*;
+
+pub(crate) fn dijkstra_cost_only(edges: &[Vec<(usize, f64)>], start: usize) -> Vec<f64> {
+    let mut distance = vec![f64::INFINITY; edges.len()];
+    let mut pq = PriorityQueue::new();
+    let mut back_pointers = vec![usize::MAX; edges.len()];
+    distance[start] = 0.;
+    pq.push(start, Reverse(OrderedFloat(0.)));
+
+    while let Some((node, Reverse(OrderedFloat(cost)))) = pq.pop() {
+        if cost > distance[node] {
+            continue;
+        }
+
+        for (nxt_node, path_weight) in &edges[node] {
+            let nxt_cost = cost + *path_weight;
+            if nxt_cost < distance[*nxt_node] {
+                pq.push_increase(*nxt_node, Reverse(OrderedFloat(nxt_cost)));
+                distance[*nxt_node] = nxt_cost;
+                back_pointers[*nxt_node] = node;
+            }
+        }
+    }
+
+    distance
+}
+
+pub(crate) fn apsp(edges: &[Vec<(usize, f64)>]) -> Vec<Vec<f64>> {
+    (0..edges.len())
+        .into_par_iter()
+        .map(|start| dijkstra_cost_only(edges, start))
+        .collect()
+}

src/algo/mod.rs

@@ -35,6 +35,7 @@ pub(crate) mod strongly_connected_components;
 pub(crate) mod top_sort;
 pub(crate) mod triangles;
 pub(crate) mod yen;
+pub(crate) mod all_pairs_shortest_path;
 
 pub(crate) trait AlgoImpl {
     fn run(

src/query/compile.rs

@@ -23,6 +23,8 @@ pub(crate) enum CompiledRuleSet {
     Algo(MagicAlgoApply),
 }
 
+unsafe impl Send for CompiledRuleSet {}
+
 #[derive(Debug, Copy, Clone, Eq, PartialEq)]
 pub(crate) enum AggrKind {
     None,

src/query/eval.rs

@@ -28,7 +28,7 @@ impl QueryLimiter {
 
 impl SessionTx {
     pub(crate) fn stratified_magic_evaluate(
-        &mut self,
+        &self,
         strata: &[CompiledProgram],
         stores: &BTreeMap<MagicSymbol, DerivedRelStore>,
         num_to_take: Option<usize>,
@@ -47,7 +47,7 @@ impl SessionTx {
         Ok(ret_area)
     }
     fn semi_naive_magic_evaluate(
-        &mut self,
+        &self,
         prog: &CompiledProgram,
         stores: &BTreeMap<MagicSymbol, DerivedRelStore>,
         num_to_take: Option<usize>,
@@ -132,7 +132,7 @@ impl SessionTx {
         Ok(())
     }
     fn algo_application_eval(
-        &mut self,
+        &self,
         rule_symb: &MagicSymbol,
         algo_apply: &MagicAlgoApply,
         stores: &BTreeMap<MagicSymbol, DerivedRelStore>,
@@ -150,7 +150,7 @@ impl SessionTx {
         )
     }
     fn initial_rule_eval(
-        &mut self,
+        &self,
         rule_symb: &MagicSymbol,
         ruleset: &[CompiledRule],
         aggr_kind: AggrKind,
@@ -224,7 +224,7 @@ impl SessionTx {
         Ok(())
     }
     fn incremental_rule_eval(
-        &mut self,
+        &self,
         rule_symb: &MagicSymbol,
         ruleset: &[CompiledRule],
         epoch: u32,