propagate changes

Author: teh-cmc

crates/store/re_chunk_store/src/events.rs

@@ -266,7 +266,7 @@ mod tests {
             .with_component_batch(row_id1, timepoint1.clone(), &MyIndex::from_iter(0..10))
             .build()?;
 
-        view.on_events(&[store.insert_chunk(&Arc::new(chunk1))?.unwrap()]);
+        view.on_events(&store.insert_chunk(&Arc::new(chunk1))?);
 
         similar_asserts::assert_eq!(
             GlobalCounts::new(
@@ -311,7 +311,7 @@ mod tests {
                 .build()?
         };
 
-        view.on_events(&[store.insert_chunk(&Arc::new(chunk2))?.unwrap()]);
+        view.on_events(&store.insert_chunk(&Arc::new(chunk2))?);
 
         similar_asserts::assert_eq!(
             GlobalCounts::new(
@@ -360,7 +360,7 @@ mod tests {
                 .build()?
         };
 
-        view.on_events(&[store.insert_chunk(&Arc::new(chunk3))?.unwrap()]);
+        view.on_events(&store.insert_chunk(&Arc::new(chunk3))?);
 
         similar_asserts::assert_eq!(
             GlobalCounts::new(

crates/store/re_entity_db/src/entity_db.rs

@@ -1,6 +1,5 @@
 use std::sync::Arc;
 
-use itertools::Itertools;
 use nohash_hasher::IntMap;
 use parking_lot::Mutex;
 
@@ -71,7 +70,11 @@ pub struct EntityDb {
 
 impl EntityDb {
     pub fn new(store_id: StoreId) -> Self {
-        let data_store = ChunkStore::new(store_id.clone(), ChunkStoreConfig::default());
+        Self::with_store_config(store_id, ChunkStoreConfig::from_env().unwrap_or_default())
+    }
+
+    pub fn with_store_config(store_id: StoreId, store_config: ChunkStoreConfig) -> Self {
+        let data_store = ChunkStore::new(store_id.clone(), store_config);
         let query_caches = re_query::Caches::new(&data_store);
 
         Self {
@@ -347,23 +350,23 @@ impl EntityDb {
     }
 
     pub fn add_chunk(&mut self, chunk: &Arc<Chunk>) -> Result<(), Error> {
-        let store_event = self.data_store.insert_chunk(chunk)?;
+        let store_events = self.data_store.insert_chunk(chunk)?;
 
         self.register_entity_path(chunk.entity_path());
 
         if self.latest_row_id < chunk.row_id_range().map(|(_, row_id_max)| row_id_max) {
             self.latest_row_id = chunk.row_id_range().map(|(_, row_id_max)| row_id_max);
         }
 
-        if let Some(store_event) = store_event {
+        {
             // Update our internal views by notifying them of resulting [`ChunkStoreEvent`]s.
-            let original_store_events = &[store_event];
-            self.times_per_timeline.on_events(original_store_events);
-            self.query_caches.on_events(original_store_events);
-            self.tree.on_store_additions(original_store_events);
+            self.times_per_timeline.on_events(&store_events);
+            self.query_caches.on_events(&store_events);
+            self.tree.on_store_additions(&store_events);
+            self.tree.on_store_deletions(&store_events);
 
             // We inform the stats last, since it measures e2e latency.
-            self.stats.on_events(original_store_events);
+            self.stats.on_events(&store_events);
         }
 
         Ok(())
@@ -435,8 +438,7 @@ impl EntityDb {
         times_per_timeline.on_events(store_events);
         query_caches.on_events(store_events);
 
-        let store_events = store_events.iter().collect_vec();
-        tree.on_store_deletions(&store_events);
+        tree.on_store_deletions(store_events);
     }
 
     /// Key used for sorting recordings in the UI.

crates/store/re_entity_db/src/entity_tree.rs

@@ -238,10 +238,9 @@ impl EntityTree {
 
     /// Updates the [`EntityTree`] by applying a batch of [`ChunkStoreEvent`]s.
     ///
-    /// Only reacts to additions (`event.kind == StoreDiffKind::Addition`).
+    /// Only reacts to deletions (`event.kind == StoreDiffKind::Deletion`).
     pub fn on_store_additions(&mut self, events: &[ChunkStoreEvent]) {
         re_tracing::profile_function!();
-
         for event in events
             .iter()
             .filter(|e| e.kind == ChunkStoreDiffKind::Addition)
@@ -297,7 +296,7 @@ impl EntityTree {
     /// Updates the [`EntityTree`] by applying a batch of [`ChunkStoreEvent`]s.
     ///
     /// Only reacts to deletions (`event.kind == StoreDiffKind::Deletion`).
-    pub fn on_store_deletions(&mut self, store_events: &[&ChunkStoreEvent]) {
+    pub fn on_store_deletions(&mut self, store_events: &[ChunkStoreEvent]) {
         re_tracing::profile_function!();
 
         let Self {
@@ -310,8 +309,9 @@ impl EntityTree {
         // Only keep events relevant to this branch of the tree.
         let subtree_events = store_events
             .iter()
-            .filter(|e| e.diff.chunk.entity_path().starts_with(path))
-            .copied() // NOTE: not actually copying, just removing the superfluous ref layer
+            .filter(|e| e.kind == ChunkStoreDiffKind::Deletion)
+            .filter(|&e| e.diff.chunk.entity_path().starts_with(path))
+            .cloned()
             .collect_vec();
 
         {
@@ -341,7 +341,7 @@ impl EntityTree {
 
         {
             re_tracing::profile_scope!("subtree");
-            for &event in &subtree_events {
+            for event in &subtree_events {
                 subtree.on_event(event);
             }
         }

crates/store/re_entity_db/src/time_histogram_per_timeline.rs

@@ -100,12 +100,13 @@ impl TimeHistogramPerTimeline {
                 });
         } else {
             for &(timeline, times) in times_per_timeline {
-                let histogram = self.times.entry(timeline).or_default();
-                for &time in times {
-                    histogram.decrement(time, n);
-                }
-                if histogram.is_empty() {
-                    self.times.remove(&timeline);
+                if let Some(histo) = self.times.get_mut(&timeline) {
+                    for &time in times {
+                        histo.decrement(time, n);
+                    }
+                    if histo.is_empty() {
+                        self.times.remove(&timeline);
+                    }
                 }
             }
         }

crates/store/re_query/benches/latest_at.rs

@@ -254,7 +254,7 @@ fn insert_chunks<'a>(msgs: impl Iterator<Item = &'a Arc<Chunk>>) -> (Caches, Chu
     let mut caches = Caches::new(&store);
 
     msgs.for_each(|chunk| {
-        caches.on_events(&[store.insert_chunk(chunk).unwrap().unwrap()]);
+        caches.on_events(&store.insert_chunk(chunk).unwrap());
     });
 
     (caches, store)

crates/store/re_query/tests/latest_at.rs

@@ -545,7 +545,7 @@ fn static_invalidation() {
 // ---
 
 fn insert_and_react(store: &mut ChunkStore, caches: &mut Caches, chunk: &Arc<Chunk>) {
-    caches.on_events(&[store.insert_chunk(chunk).unwrap().unwrap()]);
+    caches.on_events(&store.insert_chunk(chunk).unwrap());
 }
 
 fn query_and_compare(

crates/store/re_query/tests/range.rs

@@ -1017,7 +1017,7 @@ fn concurrent_multitenant_edge_case2() {
 // // ---
 
 fn insert_and_react(store: &mut ChunkStore, caches: &mut Caches, chunk: &Arc<Chunk>) {
-    caches.on_events(&[store.insert_chunk(chunk).unwrap().unwrap()]);
+    caches.on_events(&store.insert_chunk(chunk).unwrap());
 }
 
 fn query_and_compare(