feat: use scenario-matched empirical costs in the optimizer

.design_docs/optimizer-v1-implementation-plan.md

@@ -322,7 +322,8 @@ cargo run -p asap_planner --bin asap-optimizer-cli -- \
   --dataset <path/to/series-inventory.csv> \
   --data-ingestion-interval-ms 60000 \
   [--rho 1.0] \
-  [--atomic-costs <path/to/atomic_costs.json>]
+  [--atomic-costs <path/to/atomic_costs.json> \
+   --atomic-cost-workload <path/to/workload-selector.json>]
 ```
 
 Takes the same `ControllerConfig` YAML format as `asap-planner --input_config`.
@@ -332,7 +333,11 @@ group count; no live Prometheus connection is needed. A `metrics:` hints block,
 present, is checked against the dataset and mismatches fail loudly.
 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 ordinary unbenchmarked candidates use the flat stub, while
+optional; when supplied it requires `--atomic-cost-workload`, a JSON file
+containing the exact `profiles[].workload` value from that benchmark artifact.
+The loader validates the document schema and rejects a selector that matches
+zero or multiple profiles; it never mixes entries across workloads. Omit both
+flags 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
@@ -352,13 +357,15 @@ CMS-with-heap candidates warn and are dropped until a matching reference row is
 # 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
+  --atomic-costs path/to/atomic_costs.json \
+  --atomic-cost-workload path/to/workload-selector.json
 
 # Run the actual optimizer:
 cargo run -p asap_planner --bin asap-optimizer-cli -- \
   --input_config workload.yaml --data-ingestion-interval-ms 60000 \
   --dataset path/to/series-inventory.csv \
-  --atomic-costs path/to/atomic_costs.json
+  --atomic-costs path/to/atomic_costs.json \
+  --atomic-cost-workload path/to/workload-selector.json
 ```
 
 `candidate-gen-dump`'s output labels each resolved params row `[real]` or `[stub]`; candidates

CONTEXT.md

@@ -0,0 +1,24 @@
+# ASAPQuery
+
+Terms used by the cost-based sketch optimizer and its benchmarking pipeline.
+
+## Language
+
+**Grouping labels**:
+PromQL `GROUP BY` labels that partition a metric stream into label groups; a
+non-subpopulation sketch is instantiated once per distinct group.
+_Avoid_: keying labels
+
+**Sketch key**:
+The value or label tuple inserted into and queried from a keyed sketch such as
+HLL, CMS, or Hydra. Its distribution, cardinality, and encoded size may affect
+the measured atomic cost and accuracy.
+_Avoid_: grouping label, partition key
+
+**Benchmark profile**:
+A reproducible description of the input trace slice used to measure an
+atomic-cost table for a sketch family and configuration.
+
+**Planner context**:
+Query- and deployment-specific structural inputs, such as label-group count
+and retained-window count, that scale atomic costs into a plan cost.

asap-planner-rs/src/bin/candidate_gen_dump.rs

@@ -6,8 +6,8 @@ use std::path::PathBuf;
 
 use asap_planner::{
     optimizer::{
-        enumerate_candidates, extract_aqes, load_atomic_cost_table, resolve_atomic_costs,
-        AtomicCostTable, AtomicCosts, CandidateConfig, RQE,
+        enumerate_candidates, extract_aqes, load_optional_selected_atomic_cost_table,
+        resolve_atomic_costs, AtomicCostTable, AtomicCosts, CandidateConfig, RQE,
     },
     ControllerConfig,
 };
@@ -28,21 +28,27 @@ struct Args {
     #[arg(long = "data-ingestion-interval-ms")]
     scrape_interval_ms: u64,
 
-    /// Path to sketch-bench's exported atomic-cost table (see ASAPQuery#524).
+    /// Path to sketch-bench's versioned atomic-cost document.
     /// 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>,
+
+    /// JSON object containing the selected profile's `workload` and `scenario`
+    /// values, selecting exactly one measured profile.
+    #[arg(long = "atomic-cost-profile", requires = "atomic_costs")]
+    atomic_cost_profile: 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 atomic_cost_table = load_optional_selected_atomic_cost_table(
+        args.atomic_costs.as_deref(),
+        args.atomic_cost_profile.as_deref(),
+    )?
+    .unwrap_or_default();
 
     let yaml_str = std::fs::read_to_string(&args.input_config)?;
     let config: ControllerConfig = serde_yaml::from_str(&yaml_str)?;
@@ -55,6 +61,10 @@ fn main() -> anyhow::Result<()> {
             qg.queries.iter().map(|q| RQE {
                 query_string: q.clone(),
                 t_repeat_ms: qg.repetition_delay_ms,
+                max_mean_rank_error: (qg.controller_options.accuracy_sla > 0.0)
+                    .then_some(1.0 - qg.controller_options.accuracy_sla),
+                max_atomic_query_cpu_secs: (qg.controller_options.latency_sla > 0.0)
+                    .then_some(qg.controller_options.latency_sla),
             })
         })
         .collect();

