From 4ee3dd8d067c9d816888da98a581a769a0bcc223 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Sun, 28 Jun 2026 16:48:46 -0400 Subject: [PATCH] fix(planner): set range_a to scrape_interval for OnlySpatial queries --- .../src/optimizer/aqe_extractor.rs | 31 ++++++++++---- .../src/optimizer/candidate_gen.rs | 40 +++++++------------ asap-planner-rs/src/optimizer/pipeline.rs | 25 ++++++++++-- 3 files changed, 60 insertions(+), 36 deletions(-) diff --git a/asap-planner-rs/src/optimizer/aqe_extractor.rs b/asap-planner-rs/src/optimizer/aqe_extractor.rs index 8d714e9..9078ef7 100644 --- a/asap-planner-rs/src/optimizer/aqe_extractor.rs +++ b/asap-planner-rs/src/optimizer/aqe_extractor.rs @@ -60,7 +60,11 @@ impl AQEKey { /// /// Leaf queries that do not match any supported pattern (e.g. unsupported /// functions, parse errors) are skipped with a warning. -pub fn extract_aqes(rqes: &[RQE], metric_schema: &PromQLSchema) -> Vec { +pub fn extract_aqes( + rqes: &[RQE], + metric_schema: &PromQLSchema, + scrape_interval_ms: u64, +) -> Vec { // (key) -> (requirements, query_strings, sum_freq, min_t, gcd_t) let mut acc: HashMap, f64, u64, u64)> = HashMap::new(); @@ -78,7 +82,12 @@ pub fn extract_aqes(rqes: &[RQE], metric_schema: &PromQLSchema) -> Vec { for leaf in leaves { match extract_requirements(&leaf, metric_schema) { - Some(req) => { + Some(mut req) => { + // Spatial-only queries have no range vector; treat them as a + // single scrape interval so downstream window logic is explicit. + if req.data_range_ms.is_none() { + req.data_range_ms = Some(scrape_interval_ms); + } let key = AQEKey::from_requirements(&req); let entry = acc .entry(key) @@ -192,7 +201,7 @@ mod tests { #[test] fn single_temporal_query() { let rqes = vec![rqe("sum_over_time(metric[5m])", 60_000)]; - let aqes = extract_aqes(&rqes, &empty_schema()); + let aqes = extract_aqes(&rqes, &empty_schema(), 15_000); assert_eq!(aqes.len(), 1); assert!((aqes[0].query_frequency_hz - 1.0 / 60.0).abs() < 1e-9); assert_eq!(aqes[0].requirements.metric, "metric"); @@ -205,14 +214,14 @@ mod tests { "sum_over_time(metric_a[5m]) / sum_over_time(metric_b[5m])", 60_000, )]; - let aqes = extract_aqes(&rqes, &empty_schema()); + let aqes = extract_aqes(&rqes, &empty_schema(), 15_000); assert_eq!(aqes.len(), 2); } #[test] fn binary_with_scalar_produces_one_aqe() { let rqes = vec![rqe("sum_over_time(metric[5m]) * 100", 60_000)]; - let aqes = extract_aqes(&rqes, &empty_schema()); + let aqes = extract_aqes(&rqes, &empty_schema(), 15_000); assert_eq!(aqes.len(), 1); } @@ -222,7 +231,7 @@ mod tests { rqe("sum_over_time(metric[5m])", 60_000), rqe("sum_over_time(metric[5m])", 30_000), ]; - let aqes = extract_aqes(&rqes, &empty_schema()); + let aqes = extract_aqes(&rqes, &empty_schema(), 15_000); assert_eq!(aqes.len(), 1); // query_frequency_hz = sum of rates (total query load for the MIP objective) let expected_freq = 1.0 / 60.0 + 1.0 / 30.0; @@ -236,7 +245,15 @@ mod tests { #[test] fn unsupported_query_is_skipped() { let rqes = vec![rqe("not_a_real_function(metric[5m])", 60_000)]; - let aqes = extract_aqes(&rqes, &empty_schema()); + let aqes = extract_aqes(&rqes, &empty_schema(), 15_000); assert_eq!(aqes.len(), 0); } + + #[test] + fn spatial_only_query_sets_range_to_scrape_interval() { + let rqes = vec![rqe("sum(metric)", 60_000)]; + let aqes = extract_aqes(&rqes, &empty_schema(), 15_000); + assert_eq!(aqes.len(), 1); + assert_eq!(aqes[0].requirements.data_range_ms, Some(15_000)); + } } diff --git a/asap-planner-rs/src/optimizer/candidate_gen.rs b/asap-planner-rs/src/optimizer/candidate_gen.rs index ba87886..7b8a0fe 100644 --- a/asap-planner-rs/src/optimizer/candidate_gen.rs +++ b/asap-planner-rs/src/optimizer/candidate_gen.rs @@ -41,7 +41,7 @@ pub fn enumerate_candidates(aqe: &AQE, scrape_interval_ms: u64) -> Vec CandidateConfig { /// Window candidates: (WindowType, W_ms, slide_interval_ms, n_windows). /// -/// For spatial-only queries (range = None): one tumbling window of scrape_interval. -/// For range-based queries: -/// Tumbling: all W that divide range_a, are multiples of scrape_interval, and ≤ min_t_repeat. -/// Slide interval = W (a tumbling window "slides" by its own width). -/// Sliding: W = range_a exactly (overlapping sliding windows can't merge/subtract to cover -/// a larger range — only mutually-exclusive tumbling windows compose). The slide -/// interval S is still a free choice ≤ W: smaller S means more concurrent -/// overlapping sub-windows maintained internally (⌈W/S⌉), which costs more -/// ingest CPU/mem but gives fresher results. Doubling from scrape_interval up to -/// W keeps the grid small while covering that tradeoff. +/// Tumbling: all W that divide range_a, are multiples of scrape_interval, and ≤ min_t_repeat. +/// Slide interval = W (a tumbling window "slides" by its own width). +/// Sliding: W = range_a exactly (overlapping sliding windows can't merge/subtract to cover +/// a larger range — only mutually-exclusive tumbling windows compose). The slide +/// interval S is still a free choice ≤ W: smaller S means more concurrent +/// overlapping sub-windows maintained internally (⌈W/S⌉), which costs more +/// ingest CPU/mem but gives fresher results. Doubling from scrape_interval up to +/// W keeps the grid small while covering that tradeoff. fn window_candidates( - range_a_ms: Option, + range_a_ms: u64, min_t_repeat_ms: u64, scrape_interval_ms: u64, ) -> Vec<(WindowType, u64, u64, u64)> { - let Some(range_a) = range_a_ms else { - // Spatial-only: any window works (window_compatible returns true for None range). - return vec![( - WindowType::Tumbling, - scrape_interval_ms, - scrape_interval_ms, - 1, - )]; - }; + let range_a = range_a_ms; if range_a == 0 || scrape_interval_ms == 0 { return vec![]; } @@ -122,7 +112,7 @@ fn window_candidates( // Tumbling: W divides range_a, is a multiple of scrape_interval, and ≤ max_w. let mut w = scrape_interval_ms; while w <= max_w { - if range_a % w == 0 { + if range_a.is_multiple_of(w) { let n = range_a / w; out.push((WindowType::Tumbling, w, w, n)); } @@ -329,10 +319,10 @@ mod tests { } #[test] - fn spatial_only_produces_neither_candidates() { - let aqe = make_aqe(Statistic::Sum, None, 60_000); + fn spatial_only_produces_direct_candidates() { + // Spatial-only: range = scrape_interval (set by extract_aqes). One Direct window. + let aqe = make_aqe(Statistic::Sum, Some(15_000), 60_000); let candidates = enumerate_candidates(&aqe, 15_000); - // All non-EXACT candidates should be Direct (no temporal range to cover). for c in candidates.iter().filter(|c| c.config.is_some()) { assert_eq!(c.query_method, QueryMethod::Direct); } diff --git a/asap-planner-rs/src/optimizer/pipeline.rs b/asap-planner-rs/src/optimizer/pipeline.rs index b09ed8b..8251812 100644 --- a/asap-planner-rs/src/optimizer/pipeline.rs +++ b/asap-planner-rs/src/optimizer/pipeline.rs @@ -15,11 +15,12 @@ use super::translator::{translate, TranslationSummary}; fn run_pipeline( config: &ControllerConfig, schema: &PromQLSchema, + scrape_interval_ms: u64, solver_name: &str, solve: impl FnOnce(Vec) -> OptimizerSolution, ) -> (StreamingConfig, InferenceConfig) { let rqes = config_to_rqes(config); - let aqes = extract_aqes(&rqes, schema); + let aqes = extract_aqes(&rqes, schema, scrape_interval_ms); let solution = solve(aqes); let summary = TranslationSummary::from_solution(&solution); @@ -47,8 +48,15 @@ fn run_pipeline( pub fn run_all_exact_pipeline( config: &ControllerConfig, schema: &PromQLSchema, + scrape_interval_ms: u64, ) -> (StreamingConfig, InferenceConfig) { - run_pipeline(config, schema, "all-EXACT", OptimizerSolution::all_exact) + run_pipeline( + config, + schema, + scrape_interval_ms, + "all-EXACT", + OptimizerSolution::all_exact, + ) } /// Run the greedy optimizer pipeline (Phase 2): each AQE is assigned, independently, @@ -66,7 +74,7 @@ pub fn run_greedy_pipeline( scrape_interval_ms: u64, arrival_rate_hz: f64, ) -> (StreamingConfig, InferenceConfig) { - run_pipeline(config, schema, "greedy", |aqes| { + run_pipeline(config, schema, scrape_interval_ms, "greedy", |aqes| { greedy_assign( aqes, scrape_interval_ms, @@ -128,7 +136,7 @@ mod tests { ("sum(other_metric)", 30_000), ]); let schema = PromQLSchema::new(); - let (streaming, _inference) = run_all_exact_pipeline(&config, &schema); + let (streaming, _inference) = run_all_exact_pipeline(&config, &schema, 15_000); // All-EXACT: no streaming configs deployed. assert!(streaming.get_all_aggregation_configs().is_empty()); } @@ -142,6 +150,15 @@ mod tests { assert!(!inference.query_configs.is_empty()); } + #[test] + fn spatial_only_aqe_gets_explicit_range_from_pipeline() { + let config = make_config(&[("sum(metric)", 60_000)]); + let rqes = config_to_rqes(&config); + let aqes = extract_aqes(&rqes, &PromQLSchema::new(), 15_000); + assert_eq!(aqes.len(), 1); + assert_eq!(aqes[0].requirements.data_range_ms, Some(15_000)); + } + #[test] fn config_to_rqes_flattens_groups() { let config = make_config(&[