diff --git a/asap-query-engine/src/stores/epoch_columnar.rs b/asap-query-engine/src/stores/epoch_columnar.rs new file mode 100644 index 00000000..af0ebdf4 --- /dev/null +++ b/asap-query-engine/src/stores/epoch_columnar.rs @@ -0,0 +1,509 @@ +//! Epoch-partitioned columnar storage — generic payload type. +//! +//! Lifted from `simple_map_store::common` (legacy SimpleMapStore index) +//! with the payload column type made generic so the new SketchIndex +//! (Phase 5) can reuse the legacy's six storage optimizations +//! (`INDEX_DESIGN.md`) without dragging in `Arc` +//! dynamic dispatch. +//! +//! # Optimizations carried over from legacy +//! +//! | Opt | What | +//! |-----|------| +//! | 1 | Lazy `window_to_ids` index — built on first exact query, invalidated cheaply on insert | +//! | 2 | Offset-based index — stores `u32` column offsets, not payload clones | +//! | 3 | Monotonic ingest fast path — skip `HashSet` probe for consecutive same-window inserts | +//! | 4 | Batch metadata hoisting — caller responsibility (the OTLP receive path groups DPs by sid) | +//! | 5 | Columnar storage — three parallel arrays; range scan touches only `windows_col` | +//! | 6 | Pre-allocated epoch buffers on rotation | +//! +//! # Differences from legacy +//! +//! - **Generic payload type**: `MutableEpoch

