From 1a14e4418f78ef5ad66a9a9e239e6cb6487230e4 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Tue, 30 Jun 2026 13:37:20 -0400 Subject: [PATCH 1/2] fix(planner): bug fixes to planner's optimizer, added a separte binary to print candidate sketch configurations --- asap-planner-rs/Cargo.toml | 4 + asap-planner-rs/src/bin/candidate_gen_dump.rs | 179 ++++++++++++++++++ .../src/optimizer/aqe_extractor.rs | 2 +- .../src/optimizer/candidate_gen.rs | 118 ++++++++++-- 4 files changed, 284 insertions(+), 19 deletions(-) create mode 100644 asap-planner-rs/src/bin/candidate_gen_dump.rs diff --git a/asap-planner-rs/Cargo.toml b/asap-planner-rs/Cargo.toml index 99544db0..863f19dc 100644 --- a/asap-planner-rs/Cargo.toml +++ b/asap-planner-rs/Cargo.toml @@ -15,6 +15,10 @@ path = "src/main.rs" name = "asap-optimizer-cli" path = "src/bin/optimizer_cli.rs" +[[bin]] +name = "candidate-gen-dump" +path = "src/bin/candidate_gen_dump.rs" + [dependencies] asap_types.workspace = true promql_utilities.workspace = true diff --git a/asap-planner-rs/src/bin/candidate_gen_dump.rs b/asap-planner-rs/src/bin/candidate_gen_dump.rs new file mode 100644 index 00000000..aaec9450 --- /dev/null +++ b/asap-planner-rs/src/bin/candidate_gen_dump.rs @@ -0,0 +1,179 @@ +//! Dump all candidate configs produced by enumerate_candidates for each AQE, +//! before greedy selection. Useful for auditing what the candidate space looks like. + +use std::collections::HashMap; +use std::path::PathBuf; + +use asap_planner::{ + optimizer::{enumerate_candidates, extract_aqes, CandidateConfig, RQE}, + ControllerConfig, +}; +use asap_types::enums::WindowType; +use clap::Parser; +use serde_json::Value; + +#[derive(Parser)] +#[command( + name = "candidate-gen-dump", + about = "Print all candidates from enumerate_candidates (pre-selection) for a workload config" +)] +struct Args { + #[arg(long = "input_config")] + input_config: PathBuf, + + #[arg(long = "data-ingestion-interval-ms")] + scrape_interval_ms: u64, +} + +fn main() -> anyhow::Result<()> { + let args = Args::parse(); + + let yaml_str = std::fs::read_to_string(&args.input_config)?; + let config: ControllerConfig = serde_yaml::from_str(&yaml_str)?; + let schema = config.schema_from_hints(); + + let rqes: Vec = config + .query_groups + .iter() + .flat_map(|qg| { + qg.queries.iter().map(|q| RQE { + query_string: q.clone(), + t_repeat_ms: qg.repetition_delay_ms, + }) + }) + .collect(); + + let aqes = extract_aqes(&rqes, &schema, args.scrape_interval_ms); + println!("=== {} AQE(s) ===", aqes.len()); + + for (i, aqe) in aqes.iter().enumerate() { + println!( + "\n--- AQE #{i}: metric={} stat={:?} range={}ms min_t={}ms gcd_t={}ms freq={:.4}Hz ---", + aqe.requirements.metric, + aqe.requirements.statistics, + aqe.requirements.data_range_ms, + aqe.min_t_repeat_ms, + aqe.t_repeat_gcd_ms, + aqe.query_frequency_hz, + ); + println!(" queries: {:?}", aqe.query_strings); + + let candidates = enumerate_candidates(aqe, args.scrape_interval_ms); + print_candidates_grouped(&candidates); + } + + Ok(()) +} + +/// Group the flat candidate list by (agg_type, sub_type) and print hierarchically: +/// SketchType (sub_type) +/// windows (N): +/// ... +/// params (M) [× N windows = NM total]: +/// ... +/// EXACT is printed last as a single line. +fn print_candidates_grouped(candidates: &[CandidateConfig]) { + // Collect unique (agg_type_str, sub_type) keys in first-seen order. + let mut group_order: Vec<(String, String)> = Vec::new(); + // (agg_type_str, sub_type) -> (unique windows, unique params) + type WindowKey = (WindowType, u64, u64, u64); // (type, size, slide, n) + #[allow(clippy::type_complexity)] + let mut groups: HashMap<(String, String), (Vec, Vec>)> = + HashMap::new(); + + let mut has_exact = false; + + for c in candidates { + let Some(cfg) = &c.config else { + has_exact = true; + continue; + }; + + let key = ( + format!("{:?}", cfg.aggregation_type), + cfg.aggregation_sub_type.clone(), + ); + + let entry = groups.entry(key.clone()).or_insert_with(|| { + group_order.push(key); + (Vec::new(), Vec::new()) + }); + + let wk: WindowKey = ( + cfg.window_type, + cfg.window_size_ms, + cfg.slide_interval_ms, + c.n_windows, + ); + if !entry.0.contains(&wk) { + entry.0.push(wk); + } + + let mut params: Vec<(String, Value)> = cfg + .parameters + .iter() + .map(|(k, v)| (k.clone(), v.clone())) + .collect(); + params.sort_by(|(a, _), (b, _)| a.cmp(b)); + if !entry.1.contains(¶ms) { + entry.1.push(params); + } + } + + let total_sketch: usize = group_order + .iter() + .map(|k| { + let (ws, ps) = &groups[k]; + ws.len() * ps.len() + }) + .sum(); + let total = total_sketch + if has_exact { 1 } else { 0 }; + println!( + " {} candidate(s) across {} sketch group(s):", + total, + group_order.len() + ); + + for key in &group_order { + let (windows, params) = &groups[key]; + let (agg_type_str, sub_type) = key; + let sub = if sub_type.is_empty() { + String::new() + } else { + format!(" ({})", sub_type) + }; + println!( + "\n {}{} — {} window(s) × {} param(s) = {} candidate(s):", + agg_type_str, + sub, + windows.len(), + params.len(), + windows.len() * params.len(), + ); + + println!(" windows:"); + for (wtype, size, slide, n) in windows { + let window_str = match wtype { + WindowType::Tumbling => format!("Tumbling {}ms", size), + WindowType::Sliding => format!("Sliding {}ms / slide {}ms", size, slide), + }; + println!( + " {} n={} → {:?}", + window_str, + n, + // Re-derive method label from n and window type for display. + if *n == 1 { "Direct" } else { "Merge/Subtract" } + ); + } + + println!(" params:"); + for p in params { + let kv: Vec<_> = p.iter().map(|(k, v)| format!("{k}: {v}")).collect(); + println!(" {{{}}}", kv.join(", ")); + } + } + + if has_exact { + println!("\n [EXACT]"); + } +} diff --git a/asap-planner-rs/src/optimizer/aqe_extractor.rs b/asap-planner-rs/src/optimizer/aqe_extractor.rs index ece963ed..77e45dd0 100644 --- a/asap-planner-rs/src/optimizer/aqe_extractor.rs +++ b/asap-planner-rs/src/optimizer/aqe_extractor.rs @@ -131,7 +131,7 @@ pub fn extract_aqes( /// Euclidean GCD. `num-integer` is not in the workspace; this two-liner is /// sufficient and avoids a dependency. -fn gcd(a: u64, b: u64) -> u64 { +pub(super) fn gcd(a: u64, b: u64) -> u64 { if b == 0 { a } else { diff --git a/asap-planner-rs/src/optimizer/candidate_gen.rs b/asap-planner-rs/src/optimizer/candidate_gen.rs index 8e06eaa1..db62b30d 100644 --- a/asap-planner-rs/src/optimizer/candidate_gen.rs +++ b/asap-planner-rs/src/optimizer/candidate_gen.rs @@ -49,7 +49,7 @@ pub fn enumerate_candidates(aqe: &AQE, scrape_interval_ms: u64) -> Vec CandidateConfig { /// Window candidates: (WindowType, W_ms, slide_interval_ms, n_windows). /// -/// Tumbling: all W that divide range_a, are multiples of scrape_interval, and ≤ min_t_repeat. +/// Tumbling: W must divide GCD(range_a, t_repeat_gcd_ms) and be a multiple of scrape_interval. +/// Dividing the GCD ensures (a) n complete windows cover range_a exactly, and +/// (b) window completions are harmonically aligned with every dashboard refresh cycle. /// Slide interval = W (a tumbling window "slides" by its own width). /// Sliding: W = range_a / k for each W that is a multiple of scrape_interval and divides -/// range_a (k = range_a / W). At query time k staggered readings are merged or -/// subtracted to cover range_a. Freshness is bounded by S (the slide interval), -/// not by W — so the guard is S ≤ min_t_repeat_ms, not W ≤ min_t_repeat_ms. -/// S doubles from scrape_interval up to min(W, min_t_repeat_ms). +/// range_a (k = range_a / W). At query time k staggered readings spaced W apart +/// are merged or subtracted to cover range_a. +/// S must satisfy three constraints: +/// (a) S | W — so W-spaced snapshots land on slide boundaries (multi-window correctness) +/// (b) S | t_repeat_gcd — so slide boundaries align with every dashboard refresh cycle +/// (c) S < W — S=W is excluded because slide==window is tumbling (duplicate candidate) +/// S is enumerated as multiples of scrape_interval < W; the divisibility check on +/// gcd(W, t_repeat_gcd) rejects values that fail (a) or (b) without a separate bound. fn window_candidates( range_a_ms: u64, - min_t_repeat_ms: u64, + t_repeat_gcd_ms: u64, scrape_interval_ms: u64, ) -> Vec<(WindowType, u64, u64, u64)> { let range_a = range_a_ms; @@ -105,13 +111,16 @@ fn window_candidates( return vec![]; } - let max_w = range_a.min(min_t_repeat_ms); let mut out = Vec::new(); - // Tumbling: W divides range_a, is a multiple of scrape_interval, and ≤ max_w. + // Tumbling: W divides GCD(range_a, t_repeat_gcd) and is a multiple of scrape_interval. + // W | t_repeat_gcd ensures window completions align harmonically with all dashboards. + // W | range_a (implied since t_repeat_gcd | range_a is checked at generation time, but + // we verify explicitly via the gcd) ensures n windows cover range_a exactly. + let tumbling_divisor = super::aqe_extractor::gcd(range_a, t_repeat_gcd_ms); let mut w = scrape_interval_ms; - while w <= max_w { - if range_a.is_multiple_of(w) { + while w <= tumbling_divisor { + if tumbling_divisor.is_multiple_of(w) { let n = range_a / w; out.push((WindowType::Tumbling, w, w, n)); } @@ -124,10 +133,15 @@ fn window_candidates( while w <= range_a { if range_a.is_multiple_of(w) { let k = range_a / w; + // Valid S: S | gcd(W, t_repeat_gcd). Iterate up to W (exclusive); the + // divisibility check rejects anything above gcd automatically. + let slide_divisor = super::aqe_extractor::gcd(w, t_repeat_gcd_ms); let mut s = scrape_interval_ms; - while s <= w.min(min_t_repeat_ms) { - out.push((WindowType::Sliding, w, s, k)); - s *= 2; + while s < w { + if slide_divisor.is_multiple_of(s) { + out.push((WindowType::Sliding, w, s, k)); + } + s += scrape_interval_ms; } } w += scrape_interval_ms; @@ -398,10 +412,9 @@ mod tests { } #[test] - fn sliding_freshness_guard_uses_slide_not_width() { - // range_a=600_000ms > min_t_repeat=30_000ms. The old guard (range_a <= min_t_repeat) - // incorrectly rejected all Sliding candidates here. New guard: S ≤ min_t_repeat. - // W=600_000, S=30_000 ≤ min_t_repeat → full-width Direct Sliding should be present. + fn sliding_full_width_direct_generated_when_range_exceeds_t_repeat() { + // range_a=600_000ms > t_repeat_gcd=30_000ms. W=range_a is valid for sliding since + // freshness is governed by S (not W). S=30_000 | gcd(600_000, 30_000)=30_000 → emitted. let aqe = make_aqe(Statistic::Sum, 600_000, 30_000); let candidates = enumerate_candidates(&aqe, 30_000); assert!( @@ -415,6 +428,75 @@ mod tests { ); } + #[test] + fn sliding_slide_must_divide_gcd_of_window_and_t_repeat() { + // range_a=20_000, t_repeat_gcd=5_000, scrape=1_000. + // W=10_000 (k=2): slide_divisor = gcd(10_000, 5_000) = 5_000. + // Valid S: divisors of 5_000 that are multiples of 1_000 and < 10_000 → {1_000, 5_000}. + // Invalid: S=2_000 (5_000 % 2_000 ≠ 0), S=4_000 (5_000 % 4_000 ≠ 0). + let aqe = make_aqe(Statistic::Sum, 20_000, 5_000); + let candidates = enumerate_candidates(&aqe, 1_000); + + let sliding_w10: Vec<_> = candidates + .iter() + .filter(|c| { + c.config.as_ref().is_some_and(|cfg| { + cfg.window_type == WindowType::Sliding && cfg.window_size_ms == 10_000 + }) + }) + .collect(); + + let slides: Vec = sliding_w10 + .iter() + .map(|c| c.config.as_ref().unwrap().slide_interval_ms) + .collect(); + + assert!( + slides.contains(&1_000), + "S=1_000 should be valid (divides 5_000)" + ); + assert!( + slides.contains(&5_000), + "S=5_000 should be valid (divides 5_000)" + ); + assert!( + !slides.contains(&2_000), + "S=2_000 should be rejected (5_000 % 2_000 ≠ 0)" + ); + assert!( + !slides.contains(&4_000), + "S=4_000 should be rejected (5_000 % 4_000 ≠ 0)" + ); + } + + #[test] + fn sliding_slide_must_divide_window_size() { + // W=6_000, t_repeat_gcd=6_000: slide_divisor = gcd(6_000, 6_000) = 6_000. + // S=4_000: 6_000 % 4_000 = 2_000 ≠ 0 → rejected even though 4_000 < 6_000. + // S=2_000: 6_000 % 2_000 = 0 → valid. + let aqe = make_aqe(Statistic::Sum, 12_000, 6_000); + let candidates = enumerate_candidates(&aqe, 1_000); + + let slides_w6: Vec = candidates + .iter() + .filter(|c| { + c.config.as_ref().is_some_and(|cfg| { + cfg.window_type == WindowType::Sliding && cfg.window_size_ms == 6_000 + }) + }) + .map(|c| c.config.as_ref().unwrap().slide_interval_ms) + .collect(); + + assert!( + slides_w6.contains(&2_000), + "S=2_000 should be valid (6_000 % 2_000 = 0)" + ); + assert!( + !slides_w6.contains(&4_000), + "S=4_000 should be rejected (6_000 % 4_000 ≠ 0)" + ); + } + #[test] fn partial_sliding_subtractable_sketch_gets_subtract() { // Sum → CMS (subtractable): partial Sliding with k=2 should produce Subtract. From cb11d8429b913a2bd00af39b036e09e1e9f6f1cb Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Tue, 30 Jun 2026 13:43:20 -0400 Subject: [PATCH 2/2] CI bug --- asap-planner-rs/Dockerfile | 1 + 1 file changed, 1 insertion(+) diff --git a/asap-planner-rs/Dockerfile b/asap-planner-rs/Dockerfile index 6da5ede7..3f4ebb10 100644 --- a/asap-planner-rs/Dockerfile +++ b/asap-planner-rs/Dockerfile @@ -21,6 +21,7 @@ RUN mkdir -p asap-query-engine/src && echo "fn main() {}" > asap-query-engine/sr mkdir -p asap-query-engine/benches && echo "fn main() {}" > asap-query-engine/benches/simple_store_bench.rs && \ mkdir -p asap-planner-rs/src/bin && echo "fn main() {}" > asap-planner-rs/src/main.rs && \ echo "fn main() {}" > asap-planner-rs/src/bin/optimizer_cli.rs && \ + echo "fn main() {}" > asap-planner-rs/src/bin/candidate_gen_dump.rs && \ echo "pub fn placeholder() {}" >> asap-planner-rs/src/lib.rs # Build dependencies (this layer will be cached)