Feature/par execution continued

.github/workflows/test.yml

@@ -17,6 +17,10 @@ jobs:
     name: Run Rust tests
     uses: ./.github/workflows/test_rust_workflow.yml
     secrets: inherit
+  call-test-rust-storage-workflow-in-local-repo:
+    name: Run Rust storage tests
+    uses: ./.github/workflows/test_rust_disk_storage_workflow.yml
+    secrets: inherit
   call-test-python-workflow-in-local-repo:
     name: Run Python tests
     uses: ./.github/workflows/test_python_workflow.yml

.github/workflows/test_during_pr.yml

@@ -8,7 +8,6 @@ concurrency:
   group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
   cancel-in-progress: true
 
-
 jobs:
   rust-format-check:
     name: Rust format check
@@ -18,6 +17,11 @@ jobs:
     uses: ./.github/workflows/test_rust_workflow.yml
     secrets: inherit
     needs: rust-format-check
+  call-test-rust-storage-workflow-in-local-repo:
+    name: Run Rust storage tests
+    uses: ./.github/workflows/test_rust_disk_storage_workflow.yml
+    secrets: inherit
+    needs: rust-format-check
   call-test-python-workflow-in-local-repo:
     name: Run Python tests
     uses: ./.github/workflows/test_python_workflow.yml

.github/workflows/test_rust_disk_storage_workflow.yml

@@ -0,0 +1,91 @@
+name: Run Rust test
+on:
+  workflow_call:
+    inputs:
+      skip_tests:
+        type: boolean
+        default: false
+        required: false
+# DO NOT CHANGE NAME OF WORKFLOW, USED IN OTHER WORKFLOWS KEEP "Rust Tests"
+jobs:
+  rust-test:
+    if: ${{ !inputs.skip_tests }}
+    name: Rust Tests
+    runs-on: '${{ matrix.os }}'
+    env:
+      CARGO_NET_GIT_FETCH_WITH_CLI: true
+    strategy:
+      matrix:
+        include:
+          - os: macos-latest
+          - os: ubuntu-latest
+          - os: windows-latest
+    steps:
+      - uses: maxim-lobanov/setup-xcode@v1
+        name: Xcode version
+        if: "contains(matrix.os, 'macOS')"
+        with:
+          xcode-version: latest-stable
+      - uses: actions/checkout@v3
+        name: Checkout
+      - uses: actions-rs/toolchain@v1
+        name: Setup Rust
+        with:
+          profile: minimal
+          toolchain: 1.77.0
+          override: true
+          components: rustfmt, clippy
+      - name: Free up space (ubuntu)
+        if: matrix.os == 'ubuntu-latest'
+        run: |
+          sudo rm -rf /usr/share/dotnet
+          sudo rm -rf /usr/local/lib/android
+          sudo rm -rf /opt/ghc
+          sudo rm -rf "$AGENT_TOOLSDIRECTORY"
+      - uses: webfactory/[email protected]
+        name: Load pometry-storage key
+        with:
+          ssh-private-key: ${{ secrets.RA_SSH_PRIVATE_KEY }}
+      - name: Rust version
+        run: rustc --version --verbose
+      - uses: Swatinem/rust-cache@v2
+        name: Cargo cache
+        with:
+          cache-all-crates: true
+      - name: Install bininstall
+        uses: cargo-bins/cargo-binstall@main
+      - name: Install nextest
+        run: cargo binstall -y --force cargo-nextest
+      - name: Activate pometry-storage in Cargo.toml
+        run: make pull-storage
+      - name: Run all Tests (disk_graph)
+        env:
+          RUSTFLAGS: -Awarnings
+          TEMPDIR: ${{ runner.temp }}
+        run: |
+          cargo nextest run --all --no-default-features --features "storage"
+      - name: Run Tests (features=io)
+        env:
+          RUSTFLAGS: -Awarnings
+        run: |
+          cargo check -p raphtory --no-default-features --features "io"
+      - name: Run Tests (features=python)
+        env:
+          RUSTFLAGS: -Awarnings
+        run: |
+          cargo check -p raphtory --no-default-features --features "python"
+      - name: Run Tests (features=search)
+        env:
+          RUSTFLAGS: -Awarnings
+        run: |
+          cargo check -p raphtory --no-default-features --features "search"
+      - name: Run Tests (features=vectors)
+        env:
+          RUSTFLAGS: -Awarnings
+        run: |
+          cargo check -p raphtory --no-default-features --features "vectors"
+      - name: Run Tests (features=storage)
+        env:
+          RUSTFLAGS: -Awarnings
+        run: |
+          cargo check -p raphtory --no-default-features --features "storage"

