feat(sketch_db): SchemaEvictionService removes SketchIndex sids too

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,

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<String> =
+            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<u64> = {
+            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

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<SchemaRegistry>,
     backfill: Arc<BackfillRegistry>,
     store: Arc<dyn Store>,
+    /// 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<Arc<SketchIndex>>,
     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<SketchIndex>) -> 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]);