diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index eb1e05f2..dc5fbea2 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -838,7 +838,9 @@ async fn main() -> Result<()> { poll_interval: std::time::Duration::from_secs(args.schema_eviction_poll_secs), dry_run: args.schema_eviction_dry_run, }, - ); + ) + // M2.3.6d — eviction sweeps SketchIndex too. + .with_sketch_index(sketch_index.clone()); info!( poll_secs = args.schema_eviction_poll_secs, dry_run = args.schema_eviction_dry_run, diff --git a/data_plane/src/stores/sketch_db/index/mod.rs b/data_plane/src/stores/sketch_db/index/mod.rs index efb1fb04..a943319a 100644 --- a/data_plane/src/stores/sketch_db/index/mod.rs +++ b/data_plane/src/stores/sketch_db/index/mod.rs @@ -902,6 +902,54 @@ impl SketchIndex { } } +impl SketchIndex { + /// Phase 5 M2.3.6d — eviction-side helper. Removes every sid in the + /// index whose metadata was registered against `agg_cfg`, i.e. + /// shares the same metric, agg_type, parameters canonicalization, + /// and grouping-keys set the `SketchIndexSink` used at write time. + /// Returns how many sids were removed. Used by + /// `SchemaEvictionService` to drop a retired schema's residual sid + /// state. + pub fn remove_instances_for_agg_config( + &self, + agg_cfg: &asap_types::aggregation_config::AggregationConfig, + ) -> usize { + let target_metric = agg_cfg.metric.as_str(); + let target_agg_type = agg_cfg.aggregation_type; + let target_params = canonical_parameters(&agg_cfg.parameters); + let target_group_keys: BTreeSet = + agg_cfg.grouping_labels.labels.iter().cloned().collect(); + + // Collect the matching sids under a short read lock; then call + // `remove_instance` per sid (which takes its own write lock). + let to_remove: Vec = { + let g = self.instances.read().unwrap(); + g.iter() + .filter(|(_, m)| { + if m.metric_name != target_metric { + return false; + } + if m.group_by_keys != target_group_keys { + return false; + } + matches!( + &m.agg_kind, + AggKind::Precompute { agg_type, parameters_canonical } + if *agg_type == target_agg_type + && parameters_canonical == &target_params + ) + }) + .map(|(sid, _)| *sid) + .collect() + }; + let count = to_remove.len(); + for sid in to_remove { + self.remove_instance(sid); + } + count + } +} + /// Persistence harness for `SketchIndex` — Phase 5 M2.3.6c. /// /// Owns the manifest + flusher thread + part cache that back the diff --git a/data_plane/src/stores/sketch_db/schema/eviction.rs b/data_plane/src/stores/sketch_db/schema/eviction.rs index 52160807..e6dbbe56 100644 --- a/data_plane/src/stores/sketch_db/schema/eviction.rs +++ b/data_plane/src/stores/sketch_db/schema/eviction.rs @@ -53,6 +53,7 @@ use tokio::task::JoinHandle; use tracing::{info, warn}; use crate::stores::sketch_db::backfill::{BackfillRegistry, BackfillStatus}; +use crate::stores::sketch_db::index::SketchIndex; use super::{AggStatus, SchemaRegistry}; use crate::stores::traits::Store; @@ -83,6 +84,11 @@ pub struct SchemaEvictionService { schemas: Arc, backfill: Arc, store: Arc, + /// Phase 5 M2.3.6d — when set, the eviction service ALSO removes + /// the schema's residual sid state from the sketch index after + /// dropping data on the legacy store. Optional so test fixtures + /// that pre-date M2.3 stay compiling without rewiring. + sketch_index: Option>, config: SchemaEvictionConfig, } @@ -97,10 +103,19 @@ impl SchemaEvictionService { schemas, backfill, store, + sketch_index: None, config, } } + /// Attach a `SketchIndex` so the eviction sweep also removes the + /// schema's per-sid state. Returns `self` (builder-style) so + /// existing call sites can opt in with a single chained call. + pub fn with_sketch_index(mut self, index: Arc) -> Self { + self.sketch_index = Some(index); + self + } + /// Spawn as a tokio task. Returns a handle whose `shutdown` /// oneshot stops the loop cleanly on ctrl-c. pub fn spawn(self) -> SchemaEvictionHandle { @@ -209,6 +224,22 @@ impl SchemaEvictionService { } } + // Phase 5 M2.3.6d — remove the schema's residual sid state + // from the sketch index. Best-effort: a 0 count here is + // normal (nothing was ever ingested under that schema, or + // already swept by a prior tick). + if let Some(idx) = self.sketch_index.as_ref() { + let removed = idx.remove_instances_for_agg_config(&schema.config); + if removed > 0 { + info!( + agg_id, + %metric, + sids_removed = removed, + "SchemaEviction: also dropped sids in SketchIndex" + ); + } + } + // Step 3: remove the schema record. self.schemas.remove_schema(agg_id); } @@ -384,6 +415,65 @@ mod tests { assert!(schemas.get(2).is_some()); } + #[tokio::test(flavor = "current_thread")] + async fn run_once_also_removes_sketch_index_instances() { + use crate::stores::sketch_db::index::{ + canonical_parameters, compute_sid, AggKind, SketchIndex, + SketchInstanceMetadata, + }; + use std::collections::BTreeSet; + + let (schemas, backfill, store) = fixture_with_expired_1().await; + let sketch_index = Arc::new(SketchIndex::new()); + + // Register a precompute sid that matches the soon-to-expire + // agg config (agg_id=1, metric_1, Sum, no grouping). The + // eviction sweep should remove it. + let agg_cfg = sum_agg_config(1); + let sid = compute_sid( + "metric_1", + "", + &AggKind::Precompute { + agg_type: agg_cfg.aggregation_type, + parameters_canonical: canonical_parameters(&agg_cfg.parameters), + }, + ); + sketch_index.register(SketchInstanceMetadata { + sid, + metric_name: "metric_1".into(), + group_by_keys: BTreeSet::new(), + capability: None, + agg_kind: AggKind::Precompute { + agg_type: agg_cfg.aggregation_type, + parameters_canonical: canonical_parameters(&agg_cfg.parameters), + }, + accuracy: None, + first_seen_unix_ms: 0, + retired_at_ms: None, + expires_at_ms: None, + }); + assert_eq!(sketch_index.instance_count(), 1); + + let svc = SchemaEvictionService::new( + schemas.clone(), + backfill, + store.clone(), + SchemaEvictionConfig { + poll_interval: Duration::from_secs(60), + dry_run: false, + }, + ) + .with_sketch_index(sketch_index.clone()); + + svc.run_once(); + + assert_eq!( + sketch_index.instance_count(), + 0, + "expired agg_config's sids must be removed from SketchIndex" + ); + } + #[tokio::test(flavor = "current_thread")] async fn run_once_is_noop_with_no_expired_schemas() { let initial = make_streaming_config(&[1]);