From 785ba58a8fac2c62950110e439ebcc429652e967 Mon Sep 17 00:00:00 2001 From: zz_y Date: Wed, 13 May 2026 21:29:10 -0600 Subject: [PATCH] feat(sid): wire find_matching_policies + sids_for_policy into ASAPQueryEngine.execute MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Switches the query engine's candidate → sid resolution to the content-addressed fast path. PRs #203, #204, #205 landed the primitives; this PR integrates them. ## Before vs. after Before, per candidate: 1. `idx.instances_matching(metric, gbk)` walks the `RwLock>` and returns every sid whose metric + group-by-keys match. 2. Per-sid: classify, check capability satisfaction, push to `hit_sids`. After, per candidate: 1. Snapshot the streaming config; build the `PolicyRegistry`. 2. `find_matching_policies(registry, candidate)` — walks the small policy registry (≤ thousands of entries) with the full match predicate (metric + group_by + capability + window + filter). Returns `Vec`. 3. For each fp: `idx.sids_for_policy(fp)` — O(1) hash lookup over the reverse index added in PR #203. 4. Same per-sid classify + capability check as before (defensive; fast-path-discovered sids already satisfy the candidate by construction, but UNSET sids reached via the fallback don't). ## Slow-path fallback retained When the fast path yields zero sids — because either: - no policy in the registry matches the candidate (control plane hasn't published one yet), OR - the candidate's sids were registered with `PolicyFingerprint::UNSET` (legacy paths that didn't carry an `AggregationConfig` at ingest, test fixtures, raw mode) — the engine falls back to the metadata walk `instances_matching(metric, gbk)`. The per-sid capability filter below catches mismatches the fast path would have rejected at policy-match time. The fallback can be deleted in a follow-up once every code path populates `policy_fp` and existing on-disk records have aged out. ## Snapshot semantics The streaming-config snapshot is pinned once per query (not per candidate). Hot-reload swaps the underlying `Arc` mid-query are isolated by the snapshot: the query sees the policy set that was active at query start. Same isolation the legacy `streaming_config_snapshot()` call already gave the engine elsewhere. ## Test plan - [x] `cargo check --workspace` clean - [x] `cargo test --workspace --lib --bins` green (`asap_types::capability_matching::tests::avg_finds_sum_and_count` HashMap-iteration flake unchanged) - Existing query-engine tests cover the slow-path fallback (their fixtures register sids with UNSET fp); future tests on the fast path land alongside an end-to-end test that builds a streaming config with matching policies and verifies the fp-keyed lookup is used. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.7 (1M context) --- .../query_engines/asap_query_engine/engine.rs | 34 ++++++++++++++++++- 1 file changed, 33 insertions(+), 1 deletion(-) diff --git a/data_plane/src/query_engines/asap_query_engine/engine.rs b/data_plane/src/query_engines/asap_query_engine/engine.rs index 6d2daf9f..0bb57ce3 100644 --- a/data_plane/src/query_engines/asap_query_engine/engine.rs +++ b/data_plane/src/query_engines/asap_query_engine/engine.rs @@ -3517,8 +3517,40 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu None; let mut combined_t0: u64 = u64::MAX; + // Snapshot the streaming config once for this query's + // policy lookups. Hot-reload swaps the underlying Arc; the + // snapshot pins one revision for the duration. + let streaming_snap = self.streaming_config_snapshot(); + let policy_registry = streaming_snap.policy_registry(); + for candidate in &analysis.candidates { - let sids = idx.instances_matching(&candidate.metric_name, &candidate.group_by_keys); + // Fast path (PRs #203 + #204 + #205): find matching + // policies in the content-addressed registry, then + // resolve each policy_fp → {sids} via the reverse + // index. Both hops are O(1)-amortized. + // + // Slow-path fallback: when the fast path yields no + // sids — either because no policy matches (control + // plane hasn't published one yet) or because the + // candidate's sids were registered with + // `PolicyFingerprint::UNSET` (legacy paths that + // didn't carry an `AggregationConfig` at ingest) — + // fall back to the metadata walk + // `instances_matching(metric, gbk)`. The per-sid + // capability filter below catches mismatches the + // fast path would have rejected at policy-match time. + let policy_fps = control_plane::warm_tier_analysis::find_matching_policies( + &policy_registry, + candidate, + ); + let mut sids: Vec = Vec::new(); + for fp in &policy_fps { + sids.extend(idx.sids_for_policy(*fp)); + } + if sids.is_empty() { + sids = idx + .instances_matching(&candidate.metric_name, &candidate.group_by_keys); + } if sids.is_empty() { return Err(crate::query_engines::EngineError::capability_miss( asap_types::StorageBackend::SketchStore.data_source_id(),