Optimisations for node validity and neighbour filtering

Cargo.toml

@@ -48,6 +48,7 @@ debug = true
 [profile.with-debug]
 inherits = "dev"
 debug = true
+opt-level = 0
 
 # for fast one-time builds (e.g., docs/CI)
 [profile.build-fast]
@@ -92,7 +93,6 @@ itertools = "0.13.0"
 rand = "0.9.2"
 rayon = "1.11.0"
 roaring = "0.10.12"
-sorted_vector_map = "0.2.0"
 tokio = { version = "1.50.0", features = ["full"] }
 once_cell = "1.21.4"
 parking_lot = { version = "0.12.5", features = [
@@ -195,7 +195,6 @@ strum = "0.27.2"
 strum_macros = "0.27.2"
 pythonize = { version = "0.27.0" }
 test-log = "0.2.19"
-disjoint-sets = "0.4.2"
 datafusion = { version = "52.5.0" }
 datafusion-physical-expr-common = "52.5.0"
 datafusion-physical-expr = "52.5.0"

db4-graph/src/lib.rs

@@ -163,6 +163,11 @@ where
         self.logical_to_physical.flush()
     }
 
+    pub fn vacuum(&self) -> Result<(), StorageError> {
+        self.storage.vacuum()?;
+        Ok(())
+    }
+
     pub fn disk_storage_path(&self) -> Option<&Path> {
         self.graph_dir()
             .filter(|_| Extension::disk_storage_enabled())
@@ -402,7 +407,7 @@ where
         WriteLockedGraph::new(self)
     }
 
-    pub fn update_time(&self, earliest: EventTime) {
+    pub fn update_time(&self, _earliest: EventTime) {
         // self.storage.update_time(earliest);
     }
 }

db4-storage/src/api/edges.rs

@@ -5,10 +5,11 @@ use crate::{
 use parking_lot::{RwLockReadGuard, RwLockWriteGuard, lock_api::ArcRwLockReadGuard};
 use raphtory_api::core::entities::{
     LayerId,
+    edges::edge_ref::Dir,
     properties::{meta::Meta, prop::Prop, tprop::TPropOps},
 };
 use raphtory_core::{
-    entities::{EID, LayerIds, VID},
+    entities::{EID, LayerIds, VID, edges::edge_ref::EdgeRef},
     storage::timeindex::{EventTime, TimeIndexOps},
 };
 use rayon::iter::ParallelIterator;
@@ -32,7 +33,7 @@ pub trait EdgeSegmentOps: Send + Sync + std::fmt::Debug + 'static {
     fn latest(&self) -> Option<EventTime>;
     fn earliest(&self) -> Option<EventTime>;
 
-    fn t_len(&self) -> usize;
+    fn t_len(&self, layer_id: usize) -> usize;
     fn num_layers(&self) -> usize;
     // Persistent layer count, not used for up-to-date counts
     fn layer_count(&self, layer_id: LayerId) -> u32;
@@ -97,7 +98,7 @@ pub trait EdgeSegmentOps: Send + Sync + std::fmt::Debug + 'static {
         locked_head: impl Deref<Target = MemEdgeSegment>,
     ) -> Option<(VID, VID)>;
 
-    fn entry<'a>(&'a self, edge_pos: LocalPOS) -> Self::Entry<'a>;
+    fn entry<'a>(&'a self, edge_pos: LocalPOS, edge_ref: Option<EdgeRef>) -> Self::Entry<'a>;
 
     fn layer_entry<'a>(
         &'a self,
@@ -124,7 +125,11 @@ pub trait LockedESegment: Send + Sync + std::fmt::Debug {
     where
         Self: 'a;
 
-    fn entry_ref<'a>(&'a self, edge_pos: impl Into<LocalPOS>) -> Self::EntryRef<'a>
+    fn entry_ref<'a>(
+        &'a self,
+        edge_pos: impl Into<LocalPOS>,
+        edge_ref: Option<EdgeRef>,
+    ) -> Self::EntryRef<'a>
     where
         Self: 'a;
 
@@ -157,11 +162,7 @@ pub trait EdgeRefOps<'a>: Copy + Clone + Send + Sync {
     type Deletions: TimeIndexOps<'a, IndexType = EventTime>;
     type TProps: TPropOps<'a>;
 
-    fn edge(self, layer_id: LayerId) -> Option<(VID, VID)>;
-
-    fn has_layer_inner(self, layer_id: LayerId) -> bool {
-        self.edge(layer_id).is_some()
-    }
+    fn has_layer_inner(self, layer_id: LayerId) -> bool;
 
     fn internal_num_layers(self) -> usize;
 
@@ -172,9 +173,10 @@ pub trait EdgeRefOps<'a>: Copy + Clone + Send + Sync {
 
     fn layer_t_prop(self, layer_id: LayerId, prop_id: usize) -> Self::TProps;
 
-    fn src(&self) -> Option<VID>;
+    fn src(&self) -> VID;
 
-    fn dst(&self) -> Option<VID>;
+    fn dst(&self) -> VID;
 
     fn edge_id(&self) -> EID;
