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 5e144368..1f18bdf2 100644 --- a/data_plane/src/query_engines/asap_query_engine/engine.rs +++ b/data_plane/src/query_engines/asap_query_engine/engine.rs @@ -1044,101 +1044,51 @@ impl ASAPQueryEngine { params.is_exact_query ); - // M2.3.5b — when a `SketchIndex` is attached AND the agg - // config is resolvable, read precomputes from SketchIndex - // (sid-keyed) instead of the legacy SketchStore. DualWriteSink - // mirrors writes to both, so the data is identical; this just - // moves the read off the agg_id-keyed store. When sketch_index - // is `None` (tests without one) or the agg_cfg is missing - // (eviction race), fall back to the legacy store. - if let Some(idx) = self.sketch_index.as_ref() { - let cfg = self.streaming_config_snapshot(); - if let Some(agg_cfg) = cfg.get_aggregation_config(params.aggregation_id) { - let raw = idx.query_precomputes_by_agg( - ¶ms.metric, - agg_cfg.aggregation_type, - params.start_timestamp, - params.end_timestamp, - ); - let result: TimestampedBucketsMap = if params.is_exact_query { - // Sliding-window mode requires bit-exact (start, - // end) match. SketchIndex's range query returns - // any windows fully within [start, end] — filter - // post-hoc to recover the exact semantics the - // legacy `query_precomputed_output_exact` had. - raw.into_iter() - .map(|(k, v)| { - let filtered: Vec<_> = v - .into_iter() - .filter(|((s, e), _)| { - *s == params.start_timestamp - && *e == params.end_timestamp - }) - .collect(); - (k, filtered) + // M2.3.6f — engine reads precomputes from SketchIndex only. + // The legacy `Store::query_precomputed_output*` fallback has + // been retired now that DualWriteSink (M2.3.4b) → SketchIndexSink + // (M2.3.6a) writes exclusively to SketchIndex and + // BackfillService (M2.3.6e) mirrors replays there too. + // + // Tests that don't attach a SketchIndex now get `Ok(empty)` + // here. Anything deeper than smoke-test coverage was already + // setting one (M2.3.5b made it mandatory in production). + let Some(idx) = self.sketch_index.as_ref() else { + return Ok(TimestampedBucketsMap::new()); + }; + let cfg = self.streaming_config_snapshot(); + let Some(agg_cfg) = cfg.get_aggregation_config(params.aggregation_id) else { + return Ok(TimestampedBucketsMap::new()); + }; + let raw = idx.query_precomputes_by_agg( + ¶ms.metric, + agg_cfg.aggregation_type, + params.start_timestamp, + params.end_timestamp, + ); + let result: TimestampedBucketsMap = if params.is_exact_query { + // Sliding-window mode requires bit-exact (start, end) + // match. SketchIndex's range query returns any windows + // fully within [start, end] — filter post-hoc to recover + // the exact semantics the retired + // `query_precomputed_output_exact` had. + raw.into_iter() + .map(|(k, v)| { + let filtered: Vec<_> = v + .into_iter() + .filter(|((s, e), _)| { + *s == params.start_timestamp + && *e == params.end_timestamp }) - .filter(|(_, v)| !v.is_empty()) - .collect() - } else { - raw - }; - return Ok(result); - } - } - - let store_query_start_time = Instant::now(); - - let result = if params.is_exact_query { - debug!( - "Sliding window query: Looking for exact window [{}, {}]", - params.start_timestamp, params.end_timestamp - ); - let res = self.store.query_precomputed_output_exact( - ¶ms.metric, - params.aggregation_id, - params.start_timestamp, - params.end_timestamp, - ); - if let Ok(ref outputs) = res { - let store_query_duration = store_query_start_time.elapsed(); - debug!( - "Sliding window exact query took: {:.2}ms, found {} unique keys", - store_query_duration.as_secs_f64() * 1000.0, - outputs.len() - ); - } - res + .collect(); + (k, filtered) + }) + .filter(|(_, v)| !v.is_empty()) + .collect() } else { - debug!( - "Tumbling window query: range [{}, {}]", - params.start_timestamp, params.end_timestamp - ); - let res = self.store.query_precomputed_output( - ¶ms.metric, - params.aggregation_id, - params.start_timestamp, - params.end_timestamp, - ); - if res.is_ok() { - let store_query_duration = store_query_start_time.elapsed(); - debug!( - "Tumbling window range query took: {:.2}ms", - store_query_duration.as_secs_f64() * 1000.0 - ); - } - res + raw }; - - result.map_err(|e| { - format!( - "Error querying store for metric {}, agg {}, range [{}, {}]: {}", - params.metric, - params.aggregation_id, - params.start_timestamp, - params.end_timestamp, - e - ) - }) + Ok(result) } /// Executes the full store query plan and returns merged results diff --git a/data_plane/src/tests/schema_timeline_dispatch_tests.rs b/data_plane/src/tests/schema_timeline_dispatch_tests.rs index 28f6945b..cd5e88b1 100644 --- a/data_plane/src/tests/schema_timeline_dispatch_tests.rs +++ b/data_plane/src/tests/schema_timeline_dispatch_tests.rs @@ -98,23 +98,35 @@ fn build_engine( streaming_config: Arc, schemas: Arc, store: Arc, + sketch_index: Arc, _query_for_agg_id: u64, ) -> ASAPQueryEngine { let hot_reload = HotReloadStreamingConfig::from_arc(streaming_config); ASAPQueryEngine::new_with_hot_reload(store, hot_reload, 1) .with_schema_registry(schemas) + .with_sketch_index(sketch_index) } /// Insert a single `SumAccumulator` window at `ts` into `agg_id`. -/// Uses `(ts, ts)` for the window start/end pair to match the -/// existing test-utility pattern in `engine_factories` — the engine -/// treats those as single-point buckets aligned to the tumbling -/// window, so a query whose range contains `ts` picks up the data. -fn seed_sum_at(store: &SketchStore, agg_id: u64, ts: u64, host: &str, sum: f64) { +/// Mirrors the live ingest path's M2.3.6e write-path: data lands in +/// BOTH the legacy SketchStore AND the new SketchIndex so the +/// engine's M2.3.6f read path (SketchIndex-only) sees it. +fn seed_sum_at( + store: &SketchStore, + sketch_index: &crate::stores::sketch_db::index::SketchIndex, + streaming_config: &StreamingConfig, + agg_id: u64, + ts: u64, + host: &str, + sum: f64, +) { let key = Some(KeyByLabelValues { labels: vec![host.to_string()]}); let output = PrecomputedOutput::new(ts, ts, key, agg_id); let acc = SumAccumulator::with_sum(sum); + if let Some(agg_cfg) = streaming_config.get_aggregation_config(agg_id) { + sketch_index.ingest_precompute_for_agg_config(agg_cfg, &output, &acc); + } store .insert_precomputed_output(output, Box::new(acc)) .expect("seed insert must succeed"); @@ -148,10 +160,11 @@ fn sum_query_across_reconfigure_boundary_returns_combined_full_result() { // per segment: // agg_1's sub-range is `[QUERY_START_MS, BOUNDARY_MS]` // agg_2's sub-range is `[BOUNDARY_MS, QUERY_TIME_MS]` - seed_sum_at(&store, 1, AGG1_SAMPLE_MS, "A", 10.0); - seed_sum_at(&store, 2, AGG2_SAMPLE_MS, "A", 20.0); + let sketch_index = Arc::new(crate::stores::sketch_db::index::SketchIndex::new()); + seed_sum_at(&store, &sketch_index, &streaming_config, 1, AGG1_SAMPLE_MS, "A", 10.0); + seed_sum_at(&store, &sketch_index, &streaming_config, 2, AGG2_SAMPLE_MS, "A", 20.0); - let engine = build_engine(streaming_config, schemas, store, 2); + let engine = build_engine(streaming_config, schemas, store, sketch_index, 2); let (_labels, qr) = engine .handle_query_promql(TEST_QUERY.to_string(), QUERY_TIME_SEC) @@ -201,9 +214,10 @@ fn sum_query_with_purged_segment_returns_partial_with_warnings() { )); // Only agg_2 has data; agg_1's data is assumed gone with the // Purged classification. - seed_sum_at(&store, 2, AGG2_SAMPLE_MS, "A", 20.0); + let sketch_index = Arc::new(crate::stores::sketch_db::index::SketchIndex::new()); + seed_sum_at(&store, &sketch_index, &streaming_config, 2, AGG2_SAMPLE_MS, "A", 20.0); - let engine = build_engine(streaming_config, schemas, store, 2); + let engine = build_engine(streaming_config, schemas, store, sketch_index, 2); let (_labels, qr) = engine .handle_query_promql(TEST_QUERY.to_string(), QUERY_TIME_SEC) @@ -251,9 +265,10 @@ fn single_schema_query_falls_through_to_default_path() { streaming_config.clone(), CleanupPolicy::NoCleanup, )); - seed_sum_at(&store, 7, QUERY_TIME_MS, "A", 42.0); + let sketch_index = Arc::new(crate::stores::sketch_db::index::SketchIndex::new()); + seed_sum_at(&store, &sketch_index, &streaming_config, 7, QUERY_TIME_MS, "A", 42.0); - let engine = build_engine(streaming_config, schemas, store, 7); + let engine = build_engine(streaming_config, schemas, store, sketch_index, 7); let (_labels, qr) = engine .handle_query_promql(TEST_QUERY.to_string(), QUERY_TIME_SEC)