asap-planner-rs/src/bin/optimizer_cli.rs

@@ -7,7 +7,7 @@
 use std::path::PathBuf;
 
 use asap_planner::optimizer::{
-    load_atomic_cost_table, run_greedy_pipeline, AtomicCostTable, SeriesDataset,
+    load_optional_selected_atomic_cost_table, run_greedy_pipeline, AtomicCostTable, SeriesDataset,
 };
 use asap_planner::ControllerConfig;
 use clap::Parser;
@@ -35,14 +35,20 @@ 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
+    /// Path to the versioned atomic-cost document sketch-bench's `atomic-costs`
+    /// subcommand exports. Requires --atomic-cost-profile to select exactly
+    /// one measured workload profile. 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>,
 
+    /// JSON object containing the selected profile's `workload` and `scenario`
+    /// values. This prevents use of costs measured on a different wrangled CSV.
+    #[arg(long = "atomic-cost-profile", requires = "atomic_costs")]
+    atomic_cost_profile: Option<PathBuf>,
+
     #[arg(short, long, action = clap::ArgAction::Count)]
     verbose: u8,
 }
@@ -70,8 +76,11 @@ fn main() -> anyhow::Result<()> {
     let config: ControllerConfig = serde_yaml::from_str(&yaml_str)?;
     let dataset = SeriesDataset::from_path(&args.dataset)?;
 
-    let atomic_cost_table = match &args.atomic_costs {
-        Some(path) => load_atomic_cost_table(path)?,
+    let atomic_cost_table = match load_optional_selected_atomic_cost_table(
+        args.atomic_costs.as_deref(),
+        args.atomic_cost_profile.as_deref(),
+    )? {
+        Some(table) => table,
         None => {
             tracing::warn!(
                 "no --atomic-costs supplied; CMS/HLL/KLL candidates will never be selected"

asap-planner-rs/src/config/input.rs

@@ -80,6 +80,7 @@ pub struct QueryGroup {
 #[derive(Debug, Clone, Deserialize, Default)]
 pub struct ControllerOptions {
     pub accuracy_sla: f64,
+    /// Maximum atomic CPU seconds allowed to answer one request.
     pub latency_sla: f64,
 }
 

asap-planner-rs/src/optimizer/aqe_extractor.rs

@@ -17,6 +17,8 @@ use super::solution::AQE;
 pub struct RQE {
     pub query_string: String,
     pub t_repeat_ms: u64,
+    pub max_mean_rank_error: Option<f64>,
+    pub max_atomic_query_cpu_secs: Option<f64>,
 }
 
 /// Stable deduplication key for an AQE.
@@ -33,6 +35,16 @@ struct AQEKey {
     topk_count_events: Option<bool>,
 }
 
+struct AQEAccumulator {
+    requirements: QueryRequirements,
+    query_strings: Vec<String>,
+    query_frequency_hz: f64,
+    min_t_repeat_ms: u64,
+    t_repeat_gcd_ms: u64,
+    max_mean_rank_error: Option<f64>,
+    max_atomic_query_cpu_secs: Option<f64>,
+}
+
 impl AQEKey {
     fn from_requirements(req: &QueryRequirements) -> Self {
         Self {
@@ -66,7 +78,7 @@ pub fn extract_aqes(
     scrape_interval_ms: u64,
 ) -> Vec<AQE> {
     // (key) -> (requirements, query_strings, sum_freq, min_t, gcd_t)
-    let mut acc: HashMap<AQEKey, (QueryRequirements, Vec<String>, f64, u64, u64)> = HashMap::new();
+    let mut acc: HashMap<AQEKey, AQEAccumulator> = HashMap::new();
 
     for rqe in rqes {
         if rqe.t_repeat_ms == 0 {
@@ -84,20 +96,42 @@ pub fn extract_aqes(
             match extract_requirements(&leaf, metric_schema, scrape_interval_ms) {
                 Some(req) => {
                     let key = AQEKey::from_requirements(&req);
-                    let entry = acc
-                        .entry(key)
-                        .or_insert_with(|| (req, Vec::new(), 0.0, u64::MAX, 0));
-                    if !entry.1.contains(&leaf) {
-                        entry.1.push(leaf);
+                    let entry = acc.entry(key).or_insert_with(|| AQEAccumulator {
+                        requirements: req,
+                        query_strings: Vec::new(),
+                        query_frequency_hz: 0.0,
+                        min_t_repeat_ms: u64::MAX,
+                        t_repeat_gcd_ms: 0,
+                        max_mean_rank_error: rqe.max_mean_rank_error,
+                        max_atomic_query_cpu_secs: rqe.max_atomic_query_cpu_secs,
+                    });
+                    if !entry.query_strings.contains(&leaf) {
+                        entry.query_strings.push(leaf);
                     }
                     // query_frequency_hz must stay in Hz (queries per real second)
                     // regardless of t_repeat_ms's internal unit — 1000.0 / ms, not 1.0 / ms.
-                    entry.2 += 1000.0 / rqe.t_repeat_ms as f64;
-                    entry.3 = entry.3.min(rqe.t_repeat_ms);
-                    entry.4 = if entry.4 == 0 {
+                    entry.query_frequency_hz += 1000.0 / rqe.t_repeat_ms as f64;
+                    entry.min_t_repeat_ms = entry.min_t_repeat_ms.min(rqe.t_repeat_ms);
+                    entry.t_repeat_gcd_ms = if entry.t_repeat_gcd_ms == 0 {
                         rqe.t_repeat_ms
                     } else {
-                        gcd(entry.4, rqe.t_repeat_ms)
+                        gcd(entry.t_repeat_gcd_ms, rqe.t_repeat_ms)
+                    };
+                    entry.max_mean_rank_error =
+                        match (entry.max_mean_rank_error, rqe.max_mean_rank_error) {
+                            (Some(a), Some(b)) => Some(a.min(b)),
+                            (Some(a), None) => Some(a),
+                            (None, Some(b)) => Some(b),
+                            (None, None) => None,
+                        };
+                    entry.max_atomic_query_cpu_secs = match (
+                        entry.max_atomic_query_cpu_secs,
+                        rqe.max_atomic_query_cpu_secs,
+                    ) {
+                        (Some(a), Some(b)) => Some(a.min(b)),
+                        (Some(a), None) => Some(a),
+                        (None, Some(b)) => Some(b),
+                        (None, None) => None,
                     };
                 }
                 None => {
@@ -111,21 +145,15 @@ pub fn extract_aqes(
     }
 
     acc.into_values()
-        .map(
-            |(
-                requirements,
-                query_strings,
-                query_frequency_hz,
-                min_t_repeat_ms,
-                t_repeat_gcd_ms,
-            )| AQE {
-                requirements,
-                query_strings,
-                query_frequency_hz,
-                min_t_repeat_ms,
-                t_repeat_gcd_ms,
-            },
-        )
+        .map(|accumulator| AQE {
+            requirements: accumulator.requirements,
+            query_strings: accumulator.query_strings,
+            query_frequency_hz: accumulator.query_frequency_hz,
+            min_t_repeat_ms: accumulator.min_t_repeat_ms,
+            t_repeat_gcd_ms: accumulator.t_repeat_gcd_ms,
+            max_mean_rank_error: accumulator.max_mean_rank_error,
+            max_atomic_query_cpu_secs: accumulator.max_atomic_query_cpu_secs,
+        })
         .collect()
 }
 
@@ -199,6 +227,8 @@ mod tests {
         RQE {
             query_string: query.to_string(),
             t_repeat_ms: t_ms,
+            max_mean_rank_error: None,
+            max_atomic_query_cpu_secs: None,
         }
     }
 
@@ -246,6 +276,19 @@ mod tests {
         assert_eq!(aqes[0].query_strings.len(), 1); // same string, deduplicated
     }
 
+    #[test]
+    fn deduplicated_aqe_keeps_the_strictest_rank_error_limit() {
+        let mut loose = rqe("quantile_over_time(0.99, metric[5m])", 60_000);
+        loose.max_mean_rank_error = Some(0.02);
+        let mut strict = rqe("quantile_over_time(0.99, metric[5m])", 30_000);
+        strict.max_mean_rank_error = Some(0.01);
+
+        let aqes = extract_aqes(&[loose, strict], &empty_schema(), 15_000);
+
+        assert_eq!(aqes.len(), 1);
+        assert_eq!(aqes[0].max_mean_rank_error, Some(0.01));
+    }
+
     #[test]
     fn unsupported_query_is_skipped() {
         let rqes = vec![rqe("not_a_real_function(metric[5m])", 60_000)];