feat(planner): cost candidates per (sketch_type, params) from sketch-bench's table

.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<String, Value>) -> Option<AtomicCosts>
+```
+
+- `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 <path/to/workload.yaml> \
-  --prometheus_scrape_interval 60 \
-  [--rho 1.0]
+  --data-ingestion-interval-ms 60000 \
+  [--rho 1.0] \
+  [--atomic-costs <path/to/atomic_costs.json>]
 ```
 
 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()`

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<PathBuf>,
 }
 
 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<WindowKey>, Vec<Vec<(String, Value)>>)> =
-        HashMap::new();
+    let mut groups: HashMap<
+        (String, String),
+        (AggregationType, Vec<WindowKey>, Vec<Vec<(String, Value)>>),
+    > = 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(&params) {
-            entry.1.push(params);
+        if !entry.2.contains(&params) {
+            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,11 +187,39 @@ 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<String, Value> =
+                    p.iter().map(|(k, v)| (k.clone(), v.clone())).collect();
+                match resolve_atomic_costs(atomic_cost_table, *agg_type, &param_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(", "));
         }
     }
 
     if has_exact {
         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,
+    )
+}

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<PathBuf>,
+
     #[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());