feat: migrate time bucket bitmaps to BitmapSilo storage (#137)

feat: migrate time bucket bitmaps to BitmapSilo storage

Author: JustMaier

src/engine/concurrent_engine.rs

@@ -436,6 +436,7 @@ impl ConcurrentEngine {
             let flush_config = Arc::clone(&config);
             let flush_field_registry = field_registry.clone();
             let flush_mutation_rx = mutation_rx;
+            let flush_bitmap_silo = bitmap_silo_arc.clone();
             thread::spawn(move || {
                 super::flush::run_flush_thread(super::flush::FlushArgs {
                     slots: flush_slots,
@@ -461,6 +462,7 @@ impl ConcurrentEngine {
                     field_registry: flush_field_registry,
                     mutation_rx: flush_mutation_rx,
                     doc_rx,
+                    bitmap_silo: flush_bitmap_silo,
                 });
             })
         };

src/engine/flush.rs

@@ -39,6 +39,8 @@ pub struct FlushArgs {
     pub field_registry: FieldRegistry,
     pub mutation_rx: Receiver<MutationOp>,
     pub doc_rx: Receiver<(u32, StoredDoc)>,
+    /// BitmapSilo for writing time bucket SET/CLEAR ops alongside in-memory updates.
+    pub bitmap_silo: Option<Arc<parking_lot::RwLock<crate::silos::bitmap_silo::BitmapSilo>>>,
 }
 
 /// Entry point for the flush thread. Runs until `args.shutdown` is set.
@@ -72,6 +74,7 @@ pub fn run_flush_thread(args: FlushArgs) {
         field_registry: flush_field_registry,
         mutation_rx: flush_mutation_rx,
         doc_rx,
+        bitmap_silo: flush_bitmap_silo,
     } = args;
 
     let min_sleep = Duration::from_micros(flush_interval_us);
@@ -137,16 +140,48 @@ pub fn run_flush_thread(args: FlushArgs) {
                     let mut tb = tb_arc.lock();
                     if !batch.alive_inserts.is_empty() {
                         let sort_field_name = tb.sort_field_name().to_string();
+                        let field_name = tb.field_name().to_string();
+                        let bucket_names: Vec<String> = tb.bucket_names();
                         let sorts_r = flush_sorts.read();
                         if let Some(sort_field) = sorts_r.get_field(&sort_field_name) {
                             for &slot in &batch.alive_inserts {
                                 let ts = sort_field.reconstruct_value(slot) as u64;
+                                // Determine which buckets this slot qualifies for (same logic as insert_slot)
+                                let qualifying: Vec<String> = bucket_names.iter()
+                                    .filter(|name| {
+                                        if let Some(bucket) = tb.get_bucket(name) {
+                                            let cutoff = now_secs.saturating_sub(bucket.duration_secs);
+                                            ts >= cutoff && ts <= now_secs
+                                        } else {
+                                            false
+                                        }
+                                    })
+                                    .cloned()
+                                    .collect();
                                 tb.insert_slot(slot, ts, now_secs);
+                                // Mirror to silo
+                                if let Some(ref silo_arc) = flush_bitmap_silo {
+                                    let silo = silo_arc.read();
+                                    for bucket_name in &qualifying {
+                                        let _ = silo.bucket_set(&field_name, bucket_name, slot);
+                                    }
+                                }
                             }
                         }
                     }
-                    for &slot in &batch.alive_removes {
-                        tb.remove_slot(slot);
+                    if !batch.alive_removes.is_empty() {
+                        let field_name = tb.field_name().to_string();
+                        let bucket_names: Vec<String> = tb.bucket_names();
+                        for &slot in &batch.alive_removes {
+                            tb.remove_slot(slot);
+                            // Mirror to silo — unconditionally clear from all buckets
+                            if let Some(ref silo_arc) = flush_bitmap_silo {
+                                let silo = silo_arc.read();
+                                for bucket_name in &bucket_names {
+                                    let _ = silo.bucket_clear(&field_name, bucket_name, slot);
+                                }
+                            }
+                        }
                     }
                 }
             }
@@ -298,6 +333,19 @@ pub fn run_flush_thread(args: FlushArgs) {
                                     bucket.subtract_expired(&expired, new_cutoff);
                                 }
                             }
+                            // Mirror expired CLEARs to silo
+                            if !expired.is_empty() {
+                                let field_name = {
+                                    let tb = tb_arc.lock();
+                                    tb.field_name().to_string()
+                                };
+                                if let Some(ref silo_arc) = flush_bitmap_silo {
+                                    let silo = silo_arc.read();
+                                    for slot in expired.iter() {
+                                        let _ = silo.bucket_clear(&field_name, bucket_name, slot);
+                                    }
+                                }
+                            }
                             // Store diff for lazy cache application (no cache Mutex!)
                             let diff = crate::bucket_diff_log::BucketDiff {
                                 cutoff_before: *old_cutoff,

src/engine/query.rs

@@ -47,7 +47,7 @@ impl ConcurrentEngine {
             tb_guard.as_deref().map(|tb| (tb, now_unix)),
         );
         let (filter_arc, use_simple_sort) =
-            self.resolve_filters(&executor, filters, tb_guard.as_deref(), now_unix)?;
+            self.resolve_filters(&executor, filters, tb_guard.as_deref(), now_unix, silo_guard.as_deref())?;
         let result =
             executor.execute_from_bitmap(&filter_arc, sort, limit, None, use_simple_sort)?;
         Ok(result)
@@ -98,6 +98,7 @@ impl ConcurrentEngine {
                 now_secs: now_unix,
                 tolerance_pct: 0.10,
                 always_snap: true,
+                bitmap_silo: silo_guard.as_deref(),
             };
             snapped_filters = crate::query::snap_range_clauses(&query.filters, &ctx);
             &snapped_filters[..]
@@ -213,9 +214,9 @@ impl ConcurrentEngine {
         let filter_start = Instant::now();
         let (filter_arc, use_simple_sort) = if let Some(ref c) = collector {
             let _ = c;
-            self.resolve_filters(&executor, effective_filters, tb_guard.as_deref(), now_unix)?
+            self.resolve_filters(&executor, effective_filters, tb_guard.as_deref(), now_unix, silo_guard.as_deref())?
         } else {
-            self.resolve_filters(&executor, effective_filters, tb_guard.as_deref(), now_unix)?
+            self.resolve_filters(&executor, effective_filters, tb_guard.as_deref(), now_unix, silo_guard.as_deref())?
         };
         let filter_elapsed = filter_start.elapsed();
         let full_total_matched = filter_arc.len();
@@ -393,6 +394,7 @@ impl ConcurrentEngine {
         filters: &[FilterClause],
         time_buckets: Option<&TimeBucketManager>,
         now_unix: u64,
+        silo: Option<&crate::silos::bitmap_silo::BitmapSilo>,
     ) -> Result<(Arc<roaring::RoaringBitmap>, bool)> {
         // Snap range filters to pre-computed time bucket bitmaps (C3).
         // This must happen BEFORE canonicalization so cache keys use stable
@@ -406,6 +408,7 @@ impl ConcurrentEngine {
                 now_secs: now_unix,
                 tolerance_pct: 0.10,
                 always_snap: true,
+                bitmap_silo: silo,
             };
             snapped = crate::query::snap_range_clauses(filters, &ctx);
             &snapped[..]

src/query/mod.rs

@@ -205,6 +205,10 @@ pub struct BucketSnapContext<'a> {
     /// If true, queries outside tolerance snap to the nearest bucket instead of returning empty.
     /// Default: true (always snap for safety).
     pub always_snap: bool,
+    /// Optional BitmapSilo reference. When present, bucket bitmaps are read from the silo
+    /// via ops-on-read (`get_bucket_with_ops`) instead of from the in-memory TimeBucketManager.
+    /// The manager is still used for config (snap_duration, snap_nearest, bucket names/durations).
+    pub bitmap_silo: Option<&'a crate::silos::bitmap_silo::BitmapSilo>,
 }
 
 /// Pre-process filter clauses: replace range filters on bucketed timestamp fields with
@@ -239,18 +243,12 @@ fn snap_clause(clause: &FilterClause, ctx: &BucketSnapContext<'_>) -> FilterClau
                     // bucket that covers the requested duration, or the largest bucket.
                     let duration_secs = ctx.now_secs.saturating_sub(*ts as u64);
                     let bucket_name = manager.snap_nearest(duration_secs);
-                    if let Some(bucket) = manager.get_bucket(bucket_name) {
-                        FilterClause::BucketBitmap {
-                            field: field.clone(),
-                            bucket_name: bucket_name.to_string(),
-                            bitmap: Arc::clone(bucket.bitmap()),
-                        }
-                    } else {
-                        FilterClause::BucketBitmap {
-                            field: field.clone(),
-                            bucket_name: "_none".to_string(),
-                            bitmap: Arc::new(RoaringBitmap::new()),
-                        }
+                    let bitmap = resolve_bucket_bitmap(ctx.bitmap_silo, manager, field, bucket_name)
+                        .unwrap_or_else(|| Arc::new(RoaringBitmap::new()));
+                    FilterClause::BucketBitmap {
+                        field: field.clone(),
+                        bucket_name: bucket_name.to_string(),
+                        bitmap,
                     }
                 } else {
                     // Unsnapped queries allowed — return empty bitmap for out-of-range.
@@ -289,14 +287,30 @@ fn try_snap_to_bucket(
     // duration = now - ts (the window the filter requests)
     let duration_secs = ctx.now_secs.saturating_sub(ts as u64);
     let bucket_name = manager.snap_duration(duration_secs, ctx.tolerance_pct)?;
-    let bucket = manager.get_bucket(bucket_name)?;
+    let bitmap = resolve_bucket_bitmap(ctx.bitmap_silo, manager, field, bucket_name)?;
     Some(FilterClause::BucketBitmap {
         field: field.to_string(),
         bucket_name: bucket_name.to_string(),
-        bitmap: Arc::clone(bucket.bitmap()),
+        bitmap,
     })
 }
 
+/// Resolve a bucket bitmap: check silo first (ops-on-read), fall back to in-memory manager.
+fn resolve_bucket_bitmap(
+    silo: Option<&crate::silos::bitmap_silo::BitmapSilo>,
+    manager: &crate::time_buckets::TimeBucketManager,
+    field: &str,
+    bucket_name: &str,
+) -> Option<Arc<RoaringBitmap>> {
+    if let Some(silo) = silo {
+        if let Some(bm) = silo.get_bucket_with_ops(field, bucket_name) {
+            return Some(Arc::new(bm));
+        }
+    }
+    // Fall back to in-memory bucket
+    manager.get_bucket(bucket_name).map(|b| Arc::clone(b.bitmap()))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -314,6 +328,7 @@ mod tests {
             now_secs,
             tolerance_pct: 0.10,
             always_snap: true,
+            bitmap_silo: None,
         }
     }
 
@@ -411,6 +426,7 @@ mod tests {
             now_secs: now,
             tolerance_pct: 0.10,
             always_snap: false,
+            bitmap_silo: None,
         };
 
         // Duration = 200000s, outside tolerance, always_snap=false → empty bitmap
@@ -489,6 +505,92 @@ mod tests {
         assert!(matches!(&snapped[0], FilterClause::Gt(_, _)));
     }
 
+    /// When a BitmapSilo is available, snap_clause should read from it instead of
+    /// the in-memory TimeBucketManager bitmap.
+    #[test]
+    fn test_snap_reads_from_silo_when_available() {
+        let now: u64 = 1_700_000_000;
+        let dir = tempfile::tempdir().unwrap();
+
+        // Build a silo with a specific bitmap for "sortAt"/"24h"
+        let silo = crate::silos::bitmap_silo::BitmapSilo::open(dir.path()).unwrap();
+        let mut silo_bm = roaring::RoaringBitmap::new();
+        silo_bm.extend([100u32, 200, 300]); // distinct from in-memory
+        silo.save_bucket("sortAt", "24h", &silo_bm).unwrap();
+
+        // Build a TimeBucketManager with DIFFERENT in-memory bitmap (slots 1, 2, 3)
+        let mgr = make_manager_with_data(now);
+        // Verify the in-memory manager has slots 1-3 for "24h", not 100-300
+        {
+            let bm = mgr.get_bucket("24h").unwrap().bitmap();
+            assert!(bm.contains(1));
+            assert!(!bm.contains(100));
+        }
+
+        let mut managers = HashMap::new();
+        managers.insert("sortAt".to_string(), &mgr);
+
+        // Build context with silo
+        let ctx = BucketSnapContext {
+            managers: &managers,
+            now_secs: now,
+            tolerance_pct: 0.10,
+            always_snap: true,
+            bitmap_silo: Some(&silo),
+        };
+
+        // Snap to "24h" — should use silo bitmap (100, 200, 300), not in-memory (1, 2, 3)
+        let ts = (now - 86400) as i64; // exactly 24h
+        let clauses = vec![FilterClause::Gt("sortAt".to_string(), Value::Integer(ts))];
+        let snapped = snap_range_clauses(&clauses, &ctx);
+
+        match &snapped[0] {
+            FilterClause::BucketBitmap { field, bucket_name, bitmap } => {
+                assert_eq!(field, "sortAt");
+                assert_eq!(bucket_name, "24h");
+                // Should come from silo, not in-memory manager
+                assert!(bitmap.contains(100), "should have silo slot 100");
+                assert!(bitmap.contains(200), "should have silo slot 200");
+                assert!(bitmap.contains(300), "should have silo slot 300");
+                assert!(!bitmap.contains(1), "should NOT have in-memory slot 1");
+                assert_eq!(bitmap.len(), 3);
+            }
+            other => panic!("expected BucketBitmap, got {:?}", other),
+        }
+    }
+
+    /// When silo is None, snap_clause falls back to in-memory manager bitmap.
+    #[test]
+    fn test_snap_falls_back_to_in_memory_without_silo() {
+        let now: u64 = 1_700_000_000;
+        let mgr = make_manager_with_data(now);
+        let mut managers = HashMap::new();
+        managers.insert("sortAt".to_string(), &mgr);
+
+        let ctx = BucketSnapContext {
+            managers: &managers,
+            now_secs: now,
+            tolerance_pct: 0.10,
+            always_snap: true,
+            bitmap_silo: None,
+        };
+
+        let ts = (now - 86400) as i64;
+        let clauses = vec![FilterClause::Gt("sortAt".to_string(), Value::Integer(ts))];
+        let snapped = snap_range_clauses(&clauses, &ctx);
+
+        match &snapped[0] {
+            FilterClause::BucketBitmap { bitmap, .. } => {
+                // In-memory manager has slots 1-3 in "24h"
+                assert!(bitmap.contains(1));
+                assert!(bitmap.contains(2));
+                assert!(bitmap.contains(3));
+                assert_eq!(bitmap.len(), 3);
+            }
+            other => panic!("expected BucketBitmap, got {:?}", other),
+        }
+    }
+
     #[test]
     fn test_filter_clause_construction() {
         let clause = FilterClause::And(vec![