` instead of +//! `Vec>`. The new SketchIndex stores +//! `SketchSampleState` directly (typed bytes + encoding tag) — no +//! dyn dispatch, no Arc cloning, payload moves into the column. +//! - **Series-values keyed via `LabelValuesId = u32`** (renamed from +//! legacy `MetricID = u32`). The intern table maps the per-series +//! group-by VALUES vector to a compact ID, since the SketchIndex's +//! sid already captures the metric identity at the level above. +//! +//! See INDEX_DESIGN.md in `simple_map_store/` for the full complexity +//! analysis (Insert O(1), range query O(M) mutable / O(log N + k) +//! sealed, etc.) — those bounds carry over verbatim because the +//! algorithmic structure is unchanged. + +use std::collections::{BTreeMap, HashMap, HashSet}; + +/// Compact identifier for an interned label-values vector. 4 bytes — +/// hashing/comparing this in the hot loop is one CPU instruction +/// instead of a `BTreeMap` walk. +pub type LabelValuesId = u32; + +/// Monotonically increasing epoch counter. +pub type EpochId = u64; + +/// `(start_unix_ms, end_unix_ms)`. +pub type TimestampRange = (u64, u64); + +/// Intern table for per-series group-by VALUES vectors. +/// +/// Each `SidStoreData` owns one InternTable. The first time a series's +/// label-values vector is seen, it gets assigned a `LabelValuesId` +/// (`u32`); all subsequent appearances re-use the existing id. Hot-path +/// columnar arrays carry only the `u32` ids — the full label-values +/// vectors live once in the intern table and are resolved on query. +/// +/// Bumped from legacy's `Option` key to +/// `BTreeMap` because the new SketchIndex receives +/// canonicalized group-by attributes from the OTLP DataPoint (sorted +/// by key already at the receive layer). +pub struct InternTable { + label_to_id: HashMap, + id_to_label: Vec, +} + +impl InternTable { + pub fn new() -> Self { + Self { + label_to_id: HashMap::new(), + id_to_label: Vec::new(), + } + } + + /// Intern a key, assigning a new `LabelValuesId` if first seen. + /// Uses `HashMap::entry` to avoid double-hashing. + pub fn intern(&mut self, key: K) -> LabelValuesId { + let next_id = self.id_to_label.len() as LabelValuesId; + match self.label_to_id.entry(key) { + std::collections::hash_map::Entry::Occupied(e) => *e.get(), + std::collections::hash_map::Entry::Vacant(e) => { + self.id_to_label.push(e.key().clone()); + *e.insert(next_id) + } + } + } + + /// O(1) resolution by id. + pub fn resolve(&self, id: LabelValuesId) -> Option<&K> { + self.id_to_label.get(id as usize) + } + + pub fn len(&self) -> usize { + self.id_to_label.len() + } + + pub fn is_empty(&self) -> bool { + self.id_to_label.is_empty() + } +} + +impl Default for InternTable { + fn default() -> Self { + Self::new() + } +} + +/// Active (mutable) epoch: append-only insert, O(1) amortized. +/// +/// Three parallel columns (Opt 5) — windows / label-id / payload — +/// keep the range-scan hot loop hitting only `windows_col`. The +/// payload column is owned (no `Arc` indirection); legacy used +/// `Arc` because aggregation type was variable, +/// but the SketchIndex specializes to `SketchSampleState` (typed +/// bytes + encoding tag). +pub struct MutableEpoch

{ + // Columnar storage: three parallel arrays (Opt 5) + windows_col: Vec, + label_ids_col: Vec, + payloads_col: Vec

, + + // Distinct-window count for epoch rotation threshold. + windows_set: HashSet, + + // Monotonic ingest fast path: skip windows_set probe for consecutive + // same-window inserts — the common case in ordered ingestion (Opt 3). + last_window: Option, + + // Lazy offset index — Some(_) after the first exact_query; + // invalidated to None on any insert (one pointer-width write). + // Stores `Vec` column offsets, not payload clones (Opt 1 + 2). + window_to_ids: Option>>, + + // Epoch time bounds for O(1) skip-check, updated incrementally on insert. + min_start: Option, + max_end: Option, +} + +impl

MutableEpoch

{ + pub fn new() -> Self { + Self::with_capacity(0) + } + + /// Pre-allocate column buffers with a capacity hint (Opt 6). + pub fn with_capacity(cap: usize) -> Self { + Self { + windows_col: Vec::with_capacity(cap), + label_ids_col: Vec::with_capacity(cap), + payloads_col: Vec::with_capacity(cap), + windows_set: HashSet::new(), + last_window: None, + window_to_ids: None, + min_start: None, + max_end: None, + } + } + + pub fn window_count(&self) -> usize { + self.windows_set.len() + } + + pub fn len(&self) -> usize { + self.windows_col.len() + } + + pub fn is_empty(&self) -> bool { + self.windows_col.is_empty() + } + + pub fn min_start(&self) -> Option { + self.min_start + } + + pub fn max_end(&self) -> Option { + self.max_end + } + + /// Append-only insert. O(1) amortized. + /// + /// Hot path: three `Vec::push` calls + conditional `windows_set` probe + /// (skipped by Opt 3 on consecutive same-window inserts) + invalidation + /// of `window_to_ids` to `None` (single pointer-width write). + pub fn insert(&mut self, window: TimestampRange, label_id: LabelValuesId, payload: P) { + // Opt 3: monotonic same-window fast path. + if self.last_window != Some(window) { + self.windows_set.insert(window); + self.last_window = Some(window); + } + + self.windows_col.push(window); + self.label_ids_col.push(label_id); + self.payloads_col.push(payload); + + // Opt 1: invalidate the lazy index — single pointer-width write. + self.window_to_ids = None; + + // Update epoch bounds incrementally for O(1) range-skip check. + match self.min_start { + Some(s) if s <= window.0 => {} + _ => self.min_start = Some(window.0), + } + match self.max_end { + Some(e) if e >= window.1 => {} + _ => self.max_end = Some(window.1), + } + } + + /// Build (or rebuild) the lazy `window_to_ids` index from `windows_col` — + /// O(M) one-pass scan, called on the first exact query after any insert. + fn ensure_window_index(&mut self) -> &HashMap> { + if self.window_to_ids.is_none() { + let mut idx = HashMap::with_capacity(self.windows_set.len()); + for (i, w) in self.windows_col.iter().enumerate() { + idx.entry(*w).or_insert_with(Vec::new).push(i as u32); + } + self.window_to_ids = Some(idx); + } + self.window_to_ids.as_ref().unwrap() + } + + /// Return all entries whose window matches `target` exactly. + /// O(M) on first call after a write (builds the index); O(m) cached. + pub fn exact_query(&mut self, target: TimestampRange) -> Vec<(LabelValuesId, &P)> { + let idx = self.ensure_window_index(); + let offsets = match idx.get(&target) { + Some(v) => v.clone(), + None => return Vec::new(), + }; + offsets + .into_iter() + .map(|off| { + let i = off as usize; + (self.label_ids_col[i], &self.payloads_col[i]) + }) + .collect() + } + + /// Range query into a caller-provided buffer. Hot loop touches only + /// `windows_col` (Opt 5) — chase the payload pointer only on a hit. + /// O(M) — linear scan; bounded by the epoch size. + pub fn range_query_into<'a>( + &'a self, + start: u64, + end: u64, + out: &mut Vec<(TimestampRange, LabelValuesId, &'a P)>, + ) { + // O(1) skip if the epoch's bounds don't overlap the query range. + if let (Some(min_s), Some(max_e)) = (self.min_start, self.max_end) { + if min_s > end || max_e < start { + return; + } + } + for (i, w) in self.windows_col.iter().enumerate() { + if w.0 >= start && w.1 <= end { + out.push((*w, self.label_ids_col[i], &self.payloads_col[i])); + } + } + } + + /// Total accumulated entries — caller compares against + /// `epoch_capacity` to decide whether to seal + rotate. + pub fn distinct_windows(&self) -> usize { + self.windows_set.len() + } +} + +impl

Default for MutableEpoch

{ + fn default() -> Self { + Self::new() + } +} + +/// Sealed (immutable) epoch: flat sorted `Vec` for cache-friendly +/// binary-search range scans. Built once at rotation time from the +/// then-active `MutableEpoch`. +pub struct SealedEpoch

{ + /// Sorted by `(TimestampRange, LabelValuesId)`. Binary search on + /// `start_unix_ms` to seek; linear scan within the matched range. + entries: Vec<(TimestampRange, LabelValuesId, P)>, + min_start: Option, + max_end: Option, +} + +impl

SealedEpoch

{ + /// Consume a `MutableEpoch` and produce its sorted immutable form. + /// O(M log M) — paid once at rotation, off the insert hot path. + pub fn from_mutable(mut m: MutableEpoch

) -> Self { + let min_start = m.min_start; + let max_end = m.max_end; + let len = m.windows_col.len(); + let mut entries: Vec<(TimestampRange, LabelValuesId, P)> = Vec::with_capacity(len); + // Drain via swap_remove from the back to move payloads without + // cloning. Equivalent to consuming the parallel arrays in order. + for i in 0..len { + entries.push(( + m.windows_col[i], + m.label_ids_col[i], + std::mem::replace(&mut m.payloads_col[i], unsafe { + std::mem::MaybeUninit::zeroed().assume_init() + }), + )); + } + // Forget the columns to avoid double-drop (the moved-out payloads + // were replaced with zeroed memory; their drop should not run). + std::mem::forget(m.payloads_col); + entries.sort_by(|a, b| a.0.cmp(&b.0).then(a.1.cmp(&b.1))); + Self { entries, min_start, max_end } + } + + pub fn min_start(&self) -> Option { + self.min_start + } + + pub fn max_end(&self) -> Option { + self.max_end + } + + pub fn len(&self) -> usize { + self.entries.len() + } + + pub fn is_empty(&self) -> bool { + self.entries.is_empty() + } + + /// Range query — O(log N + k) via binary search to the first + /// matching entry, then linear scan. + pub fn range_query_into<'a>( + &'a self, + start: u64, + end: u64, + out: &mut Vec<(TimestampRange, LabelValuesId, &'a P)>, + ) { + if let (Some(min_s), Some(max_e)) = (self.min_start, self.max_end) { + if min_s > end || max_e < start { + return; + } + } + // Binary search for the first entry whose start >= start. + let from = self.entries.partition_point(|e| e.0 .0 < start); + for entry in &self.entries[from..] { + if entry.0 .0 > end { + break; + } + if entry.0 .1 <= end { + out.push((entry.0, entry.1, &entry.2)); + } + } + } + + /// Exact-window query — O(log N + m). Binary search for the + /// matching range; linear scan while the range matches. + pub fn exact_query(&self, target: TimestampRange) -> Vec<(LabelValuesId, &P)> { + let from = self.entries.partition_point(|e| e.0 < target); + let mut out = Vec::new(); + for entry in &self.entries[from..] { + if entry.0 != target { + break; + } + out.push((entry.1, &entry.2)); + } + out + } +} + +/// Per-sid storage — drop-in replacement for the new SketchIndex's +/// `series` map's value type. Pairs an active `MutableEpoch` with a +/// rotation-ordered `BTreeMap`. Concurrency is +/// owned by the outer `RwLock` (mirrors legacy's +/// per-key concurrency model). +pub struct SidStoreData { + pub intern: InternTable, + pub current_epoch: MutableEpoch

, + pub sealed_epochs: BTreeMap>, + pub current_epoch_id: EpochId, + pub epoch_capacity: Option, + pub max_epochs: usize, +} + +impl SidStoreData { + pub fn new() -> Self { + Self { + intern: InternTable::new(), + current_epoch: MutableEpoch::new(), + sealed_epochs: BTreeMap::new(), + current_epoch_id: 0, + epoch_capacity: None, + max_epochs: 4, + } + } + + /// Insert a labeled payload for a specific time window. Caller has + /// already canonicalized the label-values key (e.g. sorted). Hot + /// path: amortized O(1) per the optimizations above. + pub fn insert(&mut self, window: TimestampRange, label_key: K, payload: P) { + let label_id = self.intern.intern(label_key); + self.current_epoch.insert(window, label_id, payload); + self.maybe_rotate_epoch(); + } + + fn maybe_rotate_epoch(&mut self) { + let cap = match self.epoch_capacity { + Some(c) if c > 0 => c, + _ => return, + }; + if self.current_epoch.distinct_windows() < cap { + return; + } + // Seal the current epoch and rotate. + let prev_len = self.current_epoch.len(); + let sealed = SealedEpoch::from_mutable(std::mem::replace( + &mut self.current_epoch, + MutableEpoch::with_capacity(prev_len), + )); + self.sealed_epochs.insert(self.current_epoch_id, sealed); + self.current_epoch_id += 1; + + // Drop oldest sealed if we exceed `max_epochs`. + while self.sealed_epochs.len() + 1 > self.max_epochs { + // BTreeMap::pop_first is stable in 1.66+ + if let Some((id, _)) = self + .sealed_epochs + .iter() + .next() + .map(|(k, _)| (*k, ())) + { + self.sealed_epochs.remove(&id); + } else { + break; + } + } + } +} + +impl Default for SidStoreData { + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn intern_idempotent() { + let mut t = InternTable::::new(); + let a = t.intern("foo".into()); + let b = t.intern("foo".into()); + let c = t.intern("bar".into()); + assert_eq!(a, b); + assert_ne!(a, c); + assert_eq!(t.resolve(a), Some(&"foo".to_string())); + assert_eq!(t.resolve(c), Some(&"bar".to_string())); + assert_eq!(t.len(), 2); + } + + #[test] + fn mutable_epoch_insert_and_query() { + let mut e = MutableEpoch::::new(); + e.insert((0, 100), 1, "a".into()); + e.insert((100, 200), 2, "b".into()); + e.insert((100, 200), 3, "c".into()); // Opt 3 fast path: same window + e.insert((200, 300), 4, "d".into()); + assert_eq!(e.distinct_windows(), 3); + assert_eq!(e.len(), 4); + + // Exact query (builds lazy index) + let r = e.exact_query((100, 200)); + let mut ids: Vec<_> = r.iter().map(|(id, _)| *id).collect(); + ids.sort(); + assert_eq!(ids, vec![2, 3]); + + // Range query + let mut buf = Vec::new(); + e.range_query_into(50, 250, &mut buf); + let mut ids2: Vec<_> = buf.iter().map(|(_, id, _)| *id).collect(); + ids2.sort(); + assert_eq!(ids2, vec![2, 3]); // (0,100) is below; (200,300) is above + } + + #[test] + fn sealed_epoch_binary_search_range() { + let mut m = MutableEpoch::::new(); + m.insert((0, 10), 1, 100); + m.insert((10, 20), 2, 200); + m.insert((20, 30), 3, 300); + m.insert((30, 40), 4, 400); + let s = SealedEpoch::from_mutable(m); + let mut buf = Vec::new(); + s.range_query_into(10, 30, &mut buf); + let payloads: Vec = buf.iter().map(|(_, _, p)| **p).collect(); + // (10,20)=200 and (20,30)=300 are fully within [10,30] + assert!(payloads.contains(&200)); + assert!(payloads.contains(&300)); + } + + #[test] + fn sid_store_rotation() { + let mut s = SidStoreData::::new(); + s.epoch_capacity = Some(2); + s.max_epochs = 2; + s.insert((0, 10), "a".into(), "p1".into()); + s.insert((10, 20), "a".into(), "p2".into()); // capacity hit → seal + s.insert((20, 30), "a".into(), "p3".into()); // new epoch + s.insert((30, 40), "a".into(), "p4".into()); // capacity hit → seal again + s.insert((40, 50), "a".into(), "p5".into()); // third epoch; oldest sealed evicted + assert!(s.sealed_epochs.len() <= 2); + } +} diff --git a/asap-query-engine/src/stores/mod.rs b/asap-query-engine/src/stores/mod.rs index 7e3cc430..4928b9be 100644 --- a/asap-query-engine/src/stores/mod.rs +++ b/asap-query-engine/src/stores/mod.rs @@ -16,6 +16,7 @@ //! callers should not care whether it lives under `sketch_db` or //! at the `stores` top level. +pub mod epoch_columnar; pub mod promsketch_store; pub mod sketch_db; pub mod sketch_index; diff --git a/asap-query-engine/src/stores/sketch_index.rs b/asap-query-engine/src/stores/sketch_index.rs index 63868198..cd67d471 100644 --- a/asap-query-engine/src/stores/sketch_index.rs +++ b/asap-query-engine/src/stores/sketch_index.rs @@ -1,17 +1,14 @@ //! Sketch index — Phase 5 of the controller-into-backend refactor (2026-05). //! -//! Two-level index that the SimpleStore migrates to. Replaces the -//! aggregation_id-keyed lookup with a content-addressable design where -//! the index key is the `(raw_metric_name, group_by_keys, capability)` -//! tuple — represented compactly by the centrally-assigned `series_id` -//! (Phase 4) when one is available. -//! -//! Two levels: +//! Two-level index: //! - `instances`: sid → SketchInstanceMetadata (one entry per logical //! sketch instance — its metric name, group-by KEY set, capability, //! sketch_type, sketch_config, accuracy bound). -//! - `series`: sid → Vec (per-series time-windowed -//! sketch state; one entry per distinct group-by VALUES vector). +//! - `series`: sid → per-sid storage (`SidStoreData`) carrying the +//! per-window sketch state. Intern table per sid maps the group-by +//! VALUES vector to a compact `LabelValuesId = u32`; columnar +//! `MutableEpoch` + sealed-epoch ring delivers the legacy +//! SimpleMapStore's six storage optimizations end-to-end. //! //! Ghost sids (registered but never carrying state) are valid — they //! exist when an agent registers a pre-merge identity that the gateway @@ -23,6 +20,11 @@ //! `docs/design-controller-into-backend.md`. use std::collections::{BTreeMap, BTreeSet, HashMap}; +use std::sync::{Arc, RwLock}; + +use dashmap::DashMap; + +use super::epoch_columnar::{LabelValuesId, SidStoreData, TimestampRange}; /// Capability the controller's plan made for this sketch instance. /// Mirrors the design-doc Capability enum (§4.5). One Capability variant @@ -127,7 +129,7 @@ impl AccuracyBound { /// Populated at ingest time when a sketch DataPoint with a fresh sid /// arrives (or `(metric, attrs)` produces a fresh sid via the /// SeriesIdResolver). Subsequent emits of the same sid append to the -/// associated `SketchTimeSeries` without re-touching this metadata. +/// associated `SidStoreData` without re-touching this metadata. #[derive(Debug, Clone)] pub struct SketchInstanceMetadata { pub sid: u64, @@ -142,18 +144,8 @@ pub struct SketchInstanceMetadata { pub first_seen_unix_ms: i64, } -/// Per-series time-windowed sketch state. One `SketchTimeSeries` per -/// distinct group-by VALUES vector under a single sid. -#[derive(Debug, Default)] -pub struct SketchTimeSeries { - pub sid: u64, - /// The group-by VALUES (one value per key in - /// `SketchInstanceMetadata.group_by_keys`). - pub series_label_values: BTreeMap, - /// `window_end_unix_ms → sketch payload bytes + encoding tag`. - pub samples: BTreeMap, -} - +/// Per-sample sketch state. Stored as the payload column inside the +/// per-sid `SidStoreData` columnar storage. #[derive(Debug, Clone)] pub struct SketchSampleState { pub bytes: Vec, @@ -170,19 +162,39 @@ pub enum SketchEncoding { MsgpackDelta, } +/// One materialized series row returned by the query path. Resolved +/// from the per-sid intern table at read time. +#[derive(Debug, Clone)] +pub struct SketchTimeSeries { + pub sid: u64, + pub series_label_values: BTreeMap, + /// `window_end_unix_ms → sketch payload`. BTreeMap so the query + /// path can iterate in time order without an extra sort. + pub samples: BTreeMap, +} + +/// Per-sid storage value — wraps `SidStoreData` in an `RwLock` so the +/// outer DashMap stays read-mostly and per-sid writes don't block one +/// another. +type SidStore = Arc, SketchSampleState>>>; + /// Two-level sketch index. Replaces the legacy `aggregation_id`-keyed /// SimpleStore lookup once Phase 5 wiring lands at the streaming engine /// ingest path and the query path. -#[derive(Debug, Default)] +/// +/// `instances` is keyed under a `RwLock` because the registration +/// rate is low (one write per first-seen sid) and reads dominate; +/// `series` is a `DashMap` because per-sid writes happen on every DP. +#[derive(Default)] pub struct SketchIndex { /// sid → metadata. May contain ghost sids (registered identities /// whose state was merged away by an upstream gateway before /// reaching this backend). - pub instances: HashMap, - /// sid → per-series time-windowed state. Empty `Vec` (or absent + instances: RwLock>, + /// sid → per-sid columnar storage. Empty `SidStoreData` (or absent /// key) for ghost sids — query path detects this and falls through /// to Thanos archive. - pub series: HashMap>, + series: DashMap, } /// Three possible outcomes of looking up a sid in the SketchIndex. @@ -205,47 +217,125 @@ impl SketchIndex { Self::default() } + /// Classify a sid for query routing. See `SidLookup` for semantics. pub fn classify(&self, sid: u64) -> SidLookup { - match (self.instances.get(&sid), self.series.get(&sid)) { - (Some(_), Some(series)) if !series.is_empty() => SidLookup::Hit, - (Some(_), _) => SidLookup::Ghost, - (None, _) => SidLookup::Unknown, + let known = self.instances.read().unwrap().contains_key(&sid); + if !known { + return SidLookup::Unknown; + } + match self.series.get(&sid) { + Some(store) => { + let g = store.read().unwrap(); + if !g.current_epoch.is_empty() || !g.sealed_epochs.is_empty() { + SidLookup::Hit + } else { + SidLookup::Ghost + } + } + None => SidLookup::Ghost, } } /// Insert metadata for a freshly-resolved sid. - pub fn register(&mut self, meta: SketchInstanceMetadata) { - self.instances.insert(meta.sid, meta); + pub fn register(&self, meta: SketchInstanceMetadata) { + self.instances.write().unwrap().insert(meta.sid, meta); + } + + /// Look up the metadata for a sid (cloned because callers usually + /// release the index lock before working with it). + pub fn instance(&self, sid: u64) -> Option { + self.instances.read().unwrap().get(&sid).cloned() } /// Append a window's sketch state under `sid`. Caller is responsible /// for ensuring the corresponding `SketchInstanceMetadata` was /// registered (or the sketch arrives orphan and the caller chooses /// to drop / reject / register-on-the-fly). + /// + /// `window` is the OTLP DataPoint's `(start_time_unix_ms, time_unix_ms)`. pub fn append_sample( - &mut self, + &self, sid: u64, series_label_values: BTreeMap, - window_end_unix_ms: i64, + window: TimestampRange, sample: SketchSampleState, ) { - let series_vec = self.series.entry(sid).or_default(); - // Find or create the SketchTimeSeries for this label-values vector. - let ts = match series_vec - .iter_mut() - .find(|s| s.series_label_values == series_label_values) - { - Some(s) => s, - None => { - series_vec.push(SketchTimeSeries { - sid, - series_label_values, - samples: BTreeMap::new(), - }); - series_vec.last_mut().unwrap() - } + let store = self + .series + .entry(sid) + .or_insert_with(|| Arc::new(RwLock::new(SidStoreData::new()))) + .clone(); + let mut guard = store.write().unwrap(); + guard.insert(window, series_label_values, sample); + } + + /// Range-query the warm-tier state for one sid. Window-end-keyed + /// time series result, one entry per distinct group-by VALUES + /// vector. `(start, end)` is the inclusive query window; entries + /// whose `(window_start, window_end)` lies fully within the query + /// range are returned. + pub fn query_range( + &self, + sid: u64, + start_unix_ms: u64, + end_unix_ms: u64, + ) -> Vec { + let store = match self.series.get(&sid) { + Some(s) => s.clone(), + None => return Vec::new(), }; - ts.samples.insert(window_end_unix_ms, sample); + let guard = store.write().unwrap(); // exact_query may build the lazy index + let mut by_label_id: HashMap> = + HashMap::new(); + + let mut buf: Vec<(TimestampRange, LabelValuesId, &SketchSampleState)> = Vec::new(); + guard + .current_epoch + .range_query_into(start_unix_ms, end_unix_ms, &mut buf); + for (win, label_id, payload) in &buf { + by_label_id + .entry(*label_id) + .or_default() + .insert(win.1 as i64, (*payload).clone()); + } + buf.clear(); + + for sealed in guard.sealed_epochs.values() { + sealed.range_query_into(start_unix_ms, end_unix_ms, &mut buf); + for (win, label_id, payload) in &buf { + by_label_id + .entry(*label_id) + .or_default() + .insert(win.1 as i64, (*payload).clone()); + } + buf.clear(); + } + + by_label_id + .into_iter() + .map(|(label_id, samples)| { + let label_values = guard + .intern + .resolve(label_id) + .cloned() + .unwrap_or_default(); + SketchTimeSeries { + sid, + series_label_values: label_values, + samples, + } + }) + .collect() + } + + /// Number of distinct sids carrying state (excludes ghosts). + pub fn series_len(&self) -> usize { + self.series.len() + } + + /// Number of registered instances (includes ghosts). + pub fn instance_count(&self) -> usize { + self.instances.read().unwrap().len() } } @@ -253,52 +343,86 @@ impl SketchIndex { mod tests { use super::*; - #[test] - fn ghost_classification() { - let mut idx = SketchIndex::new(); - let meta = SketchInstanceMetadata { - sid: 42, + fn meta(sid: u64) -> SketchInstanceMetadata { + let cfg = SketchConfig::DDSketch { relative_accuracy: 0.01 }; + SketchInstanceMetadata { + sid, metric_name: "m".into(), group_by_keys: BTreeSet::new(), capability: Capability::QuantileApprox(SketchKindHandle::DDSketch), sketch_kind: SketchKindHandle::DDSketch, - sketch_config: SketchConfig::DDSketch { relative_accuracy: 0.01 }, - accuracy: AccuracyBound::from_config(&SketchConfig::DDSketch { - relative_accuracy: 0.01, - }), + sketch_config: cfg.clone(), + accuracy: AccuracyBound::from_config(&cfg), first_seen_unix_ms: 0, - }; - idx.register(meta); - // Metadata exists but no series state — ghost. + } + } + + fn sample(b: u8) -> SketchSampleState { + SketchSampleState { bytes: vec![b], encoding: SketchEncoding::ProtoFull } + } + + #[test] + fn ghost_classification() { + let idx = SketchIndex::new(); + idx.register(meta(42)); assert_eq!(idx.classify(42), SidLookup::Ghost); - // Unregistered sid — unknown. assert_eq!(idx.classify(999), SidLookup::Unknown); } #[test] fn hit_after_append() { - let mut idx = SketchIndex::new(); - let cfg = SketchConfig::Hll { precision: 14 }; - let meta = SketchInstanceMetadata { - sid: 7, - metric_name: "m".into(), - group_by_keys: BTreeSet::new(), - capability: Capability::CardinalityApprox, - sketch_kind: SketchKindHandle::Hll, - sketch_config: cfg.clone(), - accuracy: AccuracyBound::from_config(&cfg), - first_seen_unix_ms: 0, - }; - idx.register(meta); - idx.append_sample( - 7, - BTreeMap::new(), - 1000, - SketchSampleState { bytes: vec![1, 2, 3], encoding: SketchEncoding::ProtoFull }, - ); + let idx = SketchIndex::new(); + idx.register(meta(7)); + idx.append_sample(7, BTreeMap::new(), (1000, 1010), sample(1)); assert_eq!(idx.classify(7), SidLookup::Hit); } + #[test] + fn range_query_returns_distinct_series() { + let idx = SketchIndex::new(); + idx.register(meta(11)); + let mut lv_a = BTreeMap::new(); + lv_a.insert("host".to_string(), "a".to_string()); + let mut lv_b = BTreeMap::new(); + lv_b.insert("host".to_string(), "b".to_string()); + + idx.append_sample(11, lv_a.clone(), (0, 10), sample(1)); + idx.append_sample(11, lv_a.clone(), (10, 20), sample(2)); + idx.append_sample(11, lv_b.clone(), (10, 20), sample(3)); + idx.append_sample(11, lv_b.clone(), (20, 30), sample(4)); + + let mut series = idx.query_range(11, 0, 30); + series.sort_by(|x, y| x.series_label_values.cmp(&y.series_label_values)); + assert_eq!(series.len(), 2); + + let s_a = &series[0]; + assert_eq!(s_a.series_label_values, lv_a); + assert_eq!(s_a.samples.len(), 2); + assert_eq!(s_a.samples[&10].bytes, vec![1]); + assert_eq!(s_a.samples[&20].bytes, vec![2]); + + let s_b = &series[1]; + assert_eq!(s_b.series_label_values, lv_b); + assert_eq!(s_b.samples.len(), 2); + } + + #[test] + fn range_query_clips_to_window_bounds() { + let idx = SketchIndex::new(); + idx.register(meta(13)); + let lv = BTreeMap::new(); + idx.append_sample(13, lv.clone(), (0, 10), sample(1)); + idx.append_sample(13, lv.clone(), (10, 20), sample(2)); + idx.append_sample(13, lv.clone(), (20, 30), sample(3)); + + // Only the middle window is fully within [5, 25]. + let series = idx.query_range(13, 5, 25); + assert_eq!(series.len(), 1); + let s = &series[0]; + assert_eq!(s.samples.len(), 1); + assert!(s.samples.contains_key(&20)); + } + #[test] fn ddsketch_accuracy_bound() { let bound = AccuracyBound::from_config(&SketchConfig::DDSketch { @@ -307,4 +431,30 @@ mod tests { assert!((bound.epsilon - 0.01).abs() < 1e-9); assert!((bound.confidence - 1.0).abs() < 1e-9); } + + #[test] + fn epoch_rotation_is_visible_to_query() { + let idx = SketchIndex::new(); + idx.register(meta(17)); + + // Force aggressive rotation by touching the SidStoreData + // capacity *after* the entry is created. We do this by + // first appending one sample to materialize the entry, then + // mutating its config, then appending more. + idx.append_sample(17, BTreeMap::new(), (0, 10), sample(1)); + if let Some(s) = idx.series.get(&17) { + let mut g = s.write().unwrap(); + g.epoch_capacity = Some(2); + g.max_epochs = 4; + } + idx.append_sample(17, BTreeMap::new(), (10, 20), sample(2)); + idx.append_sample(17, BTreeMap::new(), (20, 30), sample(3)); + idx.append_sample(17, BTreeMap::new(), (30, 40), sample(4)); + + // All four windows should still be query-visible across the + // mutable + sealed boundary. + let series = idx.query_range(17, 0, 40); + assert_eq!(series.len(), 1); + assert_eq!(series[0].samples.len(), 4); + } }