+    fn edge_ref(self, dir: Dir) -> EdgeRef;
 }

db4-storage/src/gen_ts.rs

@@ -349,6 +349,7 @@ impl<'a, Ref: WithTimeCells<'a> + 'a> TimeIndexOps<'a> for GenericTimeOps<'a, Re
 
     type RangeType = Self;
 
+    #[inline]
     fn active(&self, w: Range<Self::IndexType>) -> bool {
         self.clone()
             .time_cells()

db4-storage/src/pages/edge_store.rs

@@ -12,13 +12,14 @@ use crate::{
     persist::{config::ConfigOps, strategy::PersistenceStrategy},
     segments::edge::segment::MemEdgeSegment,
 };
+use either::Either;
 use parking_lot::{RwLock, RwLockWriteGuard};
 use raphtory_api::core::entities::{
     EID, LayerId, VID,
     properties::meta::{Meta, STATIC_GRAPH_LAYER_ID},
 };
 use raphtory_core::{
-    entities::{ELID, LayerIds},
+    entities::{ELID, LayerIds, edges::edge_ref::EdgeRef},
     storage::timeindex::{AsTime, EventTime},
 };
 use rayon::prelude::*;
@@ -56,12 +57,12 @@ impl<ES: EdgeSegmentOps<Extension = EXT>, EXT: PersistenceStrategy<ES = ES>>
 
     pub fn edge_ref(
         &self,
-        e_id: impl Into<EID>,
+        e_id_ref: Either<EID, EdgeRef>,
     ) -> <<ES as EdgeSegmentOps>::ArcLockedSegment as LockedESegment>::EntryRef<'_> {
-        let e_id = e_id.into();
+        let e_id = e_id_ref.either(|eid| eid, |eref| eref.pid());
         let (page_id, pos) = self.storage.resolve_pos(e_id);
         let locked_page = &self.locked_pages[page_id];
-        locked_page.entry_ref(pos)
+        locked_page.entry_ref(pos, e_id_ref.right())
     }
 
     pub fn iter<'a, 'b: 'a>(
@@ -122,7 +123,10 @@ impl<ES: EdgeSegmentOps<Extension = EXT>, EXT: PersistenceStrategy<ES = ES>>
         .map(|(row_group_id, iter)| {
             (
                 row_group_id,
-                iter.filter(|eid| self.edge_ref(*eid).edge(LayerId(0)).is_some()),
+                iter.filter(|eid| {
+                    self.edge_ref(Either::Left(*eid))
+                        .has_layer_inner(STATIC_GRAPH_LAYER_ID)
+                }),
             )
         })
     }
@@ -223,8 +227,11 @@ impl<ES: EdgeSegmentOps<Extension = EXT>, EXT: PersistenceStrategy<ES = ES>>
         Iterator::max(self.segments.iter().filter_map(|(_, page)| page.latest()))
     }
 
-    pub fn t_len(&self) -> usize {
-        self.segments.iter().map(|(_, page)| page.t_len()).sum()
+    pub fn t_len(&self, layer_id: usize) -> usize {
+        self.segments
+            .iter()
+            .map(|(_, page)| page.t_len(layer_id))
+            .sum()
     }
 
     pub fn prop_meta(&self) -> &Arc<Meta> {
@@ -467,16 +474,16 @@ impl<ES: EdgeSegmentOps<Extension = EXT>, EXT: PersistenceStrategy<ES = ES>>
         segment.get_edge(local_edge, layer, segment.head())
     }
 
-    pub fn edge(&self, e_id: impl Into<EID>) -> ES::Entry<'_> {
-        let e_id = e_id.into();
+    pub fn edge(&self, e_id_ref: Either<EID, EdgeRef>) -> ES::Entry<'_> {
+        let e_id = e_id_ref.either(|eid| eid, |eref| eref.pid());
         let (segment_id, local_edge) = resolve_pos(e_id, self.max_page_len());
         let segment = self.segments.get(segment_id).unwrap_or_else(|| {
             panic!(
                 "{e_id:?} Not found in seg: {segment_id}, pos: {local_edge:?}, num_segments: {}",
                 self.segments.count()
             )
         });
-        segment.entry(local_edge)
+        segment.entry(local_edge, e_id_ref.right())
     }
 
     pub fn num_edges(&self) -> usize {
@@ -487,6 +494,21 @@ impl<ES: EdgeSegmentOps<Extension = EXT>, EXT: PersistenceStrategy<ES = ES>>
         self.layer_counter.get(layer_id)
     }
 
+    pub fn num_layers(&self) -> usize {
+        self.segments
+            .iter()
+            .map(|(_, page)| page.num_layers())
+            .max()
+            .unwrap_or(0)
+    }
+
+    pub fn num_temporal_edges_layer(&self, layer_id: LayerId) -> usize {
+        self.segments
+            .iter()
+            .map(|(_, page)| page.t_len(layer_id.0))
+            .sum()
+    }
+
     pub fn get_writer<'a>(
         &'a self,
         e_id: EID,