diff --git a/.design_docs/optimizer-v1-implementation-plan.md b/.design_docs/optimizer-v1-implementation-plan.md index 71fd625..efbbc43 100644 --- a/.design_docs/optimizer-v1-implementation-plan.md +++ b/.design_docs/optimizer-v1-implementation-plan.md @@ -80,6 +80,8 @@ asap-planner-rs/src/optimizer/ ├── candidate_gen.rs enumerate candidate configs per AQE [Phase 2b, done] ├── cost_model.rs ingest/query cost formulas [Phase 2c, done] ├── greedy.rs per-AQE greedy assignment [Phase 2e, done] +├── atomic_costs.rs (sketch_type, params) → AtomicCosts, from +│ sketch-bench's exported table [Phase 2f, done, #549] │ │ ── TO BE ADDED ── ├── feasibility.rs Feasible(a,g) predicate — only meaningful once configs @@ -222,6 +224,33 @@ placeholder value applied uniformly to every candidate — see open TODO below. a real `aggregation_id`, emits one `QueryConfig` per original query string, with `num_aggregates_to_retain` from `retention_count_for_assignment()`. +#### 2f — `atomic_costs.rs` (done, issue #549, carved out of #524) + +`greedy.rs` no longer applies one global `AtomicCosts` to every candidate. Per candidate: + +```rust +pub fn resolve_atomic_costs(table: &AtomicCostTable, agg_type: AggregationType, + params: &HashMap) -> Option +``` + +- `CountMinSketch` / `HLL` / `DatasketchesKLL`: translated to sketch-bench's `(algorithm, params)` + key (`sketch_bench_key()` — field names are duplicated from `candidate_gen.rs`'s `param_grid()`, + not derived from it, so a rename on either side without the other panics loudly instead of + silently mismatching) and looked up by exact key in the table. No match for that exact param + point → `None`, candidate dropped. +- Every other `AggregationType` (trivial accumulators like `Sum`/`MinMax`, and sketches + sketch-bench doesn't wrap yet like `CountMinSketchWithHeap`/`HydraKLL`) → `Some(AtomicCosts::default())`, + the pre-#549 flat stub, with a `tracing::warn!`. **This means costs for those families are still + param-invariant** (e.g. `CountMinSketchWithHeap` currently costs identically regardless of + `depth`/`width`/`heapsize` — see #524 for the analytic memory bound that's supposed to replace + the stub for it). +- No `subtract_cpu_secs` in the table (asap_sketchlib has no `subtract` yet — asap_sketchlib#69) — + always comes from the stub constant regardless of family. + +`AtomicCostEntry`/`AtomicCostTable` are a deliberate duplicate of sketch-bench's +`aqpbm_core::atomic_costs` types (same reasoning as the field-name duplication above: sketch-bench's +schema is still actively churning, so no shared crate dependency yet — see PR #547's discussion). + --- ### 🔲 Phase 3 — Full MIP with Cross-AQE Sharing @@ -236,11 +265,18 @@ Add `run_mip_pipeline()` to `pipeline.rs`. Relax `labels_compatible()` in `asap_types::capability_matching` at line 86 (TODO comment already there). Allow a config with labels ⊇ query labels to serve that AQE. This is what enables cross-AQE sharing. -#### 3c — sketch-bench cardinality sweep +#### 3c — real `AtomicCosts` from sketch-bench -- Add cardinality to sweep grids in `sketch-bench` -- Add `CountMinSketchWithHeap` wrapper in `sketch-cli/src/wrappers/` -- Plug real `AtomicCosts` values into cost model +- ✅ Plug real `AtomicCosts` values into cost model — done for CMS/HLL/KLL, see 2f above and + "Running with real sketch-bench costs" below. +- ❌ Add `CountMinSketchWithHeap` wrapper in sketch-bench, so it stops always costing at the flat + stub. Its memory is meant to be an analytic bound (`heap_size · avg_key_size`), not a + sketch-bench lookup at all, per #524 — that formula still needs implementing in + `atomic_costs.rs`/`cost_model.rs`. +- ~~Add cardinality to sweep grids in `sketch-bench`~~ — decided unnecessary: CPU/mem costs for + CMS/HLL/KLL are functions of structural params (depth×width, lg_k, K), not cardinality: only + `CountMinSketchWithHeap` is cardinality-dependent, and that's the analytic-bound case above, not + a sketch-bench sweep axis. #### 3d — Accuracy constraint @@ -265,14 +301,45 @@ standalone `asap-optimizer-cli` binary (`asap-planner-rs/src/bin/optimizer_cli.r ``` cargo run -p asap_planner --bin asap-optimizer-cli -- \ --input_config \ - --prometheus_scrape_interval 60 \ - [--rho 1.0] + --data-ingestion-interval-ms 60000 \ + [--rho 1.0] \ + [--atomic-costs ] ``` Takes the same `ControllerConfig` YAML format as `asap-planner --input_config` (with a `metrics:` hints block for label schema — no live Prometheus needed). Prints deployed streaming configs and query configs to stdout. `--rho` is the -placeholder arrival rate (see TODOs below — not real yet). +placeholder arrival rate (see TODOs below — not real yet). `--atomic-costs` is +optional; omit it and every candidate costs at the flat stub, same as before #549. + +### Running with real sketch-bench costs + +**1. Generate the table, in the `sketch-bench` repo:** + +``` +./scripts/export_atomic_costs.sh +# → out/atomic_costs.json — one row per (sketch, params) point in ASAPQuery's grid, +# built by looping candidate_gen.rs's CMS_DEPTHS×CMS_WIDTHS/HLL_PRECISIONS/KLL_KS +# through `approxbench sketchbench --flat` and reducing with `approxbench atomic-costs`. +``` + +**2. Feed it to ASAPQuery**, either binary: + +``` +# See what each candidate would cost, before selection: +cargo run -p asap_planner --bin candidate-gen-dump -- \ + --input_config workload.yaml --data-ingestion-interval-ms 60000 \ + --atomic-costs path/to/atomic_costs.json + +# Run the actual optimizer: +cargo run -p asap_planner --bin asap-optimizer-cli -- \ + --input_config workload.yaml --data-ingestion-interval-ms 60000 \ + --atomic-costs path/to/atomic_costs.json +``` + +`candidate-gen-dump`'s output labels each params row `[real]` (resolved from the table) or +`[stub]` (fell through to `AtomicCosts::default()` — either an unbenchmarked family, or a +benchmarked family's param point missing from the table). Wire-in decision (deferred): once Phase 3 (MIP + feasibility + label superset matching) lands, swap `Controller::generate()` to call `run_mip_pipeline()` diff --git a/asap-planner-rs/src/bin/candidate_gen_dump.rs b/asap-planner-rs/src/bin/candidate_gen_dump.rs index aaec945..a900125 100644 --- a/asap-planner-rs/src/bin/candidate_gen_dump.rs +++ b/asap-planner-rs/src/bin/candidate_gen_dump.rs @@ -5,11 +5,15 @@ use std::collections::HashMap; use std::path::PathBuf; use asap_planner::{ - optimizer::{enumerate_candidates, extract_aqes, CandidateConfig, RQE}, + optimizer::{ + enumerate_candidates, extract_aqes, load_atomic_cost_table, resolve_atomic_costs, + AtomicCostTable, AtomicCosts, CandidateConfig, RQE, + }, ControllerConfig, }; use asap_types::enums::WindowType; use clap::Parser; +use promql_utilities::query_logics::enums::AggregationType; use serde_json::Value; #[derive(Parser)] @@ -23,11 +27,23 @@ struct Args { #[arg(long = "data-ingestion-interval-ms")] scrape_interval_ms: u64, + + /// Path to sketch-bench's exported atomic-cost table (see ASAPQuery#524). + /// When given, each params row also prints its resolved AtomicCosts -- + /// real (from the table) or the flat stub (unbenchmarked family, or this + /// exact param point missing from the table) -- labeled which. + #[arg(long = "atomic-costs")] + atomic_costs: Option, } fn main() -> anyhow::Result<()> { let args = Args::parse(); + let atomic_cost_table = match &args.atomic_costs { + Some(path) => load_atomic_cost_table(path)?, + None => AtomicCostTable::default(), + }; + 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(); @@ -59,7 +75,7 @@ fn main() -> anyhow::Result<()> { println!(" queries: {:?}", aqe.query_strings); let candidates = enumerate_candidates(aqe, args.scrape_interval_ms); - print_candidates_grouped(&candidates); + print_candidates_grouped(&candidates, &atomic_cost_table); } Ok(()) @@ -72,14 +88,16 @@ fn main() -> anyhow::Result<()> { /// params (M) [× N windows = NM total]: /// ... /// EXACT is printed last as a single line. -fn print_candidates_grouped(candidates: &[CandidateConfig]) { +fn print_candidates_grouped(candidates: &[CandidateConfig], atomic_cost_table: &AtomicCostTable) { // 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) + // (agg_type_str, sub_type) -> (agg_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 groups: HashMap< + (String, String), + (AggregationType, Vec, Vec>), + > = HashMap::new(); let mut has_exact = false; @@ -96,7 +114,7 @@ fn print_candidates_grouped(candidates: &[CandidateConfig]) { let entry = groups.entry(key.clone()).or_insert_with(|| { group_order.push(key); - (Vec::new(), Vec::new()) + (cfg.aggregation_type, Vec::new(), Vec::new()) }); let wk: WindowKey = ( @@ -105,8 +123,8 @@ fn print_candidates_grouped(candidates: &[CandidateConfig]) { cfg.slide_interval_ms, c.n_windows, ); - if !entry.0.contains(&wk) { - entry.0.push(wk); + if !entry.1.contains(&wk) { + entry.1.push(wk); } let mut params: Vec<(String, Value)> = cfg @@ -115,15 +133,15 @@ fn print_candidates_grouped(candidates: &[CandidateConfig]) { .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); + if !entry.2.contains(¶ms) { + entry.2.push(params); } } let total_sketch: usize = group_order .iter() .map(|k| { - let (ws, ps) = &groups[k]; + let (_, ws, ps) = &groups[k]; ws.len() * ps.len() }) .sum(); @@ -135,7 +153,7 @@ fn print_candidates_grouped(candidates: &[CandidateConfig]) { ); for key in &group_order { - let (windows, params) = &groups[key]; + let (agg_type, windows, params) = &groups[key]; let (agg_type_str, sub_type) = key; let sub = if sub_type.is_empty() { String::new() @@ -169,7 +187,24 @@ fn print_candidates_grouped(candidates: &[CandidateConfig]) { println!(" params:"); for p in params { let kv: Vec<_> = p.iter().map(|(k, v)| format!("{k}: {v}")).collect(); - println!(" {{{}}}", kv.join(", ")); + let costs_str = if atomic_cost_table.is_empty() { + String::new() + } else { + let param_map: HashMap = + p.iter().map(|(k, v)| (k.clone(), v.clone())).collect(); + match resolve_atomic_costs(atomic_cost_table, *agg_type, ¶m_map) { + Some(costs) => { + let label = if costs == AtomicCosts::default() { + "stub" + } else { + "real" + }; + format!(" -> [{label}] {}", format_costs(&costs)) + } + None => " -> DROPPED (no matching table row)".to_string(), + } + }; + println!(" {{{}}}{costs_str}", kv.join(", ")); } } @@ -177,3 +212,14 @@ fn print_candidates_grouped(candidates: &[CandidateConfig]) { println!("\n [EXACT]"); } } + +fn format_costs(costs: &AtomicCosts) -> String { + format!( + "mem={:.0}B insert={:.3e}s merge={:.3e}s subtract={:.3e}s query={:.3e}s", + costs.mem_bytes_per_instance, + costs.insert_cpu_secs, + costs.merge_cpu_secs, + costs.subtract_cpu_secs, + costs.query_cpu_secs, + ) +} diff --git a/asap-planner-rs/src/bin/optimizer_cli.rs b/asap-planner-rs/src/bin/optimizer_cli.rs index 2ce62d4..df40b22 100644 --- a/asap-planner-rs/src/bin/optimizer_cli.rs +++ b/asap-planner-rs/src/bin/optimizer_cli.rs @@ -6,7 +6,7 @@ use std::path::PathBuf; -use asap_planner::optimizer::run_greedy_pipeline; +use asap_planner::optimizer::{load_atomic_cost_table, run_greedy_pipeline, AtomicCostTable}; use asap_planner::ControllerConfig; use clap::Parser; @@ -29,6 +29,14 @@ struct Args { #[arg(long = "rho", default_value = "1.0", value_parser = parse_positive_finite)] rho: f64, + /// Path to the atomic-cost table sketch-bench's `atomic-costs` subcommand + /// exports (see ASAPQuery#524, sketch-bench#30). Omitted: every + /// benchmarked-family candidate (CMS/HLL/KLL) is dropped, since there is + /// no data to cost it at — only trivial accumulators and EXACT remain + /// selectable. + #[arg(long = "atomic-costs")] + atomic_costs: Option, + #[arg(short, long, action = clap::ArgAction::Count)] verbose: u8, } @@ -56,8 +64,23 @@ fn main() -> anyhow::Result<()> { let config: ControllerConfig = serde_yaml::from_str(&yaml_str)?; let schema = config.schema_from_hints(); - let (streaming, inference) = - run_greedy_pipeline(&config, &schema, args.data_ingestion_interval_ms, args.rho); + let atomic_cost_table = match &args.atomic_costs { + Some(path) => load_atomic_cost_table(path)?, + None => { + tracing::warn!( + "no --atomic-costs supplied; CMS/HLL/KLL candidates will never be selected" + ); + AtomicCostTable::default() + } + }; + + let (streaming, inference) = run_greedy_pipeline( + &config, + &schema, + args.data_ingestion_interval_ms, + args.rho, + &atomic_cost_table, + ); let deployed = streaming.get_all_aggregation_configs(); println!("=== Deployed streaming configs: {} ===", deployed.len()); diff --git a/asap-planner-rs/src/optimizer/atomic_costs.rs b/asap-planner-rs/src/optimizer/atomic_costs.rs new file mode 100644 index 0000000..2b48965 --- /dev/null +++ b/asap-planner-rs/src/optimizer/atomic_costs.rs @@ -0,0 +1,266 @@ +//! The atomic-cost table exported by sketch-bench (sketch-bench#30, +//! `scripts/export_atomic_costs.sh`), and the (sketch_type, params) lookup +//! that resolves a candidate's [`AtomicCosts`] from it. +//! +//! `AtomicCostEntry`/`AtomicCostTable` are a deliberate duplicate of +//! sketch-bench's `aqpbm_core::atomic_costs` types, not a shared dependency — +//! see ASAPQuery#524 and sketch-bench#30 for why. Keep the two in sync by +//! hand; `atomic_cost_entry_deserializes_sketch_benchs_documented_shape` +//! below is a canary for drift. + +use std::collections::HashMap; +use std::path::Path; + +use promql_utilities::query_logics::enums::AggregationType; +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +use super::constants::{EXACT_QUERY_CPU_SECS, SUBTRACT_CPU_SECS}; +use super::cost_model::AtomicCosts; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct AtomicCostEntry { + pub sketch: String, + pub sketch_config: Value, + pub mem_bytes_per_instance: f64, + pub insert_cpu_secs: f64, + pub merge_cpu_secs: f64, + pub query_cpu_secs: f64, +} + +pub type AtomicCostTable = Vec; + +/// Read an `AtomicCostTable` exported by `sketch-bench atomic-costs`. +pub fn load_atomic_cost_table(path: &Path) -> anyhow::Result { + let raw = std::fs::read_to_string(path) + .map_err(|e| anyhow::anyhow!("reading atomic-cost table {}: {e}", path.display()))?; + serde_json::from_str(&raw) + .map_err(|e| anyhow::anyhow!("parsing atomic-cost table {}: {e}", path.display())) +} + +/// sketch-bench's (algorithm, params) key for one of ASAPQuery's benchmarked +/// families, or `None` if `agg_type` isn't one sketch-bench measures at all +/// (trivial O(1) accumulators — Sum/Increase/MinMax/... — and sketch types +/// sketch-bench has no wrapper for yet — CountMinSketchWithHeap, HydraKLL). +/// +/// Field names differ from ASAPQuery's own `parameters` map by design: each +/// side picked its own config vocabulary independently, so this is a real +/// translation, not a passthrough. The field names read here (`"depth"`, +/// `"width"`, `"precision"`, `"K"`) are duplicated from `candidate_gen.rs`'s +/// `param_grid()` — not derived from it — so a rename on either side without +/// the other silently breaks this lookup. Guarded by panicking below rather +/// than treating a missing key the same as "not a benchmarked family": a +/// `CountMinSketch`/`HLL`/`DatasketchesKLL` candidate is only ever built by +/// `param_grid()`, which always sets these keys, so their absence means the +/// two have drifted, not that there's no data for this family. +fn sketch_bench_key( + agg_type: AggregationType, + params: &HashMap, +) -> Option<(&'static str, Value)> { + fn require<'a>( + params: &'a HashMap, + key: &str, + agg_type: AggregationType, + ) -> &'a Value { + params.get(key).unwrap_or_else(|| { + panic!( + "{agg_type:?} candidate has no \"{key}\" param; sketch_bench_key's field \ + names have drifted from candidate_gen.rs's param_grid()" + ) + }) + } + + match agg_type { + AggregationType::CountMinSketch => Some(( + "cms-fastpath-vector2d", + serde_json::json!({ + "rows": require(params, "depth", agg_type), + "cols": require(params, "width", agg_type), + }), + )), + AggregationType::HLL => Some(( + "hll", + serde_json::json!({ "lg_k": require(params, "precision", agg_type) }), + )), + AggregationType::DatasketchesKLL => Some(( + "kll-percall", + serde_json::json!({ "k": require(params, "K", agg_type) }), + )), + _ => None, + } +} + +/// Resolve the [`AtomicCosts`] a candidate should be costed at. +/// +/// - `agg_type` outside the benchmarked families (see [`sketch_bench_key`]): +/// `Some(AtomicCosts::default())` — the flat stub, unchanged from before +/// this table existed. Logged, since it's silently wrong for anything +/// sketch-bench could plausibly measure later. +/// TODO(#524): remove this fallback once every family the optimizer can +/// select has a real sketch-bench entry; costing should end up 100% +/// empirical, with nothing left reading `AtomicCosts::default()`. +/// - Benchmarked family, matching table row found: `Some(costs)` built from +/// it (`subtract_cpu_secs`/`exact_query_cpu_secs` still come from the +/// stub — the table has neither: subtract isn't implemented upstream yet, +/// and EXACT isn't a sketch sketch-bench could measure). +/// - Benchmarked family, no matching row (e.g. a param point outside the +/// swept grid, or a family sketch-bench doesn't wrap yet like +/// `CountMinSketchWithHeap`): `None` — drop the candidate, per #524. +pub fn resolve_atomic_costs( + table: &AtomicCostTable, + agg_type: AggregationType, + params: &HashMap, +) -> Option { + let Some((sketch, sketch_params)) = sketch_bench_key(agg_type, params) else { + tracing::warn!( + ?agg_type, + "no sketch-bench atomic-cost data for this family; using the flat AtomicCosts stub" + ); + return Some(AtomicCosts::default()); + }; + + let expected_config = serde_json::json!({ "algorithm": sketch, "params": sketch_params }); + table + .iter() + .find(|e| e.sketch == sketch && e.sketch_config == expected_config) + .map(|entry| AtomicCosts { + mem_bytes_per_instance: entry.mem_bytes_per_instance, + insert_cpu_secs: entry.insert_cpu_secs, + merge_cpu_secs: entry.merge_cpu_secs, + subtract_cpu_secs: SUBTRACT_CPU_SECS, + query_cpu_secs: entry.query_cpu_secs, + exact_query_cpu_secs: EXACT_QUERY_CPU_SECS, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn cms_entry(depth: i64, width: i64) -> AtomicCostEntry { + AtomicCostEntry { + sketch: "cms-fastpath-vector2d".into(), + sketch_config: serde_json::json!({ + "algorithm": "cms-fastpath-vector2d", + "params": { "cols": width, "rows": depth } + }), + mem_bytes_per_instance: (depth * width * 4) as f64, + insert_cpu_secs: 8e-9, + merge_cpu_secs: 4.5e-4, + query_cpu_secs: 7.8e-8, + } + } + + fn cms_params(depth: u64, width: u64) -> HashMap { + HashMap::from([ + ("depth".to_string(), Value::from(depth)), + ("width".to_string(), Value::from(width)), + ]) + } + + #[test] + fn atomic_cost_entry_deserializes_sketch_benchs_documented_shape() { + // Pinned against a real row sketch-bench's `atomic-costs` subcommand + // actually emitted (out/atomic_costs.json, cms-fastpath-vector2d + // rows=3 cols=1024) -- a canary for the two structs drifting apart. + let json = r#"{"sketch":"cms-fastpath-vector2d","sketch_config":{"algorithm":"cms-fastpath-vector2d","params":{"cols":1024,"rows":3}},"mem_bytes_per_instance":12288.0,"insert_cpu_secs":8.484689139741214e-9,"merge_cpu_secs":0.00045364040539336466,"query_cpu_secs":7.799774697708031e-8}"#; + let entry: AtomicCostEntry = serde_json::from_str(json).expect("documented shape parses"); + assert_eq!(entry.sketch, "cms-fastpath-vector2d"); + assert_eq!(entry.mem_bytes_per_instance, 12288.0); + } + + #[test] + fn cms_candidate_resolves_by_exact_key_regardless_of_value_type() { + // ASAPQuery's grid stores depth/width as u64; sketch-bench's exported + // JSON round-trips CLI-parsed integers as i64. The lookup must not + // care which Rust integer type produced the JSON number. + let table = vec![cms_entry(3, 1024), cms_entry(5, 2048)]; + let costs = resolve_atomic_costs( + &table, + AggregationType::CountMinSketch, + &cms_params(3, 1024), + ) + .expect("exact grid point must resolve"); + assert_eq!(costs.mem_bytes_per_instance, 3.0 * 1024.0 * 4.0); + assert_eq!(costs.insert_cpu_secs, 8e-9); + // Not from the table -- sketch-bench has neither, so these stay stub. + assert_eq!(costs.subtract_cpu_secs, SUBTRACT_CPU_SECS); + assert_eq!(costs.exact_query_cpu_secs, EXACT_QUERY_CPU_SECS); + } + + #[test] + fn cms_param_point_outside_the_grid_drops_the_candidate() { + let table = vec![cms_entry(3, 1024)]; + assert!( + resolve_atomic_costs(&table, AggregationType::CountMinSketch, &cms_params(7, 999)) + .is_none() + ); + } + + #[test] + fn unbenchmarked_family_falls_back_to_the_stub() { + let table: AtomicCostTable = vec![]; + let costs = resolve_atomic_costs(&table, AggregationType::Sum, &HashMap::new()) + .expect("unbenchmarked families still get a usable (stub) cost"); + assert_eq!( + costs.mem_bytes_per_instance, + AtomicCosts::default().mem_bytes_per_instance + ); + } + + #[test] + fn cms_with_heap_has_no_translation_and_falls_back_to_the_stub() { + // Real sketch (not trivial), just not wrapped by sketch-bench yet -- + // still goes through the stub path, same as a trivial accumulator, + // per the ASAPQuery#524 scope decision. + let table: AtomicCostTable = vec![]; + assert!(resolve_atomic_costs( + &table, + AggregationType::CountMinSketchWithHeap, + &HashMap::new() + ) + .is_some()); + } + + #[test] + #[should_panic(expected = "has no \"depth\" param")] + fn cms_candidate_missing_its_expected_param_panics_instead_of_silently_stubbing() { + // A CountMinSketch candidate only ever comes from candidate_gen.rs's + // param_grid(), which always sets "depth"/"width". Landing here without + // one means sketch_bench_key's field names have drifted from + // param_grid()'s -- a real bug, not "this family has no data" (which + // resolve_atomic_costs already covers via cms_with_heap above and + // must stay visibly different from this case). + let table: AtomicCostTable = vec![]; + let params = HashMap::from([("width".to_string(), Value::from(1024u64))]); + resolve_atomic_costs(&table, AggregationType::CountMinSketch, ¶ms); + } + + #[test] + fn hll_and_kll_translate_and_resolve() { + let hll_table = vec![AtomicCostEntry { + sketch: "hll".into(), + sketch_config: serde_json::json!({"algorithm": "hll", "params": {"lg_k": 14}}), + mem_bytes_per_instance: 16384.0, + insert_cpu_secs: 1.68e-9, + merge_cpu_secs: 2.76e-4, + query_cpu_secs: 1.23e-4, + }]; + let hll_params = HashMap::from([("precision".to_string(), Value::from(14u64))]); + assert!(resolve_atomic_costs(&hll_table, AggregationType::HLL, &hll_params).is_some()); + + let kll_table = vec![AtomicCostEntry { + sketch: "kll-percall".into(), + sketch_config: serde_json::json!({"algorithm": "kll-percall", "params": {"k": 200}}), + mem_bytes_per_instance: 6400.0, + insert_cpu_secs: 1.6e-8, + merge_cpu_secs: 1.0e-3, + query_cpu_secs: 1.6e-4, + }]; + let kll_params = HashMap::from([("K".to_string(), Value::from(200u64))]); + assert!( + resolve_atomic_costs(&kll_table, AggregationType::DatasketchesKLL, &kll_params) + .is_some() + ); + } +} diff --git a/asap-planner-rs/src/optimizer/cost_model.rs b/asap-planner-rs/src/optimizer/cost_model.rs index b82015e..50b7870 100644 --- a/asap-planner-rs/src/optimizer/cost_model.rs +++ b/asap-planner-rs/src/optimizer/cost_model.rs @@ -11,7 +11,7 @@ use super::solution::{QueryMethod, AQE}; /// Per-operation costs for one sketch instance. Stub defaults for v1 — real /// values come from sketch-bench in Phase 3 (see implementation plan, 3c). -#[derive(Debug, Clone, Copy)] +#[derive(Debug, Clone, Copy, PartialEq)] pub struct AtomicCosts { pub mem_bytes_per_instance: f64, pub insert_cpu_secs: f64, diff --git a/asap-planner-rs/src/optimizer/greedy.rs b/asap-planner-rs/src/optimizer/greedy.rs index 4cf19b1..3275a32 100644 --- a/asap-planner-rs/src/optimizer/greedy.rs +++ b/asap-planner-rs/src/optimizer/greedy.rs @@ -3,6 +3,7 @@ use std::collections::HashMap; use asap_types::aggregation_config::AggregationConfig; use tracing::debug; +use super::atomic_costs::{resolve_atomic_costs, AtomicCostTable}; use super::candidate_gen::enumerate_candidates; use super::cost_model::{ingest_cost, query_cost, total_cost_rate, AtomicCosts, CostWeights}; use super::solution::{AQEAssignment, OptimizerSolution, AQE}; @@ -16,11 +17,16 @@ use super::solution::{AQEAssignment, OptimizerSolution, AQE}; /// `arrival_rate_hz` is the per-item arrival rate used for every candidate's IngestCost. /// Real per-config rates need Prometheus scrape-rate × series-count data, /// which isn't wired up yet — a single placeholder value is applied uniformly. +/// +/// Each candidate is costed at its own `(sketch_type, params)` via +/// `atomic_cost_table` (see ASAPQuery#524) rather than one cost applied to +/// every candidate; a candidate whose config has no matching table entry is +/// dropped from consideration (`resolve_atomic_costs` returns `None`). pub fn greedy_assign( aqes: Vec, scrape_interval_ms: u64, arrival_rate_hz: f64, - costs: &AtomicCosts, + atomic_cost_table: &AtomicCostTable, weights: &CostWeights, ) -> OptimizerSolution { let mut deployed_configs: HashMap = HashMap::new(); @@ -32,19 +38,32 @@ pub fn greedy_assign( for aqe in aqes { let candidates = enumerate_candidates(&aqe, scrape_interval_ms); - let best = candidates + let (best, costs) = candidates .into_iter() - .map(|c| { - let cost = total_cost_rate(&aqe, &c, arrival_rate_hz, costs, weights); - (c, cost) + .filter_map(|c| { + // EXACT (config: None) always costs at the flat stub — it has + // no sketch_type/params for the table to key on. + let costs = match &c.config { + None => AtomicCosts::default(), + Some(cfg) => resolve_atomic_costs( + atomic_cost_table, + cfg.aggregation_type, + &cfg.parameters, + )?, + }; + let cost = total_cost_rate(&aqe, &c, arrival_rate_hz, &costs, weights); + Some((c, costs, cost)) }) // total_cmp (not partial_cmp().unwrap()) so a stray NaN cost can't panic. - .min_by(|(_, a), (_, b)| a.total_cmp(b)) - .map(|(c, _)| c) - .expect("enumerate_candidates always returns at least the EXACT fallback"); + .min_by(|(_, _, a), (_, _, b)| a.total_cmp(b)) + .map(|(c, costs, _)| (c, costs)) + .expect( + "enumerate_candidates always returns at least the EXACT fallback, \ + which always resolves (flat stub, no table lookup)", + ); - let ingest = ingest_cost(&best, arrival_rate_hz, costs, weights); - let query_rate = aqe.query_frequency_hz * query_cost(&aqe, &best, costs, weights); + let ingest = ingest_cost(&best, arrival_rate_hz, &costs, weights); + let query_rate = aqe.query_frequency_hz * query_cost(&aqe, &best, &costs, weights); let query_method = best.query_method.clone(); let aggregation_id = match best.config { @@ -121,7 +140,7 @@ mod tests { aqes, 60_000, 1.0, - &AtomicCosts::default(), + &AtomicCostTable::default(), &CostWeights::default(), ); @@ -155,7 +174,7 @@ mod tests { vec![aqe], 60_000, 1.0, - &AtomicCosts::default(), + &AtomicCostTable::default(), &CostWeights::default(), ); assert_eq!(solution.num_exact_fallback(), 1); diff --git a/asap-planner-rs/src/optimizer/mod.rs b/asap-planner-rs/src/optimizer/mod.rs index 534f4f6..2d1691d 100644 --- a/asap-planner-rs/src/optimizer/mod.rs +++ b/asap-planner-rs/src/optimizer/mod.rs @@ -1,4 +1,5 @@ pub mod aqe_extractor; +pub mod atomic_costs; pub mod candidate_gen; pub mod constants; pub mod cost_model; @@ -9,6 +10,9 @@ pub mod solution; pub mod translator; pub use aqe_extractor::{extract_aqes, RQE}; +pub use atomic_costs::{ + load_atomic_cost_table, resolve_atomic_costs, AtomicCostEntry, AtomicCostTable, +}; pub use candidate_gen::{enumerate_candidates, CandidateConfig}; pub use cost_model::{ingest_cost, query_cost, total_cost_rate, AtomicCosts, CostWeights}; pub use greedy::greedy_assign; diff --git a/asap-planner-rs/src/optimizer/pipeline.rs b/asap-planner-rs/src/optimizer/pipeline.rs index 0954764..627ea9a 100644 --- a/asap-planner-rs/src/optimizer/pipeline.rs +++ b/asap-planner-rs/src/optimizer/pipeline.rs @@ -5,7 +5,8 @@ use asap_types::PromQLSchema; use crate::config::input::ControllerConfig; use super::aqe_extractor::{extract_aqes, RQE}; -use super::cost_model::{AtomicCosts, CostWeights}; +use super::atomic_costs::AtomicCostTable; +use super::cost_model::CostWeights; use super::greedy::greedy_assign; use super::solution::{OptimizerSolution, AQE}; use super::translator::{translate, TranslationSummary}; @@ -73,13 +74,14 @@ pub fn run_greedy_pipeline( schema: &PromQLSchema, scrape_interval_ms: u64, arrival_rate_hz: f64, + atomic_cost_table: &AtomicCostTable, ) -> (StreamingConfig, InferenceConfig) { run_pipeline(config, schema, scrape_interval_ms, "greedy", |aqes| { greedy_assign( aqes, scrape_interval_ms, arrival_rate_hz, - &AtomicCosts::default(), + atomic_cost_table, &CostWeights::default(), ) }) @@ -145,7 +147,8 @@ mod tests { fn greedy_pipeline_deploys_a_config_for_a_mergeable_aqe() { let config = make_config(&[("min_over_time(metric[5m])", 60_000)]); let schema = PromQLSchema::new(); - let (streaming, inference) = run_greedy_pipeline(&config, &schema, 60_000, 1.0); + let (streaming, inference) = + run_greedy_pipeline(&config, &schema, 60_000, 1.0, &AtomicCostTable::default()); assert!(!streaming.get_all_aggregation_configs().is_empty()); assert!(!inference.query_configs.is_empty()); }