.github/workflows/test_rust_workflow.yml

@@ -8,7 +8,7 @@ on:
         required: false
 # DO NOT CHANGE NAME OF WORKFLOW, USED IN OTHER WORKFLOWS KEEP "Rust Tests"
 jobs:
-  rust-test:
+  rust-test-no-storage:
     if: ${{ !inputs.skip_tests }}
     name: Rust Tests
     runs-on: '${{ matrix.os }}'
@@ -62,39 +62,6 @@ jobs:
           TEMPDIR: ${{ runner.temp }}
         run: |
           cargo nextest run --all --no-default-features
-      - name: Activate pometry-storage in Cargo.toml
-        run: make pull-storage
-      - name: Run all Tests (disk_graph)
-        env:
-          RUSTFLAGS: -Awarnings
-          TEMPDIR: ${{ runner.temp }}
-        run: |
-          cargo nextest run --all --no-default-features --features "storage"
-      - name: Run Tests (features=io)
-        env:
-          RUSTFLAGS: -Awarnings
-        run: |
-          cargo check -p raphtory --no-default-features --features "io"
-      - name: Run Tests (features=python)
-        env:
-          RUSTFLAGS: -Awarnings
-        run: |
-          cargo check -p raphtory --no-default-features --features "python"
-      - name: Run Tests (features=search)
-        env:
-          RUSTFLAGS: -Awarnings
-        run: |
-          cargo check -p raphtory --no-default-features --features "search"
-      - name: Run Tests (features=vectors)
-        env:
-          RUSTFLAGS: -Awarnings
-        run: |
-          cargo check -p raphtory --no-default-features --features "vectors"
-      - name: Run Tests (features=storage)
-        env:
-          RUSTFLAGS: -Awarnings
-        run: |
-          cargo check -p raphtory --no-default-features --features "storage"
   doc-test:
     if: ${{ !inputs.skip_tests }}
     name: "Doc tests"

js-raphtory/src/graph/misc.rs

@@ -54,7 +54,7 @@ impl From<JsProp> for JsValue {
             )
             .into(),
             Prop::Graph(v) => Graph(UnderGraph::TGraph(Arc::new(v))).into(),
-            Prop::PersistentGraph(v) => todo!("PersistentGraph not yet implemented"),
+            Prop::PersistentGraph(_v) => todo!("PersistentGraph not yet implemented"),
             Prop::List(v) => {
                 let v: Array = v.iter().map(|v| JsValue::from(JsProp(v.clone()))).collect();
                 v.into()

raphtory-benchmark/benches/common/mod.rs

@@ -4,7 +4,11 @@ use criterion::{
     black_box, measurement::WallTime, BatchSize, Bencher, BenchmarkGroup, BenchmarkId,
 };
 use rand::{distributions::Uniform, seq::*, Rng};
-use raphtory::{core::entities::LayerIds, db::api::view::StaticGraphViewOps, prelude::*};
+use raphtory::{
+    core::entities::{LayerIds, VID},
+    db::api::view::StaticGraphViewOps,
+    prelude::*,
+};
 use std::collections::HashSet;
 
 fn make_index_gen() -> Box<dyn Iterator<Item = u64>> {
@@ -285,7 +289,7 @@ pub fn run_analysis_benchmarks<F, G>(
         .map(|e| (e.src().id(), e.dst().id(), e.time().expect("need time")))
         .collect::<Vec<_>>();
 
-    let nodes: HashSet<u64> = graph.nodes().id().collect();
+    let nodes: HashSet<u64> = graph.nodes().id().into_iter().collect();
 
     bench(group, "num_edges", parameter, |b: &mut Bencher| {
         b.iter(|| graph.count_edges())
@@ -422,19 +426,15 @@ pub fn run_analysis_benchmarks<F, G>(
             let mg = graph.materialize();
             black_box(mg)
         })
-    })
-
-    // Too noisy due to degree variability and confuses criterion
-    // bench(
-    //     group,
-    //     "max_neighbour_degree",
-    //     parameter,
-    //     |b: &mut Bencher| {
-    //         let mut rng = rand::thread_rng();
-    //         let v = graph
-    //             .node(*nodes.iter().choose(&mut rng).expect("non-empty graph"))
-    //             .expect("existing node");
-    //         b.iter(|| v.neighbours().degree().max())
-    //     },
-    // );
+    });
+
+    bench(
+        group,
+        "max_neighbour_degree",
+        parameter,
+        |b: &mut Bencher| {
+            let v = graph.node(VID(0)).expect("graph should not be empty");
+            b.iter(|| v.neighbours().degree().max())
+        },
+    );
 }

