diff --git a/.design_docs/optimizer-v1-implementation-plan.md b/.design_docs/optimizer-v1-implementation-plan.md index 1344b78..8778ab4 100644 --- a/.design_docs/optimizer-v1-implementation-plan.md +++ b/.design_docs/optimizer-v1-implementation-plan.md @@ -270,24 +270,24 @@ Allow a config with labels ⊇ query labels to serve that AQE. This is what enab - ✅ Plug real `AtomicCosts` values into cost model — done for CMS/HLL/KLL, see 2f above and "Running with real sketch-bench costs" below. - ✅ `CountMinSketchWithHeap` wrapper in sketch-bench (`cms-heap` family — sketch-bench PR #104, - closed sketch-bench#95) — but not wired to ASAPQuery's cost lookup yet, see status below. + closed sketch-bench#95) — wired to ASAPQuery's temporary reference cost model below. - ~~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 below, not a sketch-bench sweep axis. -**Status as of 2026-08-28:** +**Status as of 2026-09-03:** 1. `export_atomic_costs.sh` doesn't sweep `cms-heap` yet. 2. Mapping gap: sketch-bench's `CmsHeapParams` is `{rows, cols}` only — `top_k` is a compile-time constant, Frequency-vs-TopK is two registry variants, not a param. ASAPQuery's grid sweeps `{depth, width, heapsize, count_events}` — `heapsize` has no home in sketch-bench's data at - all, confirming it must stay analytic. Unresolved: which of the 4 sketch-bench variants - (FastPath/RegularPath × Frequency/TopK) matches what ASAPQuery's `CountMinSketchWithHeap` - accumulator actually deploys, and what `count_events` maps to. -3. Once (2) is resolved: extend `sketch_bench_key()` in `atomic_costs.rs` to translate - `CountMinSketchWithHeap` → the right variant + `{rows,cols}`, for CPU costs only. -4. Analytic memory bound (`heap_size · avg_key_size`) still not implemented — needed regardless - of (1)–(3), memory was always meant to be analytic here, never a lookup. + all, confirming it must stay analytic. Resolved for the current runtime: use + `cms-heap-topk-regularpath-vector2d`; `count_events` is intentionally ignored because it + does not change the structural cost model. +3. ✅ `atomic_costs.rs` translates `CountMinSketchWithHeap` to the selected `{rows,cols}` + reference row and scales CPU costs linearly from the fixed top-k=32 benchmark. +4. ✅ Analytic memory bound: runtime CMS counters use 8-byte values, and heap memory uses + explicit average-key-size and per-entry-overhead assumptions. 5. sketch-bench#14 (HLL register-width memory bug) — still open, still un-fixed. 6. asap_sketchlib#69 (`subtract` unimplemented) — still open, `subtract_cpu_secs` absent by design. @@ -323,7 +323,8 @@ 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). `--atomic-costs` is -optional; omit it and every candidate costs at the flat stub, same as before #549. +optional; omit it and ordinary unbenchmarked candidates use the flat stub, while +CMS-with-heap candidates warn and are dropped until a matching reference row is available. ### Running with real sketch-bench costs @@ -350,9 +351,9 @@ cargo run -p asap_planner --bin asap-optimizer-cli -- \ --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). +`candidate-gen-dump`'s output labels each resolved params row `[real]` or `[stub]`; candidates +whose required cost row is missing are shown as `DROPPED`. CMS-with-heap uses its fixed-top-k +reference model when the `{rows,cols}` row is present and is dropped otherwise. Wire-in decision (deferred): once Phase 3 (MIP + feasibility + label superset matching) lands, swap `Controller::generate()` to call `run_mip_pipeline()` @@ -372,7 +373,7 @@ instead of `generator::generate_plan()`, likely behind an opt-in flag first. | `translator.rs` | `retention_count_for_assignment(Subtract)` | Returns hardcoded `1`; should be the actual checkpoint count needed to cover the full lookback | | `promql/generator.rs` | `generate_plan()` doc comment | Flags that `Controller::generate()` still uses the hardcoded path, not the optimizer — see "Offline Testing" section above | | — | Accuracy constraint | No `Error(a,g) ≤ ε_a` check exists anywhere; nothing stops picking an under-provisioned sketch (Phase 3d) | -| — | sketch-bench | No cardinality sweep, no `CountMinSketchWithHeap` wrapper; `AtomicCosts` are still stub numbers (Phase 3c) | +| — | sketch-bench | `export_atomic_costs.sh` does not yet sweep `cms-heap`; CMS-with-heap currently uses the temporary fixed-top-k reference model and must fail louder once swept coverage exists | --- diff --git a/asap-planner-rs/src/optimizer/atomic_costs.rs b/asap-planner-rs/src/optimizer/atomic_costs.rs index 2b48965..b7e0e8f 100644 --- a/asap-planner-rs/src/optimizer/atomic_costs.rs +++ b/asap-planner-rs/src/optimizer/atomic_costs.rs @@ -15,9 +15,14 @@ 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::constants::{ + CMS_HEAP_AVERAGE_KEY_BYTES, CMS_HEAP_COUNTER_BYTES, CMS_HEAP_ENTRY_OVERHEAD_BYTES, + CMS_HEAP_REFERENCE_HEAP_SIZE, EXACT_QUERY_CPU_SECS, SUBTRACT_CPU_SECS, +}; use super::cost_model::AtomicCosts; +const CMS_HEAP_BENCHMARK: &str = "cms-heap-topk-regularpath-vector2d"; + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct AtomicCostEntry { pub sketch: String, @@ -41,7 +46,7 @@ pub fn load_atomic_cost_table(path: &Path) -> anyhow::Result { /// 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). +/// sketch-bench has no wrapper for yet — 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 @@ -57,19 +62,6 @@ 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", @@ -103,14 +95,20 @@ fn sketch_bench_key( /// 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. +/// - `CountMinSketchWithHeap`: resolve through the temporary fixed-top-k +/// reference model in [`resolve_cms_heap_costs`]. Missing or malformed +/// reference data returns `None` and drops the candidate. +/// - Other benchmarked families, no matching row: `None` — drop the candidate, +/// per #524. pub fn resolve_atomic_costs( table: &AtomicCostTable, agg_type: AggregationType, params: &HashMap, ) -> Option { + if agg_type == AggregationType::CountMinSketchWithHeap { + return resolve_cms_heap_costs(table, params, &CmsHeapCostAssumptions::default()); + } + let Some((sketch, sketch_params)) = sketch_bench_key(agg_type, params) else { tracing::warn!( ?agg_type, @@ -133,6 +131,183 @@ pub fn resolve_atomic_costs( }) } +/// Temporary cost model for the runtime CMS-with-heap implementation. +/// +/// sketch-bench currently measures a fixed top-k=32 wrapper, while the +/// runtime's heap size is a candidate parameter. CPU costs therefore scale +/// linearly from the matching regular-path/top-k benchmark row. Memory is +/// computed from the runtime's i64 CMS counters and an explicit estimate for +/// each heap entry; the benchmark's i32-only matrix memory is not reused. +#[derive(Debug, Clone, Copy, PartialEq)] +struct CmsHeapCostAssumptions { + reference_heap_size: u64, + counter_bytes: f64, + average_key_bytes: f64, + heap_entry_overhead_bytes: f64, +} + +impl Default for CmsHeapCostAssumptions { + fn default() -> Self { + Self { + reference_heap_size: CMS_HEAP_REFERENCE_HEAP_SIZE, + counter_bytes: CMS_HEAP_COUNTER_BYTES, + average_key_bytes: CMS_HEAP_AVERAGE_KEY_BYTES, + heap_entry_overhead_bytes: CMS_HEAP_ENTRY_OVERHEAD_BYTES, + } + } +} + +impl CmsHeapCostAssumptions { + fn validate(self) { + assert!( + self.reference_heap_size > 0, + "CMS-with-heap reference_heap_size must be greater than zero" + ); + assert!( + self.counter_bytes.is_finite() && self.counter_bytes >= 0.0, + "CMS-with-heap counter_bytes must be finite and non-negative" + ); + assert!( + self.average_key_bytes.is_finite() && self.average_key_bytes >= 0.0, + "CMS-with-heap average_key_bytes must be finite and non-negative" + ); + assert!( + self.heap_entry_overhead_bytes.is_finite() && self.heap_entry_overhead_bytes >= 0.0, + "CMS-with-heap heap_entry_overhead_bytes must be finite and non-negative" + ); + } +} + +/// Resolve a CMS-with-heap candidate from the fixed-top-k benchmark reference. +/// +/// The helper deliberately returns `None` when the reference row is absent or +/// malformed. The caller then drops this candidate, leaving the always-feasible +/// EXACT candidate available. TODO(#651): turn these temporary warning paths +/// into hard errors once sketch-bench sweeps cover the candidate grid. +fn resolve_cms_heap_costs( + table: &AtomicCostTable, + params: &HashMap, + assumptions: &CmsHeapCostAssumptions, +) -> Option { + assumptions.validate(); + + let agg_type = AggregationType::CountMinSketchWithHeap; + let depth = require_u64(params, "depth", agg_type); + let width = require_u64(params, "width", agg_type); + let heap_size = require_u64(params, "heapsize", agg_type); + let count_events = require(params, "count_events", agg_type); + let heap_size_f64 = heap_size as f64; + let scale = heap_size_f64 / assumptions.reference_heap_size as f64; + let expected_config = serde_json::json!({ + "algorithm": CMS_HEAP_BENCHMARK, + "params": { "rows": depth, "cols": width }, + }); + + let Some(entry) = table + .iter() + .find(|entry| entry.sketch == CMS_HEAP_BENCHMARK && entry.sketch_config == expected_config) + else { + tracing::info!( + status = "missing_reference", + sketch = CMS_HEAP_BENCHMARK, + depth, + width, + heap_size, + count_events = ?count_events, + "cms-with-heap atomic cost measurement" + ); + tracing::warn!( + sketch = CMS_HEAP_BENCHMARK, + depth, + width, + heap_size, + count_events = ?count_events, + "no CMS-with-heap reference cost for candidate; dropping candidate; \ + TODO(#651): fail loudly once sketch-bench sweeps cover this grid" + ); + return None; + }; + + if !valid_cost_entry(entry) { + tracing::info!( + status = "invalid_reference", + sketch = CMS_HEAP_BENCHMARK, + depth, + width, + heap_size, + count_events = ?count_events, + "cms-with-heap atomic cost measurement" + ); + tracing::warn!( + sketch = CMS_HEAP_BENCHMARK, + depth, + width, + heap_size, + count_events = ?count_events, + "invalid CMS-with-heap reference cost for candidate; dropping candidate; \ + TODO(#651): fail loudly once sketch-bench sweeps cover this grid" + ); + return None; + } + + let costs = AtomicCosts { + mem_bytes_per_instance: depth as f64 * width as f64 * assumptions.counter_bytes + + heap_size_f64 + * (assumptions.average_key_bytes + assumptions.heap_entry_overhead_bytes), + insert_cpu_secs: entry.insert_cpu_secs * scale, + merge_cpu_secs: entry.merge_cpu_secs * scale, + subtract_cpu_secs: SUBTRACT_CPU_SECS, + query_cpu_secs: entry.query_cpu_secs * scale, + exact_query_cpu_secs: EXACT_QUERY_CPU_SECS, + }; + + tracing::info!( + status = "modeled", + sketch = CMS_HEAP_BENCHMARK, + depth, + width, + heap_size, + count_events = ?count_events, + mem_bytes_per_instance = costs.mem_bytes_per_instance, + insert_cpu_secs = costs.insert_cpu_secs, + merge_cpu_secs = costs.merge_cpu_secs, + query_cpu_secs = costs.query_cpu_secs, + "cms-with-heap atomic cost measurement" + ); + + Some(costs) +} + +fn require_u64(params: &HashMap, key: &str, agg_type: AggregationType) -> u64 { + require(params, key, agg_type) + .as_u64() + .unwrap_or_else(|| panic!("{agg_type:?} candidate has non-integer \"{key}\" param")) +} + +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; candidate_gen.rs's param_grid() \ + has drifted" + ) + }) +} + +fn valid_cost_entry(entry: &AtomicCostEntry) -> bool { + [ + entry.mem_bytes_per_instance, + entry.insert_cpu_secs, + entry.merge_cpu_secs, + entry.query_cpu_secs, + ] + .iter() + .all(|cost| cost.is_finite() && *cost >= 0.0) +} + #[cfg(test)] mod tests { use super::*; @@ -158,6 +333,34 @@ mod tests { ]) } + fn cms_heap_entry(depth: u64, width: u64) -> AtomicCostEntry { + AtomicCostEntry { + sketch: CMS_HEAP_BENCHMARK.into(), + sketch_config: serde_json::json!({ + "algorithm": CMS_HEAP_BENCHMARK, + "params": { "rows": depth, "cols": width } + }), + mem_bytes_per_instance: 1.0, + insert_cpu_secs: 2.0, + merge_cpu_secs: 4.0, + query_cpu_secs: 8.0, + } + } + + fn cms_heap_params( + depth: u64, + width: u64, + heap_size: u64, + count_events: bool, + ) -> HashMap { + HashMap::from([ + ("depth".to_string(), Value::from(depth)), + ("width".to_string(), Value::from(width)), + ("heapsize".to_string(), Value::from(heap_size)), + ("count_events".to_string(), Value::from(count_events)), + ]) + } + #[test] fn atomic_cost_entry_deserializes_sketch_benchs_documented_shape() { // Pinned against a real row sketch-bench's `atomic-costs` subcommand @@ -209,17 +412,81 @@ mod tests { } #[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. + fn cms_with_heap_without_reference_cost_drops_the_candidate() { + // Until sketch-bench has a matching reference row, CMS-with-heap must + // not inherit the flat stub: the optimizer should retain EXACT as its + // visible fallback instead of silently selecting an uncosted sketch. let table: AtomicCostTable = vec![]; - assert!(resolve_atomic_costs( - &table, - AggregationType::CountMinSketchWithHeap, - &HashMap::new() - ) - .is_some()); + let params = cms_heap_params(3, 1024, 40, true); + assert!( + resolve_atomic_costs(&table, AggregationType::CountMinSketchWithHeap, ¶ms) + .is_none() + ); + } + + #[test] + fn cms_with_heap_scales_cpu_and_models_runtime_memory() { + let table = vec![cms_heap_entry(3, 1024)]; + let assumptions = CmsHeapCostAssumptions { + reference_heap_size: 32, + counter_bytes: 8.0, + average_key_bytes: 10.0, + heap_entry_overhead_bytes: 6.0, + }; + let params = cms_heap_params(3, 1024, 64, true); + let costs = resolve_cms_heap_costs(&table, ¶ms, &assumptions) + .expect("matching CMS-with-heap reference row must resolve"); + + assert_eq!( + costs.mem_bytes_per_instance, + 3.0 * 1024.0 * 8.0 + 64.0 * 16.0 + ); + assert_eq!(costs.insert_cpu_secs, 4.0); + assert_eq!(costs.merge_cpu_secs, 8.0); + assert_eq!(costs.query_cpu_secs, 16.0); + assert_eq!(costs.subtract_cpu_secs, SUBTRACT_CPU_SECS); + assert_eq!(costs.exact_query_cpu_secs, EXACT_QUERY_CPU_SECS); + } + + #[test] + fn public_resolver_dispatches_cms_with_heap_to_the_reference_model() { + let table = vec![cms_heap_entry(3, 1024)]; + let params = cms_heap_params(3, 1024, 32, true); + let costs = resolve_atomic_costs(&table, AggregationType::CountMinSketchWithHeap, ¶ms) + .expect("public resolver must dispatch CMS-with-heap candidates"); + + assert_eq!( + costs.mem_bytes_per_instance, + 3.0 * 1024.0 * 8.0 + 32.0 * 64.0 + ); + assert_eq!(costs.insert_cpu_secs, 2.0); + assert_eq!(costs.merge_cpu_secs, 4.0); + assert_eq!(costs.query_cpu_secs, 8.0); + } + + #[test] + fn cms_with_heap_costs_ignore_count_events() { + let table = vec![cms_heap_entry(3, 1024)]; + let assumptions = CmsHeapCostAssumptions::default(); + let count_params = cms_heap_params(3, 1024, 40, true); + let value_params = cms_heap_params(3, 1024, 40, false); + + assert_eq!( + resolve_cms_heap_costs(&table, &count_params, &assumptions), + resolve_cms_heap_costs(&table, &value_params, &assumptions) + ); + } + + #[test] + #[should_panic(expected = "reference_heap_size must be greater than zero")] + fn cms_with_heap_rejects_invalid_assumptions() { + let table = vec![cms_heap_entry(3, 1024)]; + let params = cms_heap_params(3, 1024, 40, true); + let assumptions = CmsHeapCostAssumptions { + reference_heap_size: 0, + ..CmsHeapCostAssumptions::default() + }; + resolve_cms_heap_costs(&table, ¶ms, &assumptions); } #[test] @@ -229,8 +496,8 @@ mod tests { // 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). + // the CMS-with-heap resolver handles separately 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); diff --git a/asap-planner-rs/src/optimizer/constants.rs b/asap-planner-rs/src/optimizer/constants.rs index cd4bd0c..5ca899d 100644 --- a/asap-planner-rs/src/optimizer/constants.rs +++ b/asap-planner-rs/src/optimizer/constants.rs @@ -5,6 +5,13 @@ pub const CMS_DEPTHS: &[u64] = &[3, 5]; pub const CMS_WIDTHS: &[u64] = &[512, 1024, 2048]; pub const CMS_HEAP_SIZES: &[u64] = &[40, 200, 1000]; +// Temporary CMS-with-heap cost-model assumptions. The reference CPU costs +// come from the fixed-top-k sketch-bench wrapper; these values describe the +// runtime representation until sketch-bench sweeps the actual heap sizes. +pub const CMS_HEAP_REFERENCE_HEAP_SIZE: u64 = 32; +pub const CMS_HEAP_COUNTER_BYTES: f64 = 8.0; +pub const CMS_HEAP_AVERAGE_KEY_BYTES: f64 = 32.0; +pub const CMS_HEAP_ENTRY_OVERHEAD_BYTES: f64 = 32.0; pub const KLL_KS: &[u64] = &[200, 500]; pub const HYDRA_ROWS: &[u64] = &[3, 5]; pub const HYDRA_COLS: &[u64] = &[512, 1024]; diff --git a/asap-planner-rs/src/optimizer/greedy.rs b/asap-planner-rs/src/optimizer/greedy.rs index 086bfd7..8be8091 100644 --- a/asap-planner-rs/src/optimizer/greedy.rs +++ b/asap-planner-rs/src/optimizer/greedy.rs @@ -87,9 +87,10 @@ pub fn greedy_assign( #[cfg(test)] mod tests { use super::*; + use crate::optimizer::atomic_costs::AtomicCostEntry; use asap_types::query_requirements::QueryRequirements; use promql_utilities::data_model::KeyByLabelNames; - use promql_utilities::query_logics::enums::Statistic; + use promql_utilities::query_logics::enums::{AggregationType, Statistic}; use std::collections::HashMap as StdHashMap; fn make_aqe(stat: Statistic, range_ms: u64, min_t: u64, freq_hz: f64) -> AQE { @@ -159,4 +160,54 @@ mod tests { assert_eq!(solution.num_exact_fallback(), 1); assert!(solution.deployed_configs().is_empty()); } + + #[test] + fn missing_cms_with_heap_reference_cost_falls_back_to_exact() { + // Regression coverage for #651: an uncosted CMS-with-heap candidate + // must be dropped rather than inheriting the flat stub and winning. + let solution = greedy_assign( + vec![make_aqe(Statistic::Topk, 60_000, 60_000, 1.0 / 60.0)], + 60_000, + 1.0, + &AtomicCostTable::default(), + &CostWeights::default(), + ); + + assert_eq!(solution.num_exact_fallback(), 1); + assert!(solution.deployed_configs().is_empty()); + } + + #[test] + fn matching_cms_with_heap_reference_cost_can_be_deployed() { + let table = vec![AtomicCostEntry { + sketch: "cms-heap-topk-regularpath-vector2d".into(), + sketch_config: serde_json::json!({ + "algorithm": "cms-heap-topk-regularpath-vector2d", + "params": { "rows": 3, "cols": 512 } + }), + mem_bytes_per_instance: 1.0, + insert_cpu_secs: 0.0, + merge_cpu_secs: 0.0, + query_cpu_secs: 0.0, + }]; + let solution = greedy_assign( + vec![make_aqe(Statistic::Topk, 60_000, 60_000, 1.0 / 60.0)], + 60_000, + 1.0, + &table, + &CostWeights::default(), + ); + + assert_eq!(solution.num_exact_fallback(), 0); + assert_eq!(solution.deployed_configs().len(), 1); + assert_eq!( + solution + .deployed_configs() + .values() + .next() + .expect("one CMS-with-heap config deployed") + .aggregation_type, + AggregationType::CountMinSketchWithHeap + ); + } }