fix materialize issues for persistent graphs

Pometry · Dec 17, 2024 · 5a7758b · 5a7758b
1 parent c2b656a
commit 5a7758b
Show file tree

Hide file tree

Showing 4 changed files with 50 additions and 12 deletions.
diff --git a/raphtory/src/core/entities/properties/tprop.rs b/raphtory/src/core/entities/properties/tprop.rs
@@ -51,13 +51,22 @@ impl<'a> TPropCell<'a> {
             log,
         }
     }
+
+    fn iter_window_inner(
+        self,
+        r: Range<TimeIndexEntry>,
+    ) -> impl DoubleEndedIterator<Item = (TimeIndexEntry, Prop)> + Send + 'a {
+        self.t_cell.into_iter().flat_map(move |t_cell| {
+            t_cell
+                .iter_window(r.clone())
+                .filter_map(move |(t, &id)| self.log?.get(id?).map(|prop| (*t, prop)))
+        })
+    }
 }
 
 impl<'a> TPropOps<'a> for TPropCell<'a> {
     fn last_before(&self, t: TimeIndexEntry) -> Option<(TimeIndexEntry, Prop)> {
-        self.t_cell?
-            .last_before(t)
-            .and_then(|(t, &id)| self.log?.get(id?).map(|prop| (t, prop))) // FIXME: is this correct?
+        self.iter_window_inner(TimeIndexEntry::MIN..t).next_back()
     }
 
     fn iter(self) -> impl Iterator<Item = (TimeIndexEntry, Prop)> + Send + 'a {
@@ -72,11 +81,7 @@ impl<'a> TPropOps<'a> for TPropCell<'a> {
         self,
         r: Range<TimeIndexEntry>,
     ) -> impl Iterator<Item = (TimeIndexEntry, Prop)> + Send + 'a {
-        self.t_cell.into_iter().flat_map(move |t_cell| {
-            t_cell
-                .iter_window(r.clone())
-                .filter_map(move |(t, &id)| self.log?.get(id?).map(|prop| (*t, prop)))
-        })
+        self.iter_window_inner(r)
     }
 
     fn at(self, ti: &TimeIndexEntry) -> Option<Prop> {

diff --git a/raphtory/src/core/storage/node_entry.rs b/raphtory/src/core/storage/node_entry.rs
@@ -66,6 +66,17 @@ impl<'a> NodePtr<'a> {
             })
     }
 
+    pub fn last_before_row(self, t: TimeIndexEntry) -> Vec<(usize, Prop)> {
+        self.t_props_log
+            .iter()
+            .enumerate()
+            .filter_map(|(prop_id, _)| {
+                let t_prop = self.t_prop(prop_id);
+                t_prop.last_before(t).map(|(_, v)| (prop_id, v))
+            })
+            .collect()
+    }
+
     pub fn into_rows_window(
         self,
         w: Range<TimeIndexEntry>,

diff --git a/raphtory/src/db/api/storage/graph/nodes/node_ref.rs b/raphtory/src/db/api/storage/graph/nodes/node_ref.rs
@@ -49,6 +49,14 @@ impl<'a> NodeStorageRef<'a> {
             NodeStorageRef::Disk(disk_node) => disk_node.into_rows_window(window).into_dyn_boxed(),
         }
     }
+
+    pub fn last_before_row(self, t: TimeIndexEntry) -> Vec<(usize, Prop)> {
+        match self {
+            NodeStorageRef::Mem(node_entry) => node_entry.last_before_row(t),
+            #[cfg(feature = "storage")]
+            NodeStorageRef::Disk(disk_node) => disk_node.last_before_row(t),
+        }
+    }
 }
 
 impl<'a> From<NodePtr<'a>> for NodeStorageRef<'a> {

diff --git a/raphtory/src/db/graph/views/deletion_graph.rs b/raphtory/src/db/graph/views/deletion_graph.rs
@@ -268,7 +268,18 @@ impl TimeSemantics for PersistentGraph {
         v: VID,
         w: Option<Range<i64>>,
     ) -> BoxedLIter<(TimeIndexEntry, Vec<(usize, Prop)>)> {
-        self.0.node_history_rows(v, w)
+        // if window exists, we need to add the first row before the window
+        if let Some(w) = w {
+            let t = TimeIndexEntry::start(w.start.saturating_add(1));
+            let first_row = self.core_node_entry(v).as_ref().last_before_row(t);
+            Box::new(
+                std::iter::once(first_row)
+                    .map(move |row| (TimeIndexEntry::start(w.start), row))
+                    .chain(self.0.node_history_rows(v, Some(w))),
+            )
+        } else {
+            self.0.node_history_rows(v, w)
+        }
     }
 
     fn node_property_history(&self, v: VID, w: Option<Range<i64>>) -> BoxedLIter<TimeIndexEntry> {
@@ -686,6 +697,7 @@ mod test_deletions {
             },
         },
         prelude::*,
+        test_storage,
     };
     use itertools::Itertools;
     use raphtory_api::core::{entities::GID, utils::logging::global_info_logger};
@@ -893,9 +905,11 @@ mod test_deletions {
         let g = PersistentGraph::new();
         g.add_node(0, 1, [("test", "test")], None).unwrap();
 
-        let wg = g.window(3, 5);
-        let mg = wg.materialize().unwrap();
-        assert_graph_equal(&wg, &mg);
+        test_storage!(&g, |g| {
+            let wg = g.window(3, 5);
+            let mg = wg.materialize().unwrap();
+            assert_graph_equal(&wg, &mg);
+        })
     }
 
     #[test]