raphtory/src/algorithms/community_detection/label_propagation.rs

@@ -50,7 +50,7 @@ where
             }
 
             if let Some(max_label) = find_max_label(&label_count) {
-                if max_label != labels[&node] {
+                if max_label != labels[node] {
                     labels.insert(node.clone(), max_label);
                     changed = true;
                 }

raphtory/src/algorithms/metrics/degree.rs

@@ -43,7 +43,8 @@
 //! print!("Average degree: {:?}", average_degree(&windowed_graph));
 //! ```
 //!
-use crate::db::api::view::*;
+use crate::{db::api::view::*, prelude::*};
+use rayon::prelude::*;
 
 /// The maximum degree of any node in the graph
 pub fn max_degree<'graph, G: GraphViewOps<'graph>>(graph: &G) -> usize {
@@ -80,9 +81,14 @@ pub fn average_degree<'graph, G: GraphViewOps<'graph>>(graph: &'graph G) -> f64
     let (deg_sum, count) = graph
         .nodes()
         .degree()
-        .fold((0usize, 0usize), |(deg_sum, count), deg| {
+        .par_values()
+        .fold_with((0usize, 0usize), |(deg_sum, count), deg| {
             (deg_sum + deg, count + 1)
-        });
+        })
+        .reduce_with(|(deg_sum1, count1), (deg_sum2, count2)| {
+            (deg_sum1 + deg_sum2, count1 + count2)
+        })
+        .unwrap_or((0, 1));
 
     deg_sum as f64 / count as f64
 }

raphtory/src/algorithms/pathing/dijkstra.rs

@@ -155,7 +155,7 @@ pub fn dijkstra_single_source_shortest_paths<G: StaticGraphViewOps, T: AsNodeRef
             let mut current_node_name = node_name.clone();
             while let Some(prev_node) = predecessor.get(&current_node_name) {
                 path.push(prev_node.clone());
-                current_node_name = prev_node.clone();
+                current_node_name.clone_from(prev_node);
             }
             path.reverse();
             paths.insert(node_name.clone(), (cost.clone(), path));

raphtory/src/algorithms/projections/temporal_bipartite_projection.rs

@@ -1,7 +1,6 @@
 use itertools::Itertools;
 use num_integer::average_floor;
 extern crate num_integer;
-
 use crate::{
     core::entities::nodes::node_ref::AsNodeRef,
     db::{
@@ -62,8 +61,6 @@ fn populate_edges<G: StaticGraphViewOps, V: AsNodeRef>(g: &G, new_graph: &Graph,
             }
             to_process.push(nb.clone());
         }
-    } else {
-        return;
     }
 }
 

raphtory/src/core/storage/mod.rs

@@ -46,6 +46,12 @@ impl<T: PartialEq + Default> PartialEq for LockVec<T> {
     }
 }
 
+impl<T: Default> Default for LockVec<T> {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 impl<T: Default> LockVec<T> {
     pub fn new() -> Self {
         Self {

raphtory/src/db/api/mod.rs

@@ -1,4 +1,5 @@
 pub mod mutation;
 pub mod properties;
+pub mod state;
 pub mod storage;
 pub mod view;

raphtory/src/db/api/properties/temporal_props.rs

@@ -71,11 +71,9 @@ impl<P: PropertiesOps> TemporalPropertyView<P> {
                     current_entry = Some((t, prop.clone()));
                 }
                 last_seen_value = Some(prop.clone());
-            } else {
-                if last_seen_value != Some(prop.clone()) {
-                    result.push((t, prop.clone()));
-                    last_seen_value = Some(prop.clone());
-                }
+            } else if last_seen_value != Some(prop.clone()) {
+                result.push((t, prop.clone()));
+                last_seen_value = Some(prop.clone());
             }
         }
 

raphtory/src/db/api/state/group_by.rs

@@ -0,0 +1 @@
+