rqe-optimizer: read sketch accuracy from saturation curves

aqpbm-core/src/atomic_costs.rs

@@ -72,6 +72,36 @@ pub struct MeasuredAt {
     pub distribution: Option<DataDistribution>,
 }
 
+impl MeasuredAt {
+    /// The data shape the row was measured at, keyed the way the saturation
+    /// study keys its grid points, or `None` for a distribution the study
+    /// doesn't sweep. A single-column workload's Zipf population is its `K`.
+    pub fn data_shape(&self) -> Option<MeasuredShape> {
+        match self.distribution.as_ref()? {
+            DataDistribution::Zipf(p) => Some(MeasuredShape::Zipf {
+                skew: p.skewness,
+                keys: p.population_size as f64,
+            }),
+            DataDistribution::Pareto(p) => Some(MeasuredShape::Pareto {
+                tail_index: p.alpha,
+            }),
+            DataDistribution::Uniform(_) | DataDistribution::Normal(_) => None,
+        }
+    }
+}
+
+/// The data parameters a sketch's accuracy was measured at. A cost-table
+/// row ([`MeasuredAt::data_shape`]) and a saturation-study grid point share
+/// this key, so a row is one point on that grid point's curve.
+#[derive(Debug, Clone, Copy, PartialEq)]
+pub enum MeasuredShape {
+    /// Keys drawn Zipf(`skew`) over `keys` distinct keys (frequency, top-k,
+    /// cardinality).
+    Zipf { skew: f64, keys: f64 },
+    /// Values drawn Pareto with tail index `tail_index` (quantiles).
+    Pareto { tail_index: f64 },
+}
+
 pub type AtomicCostTable = Vec<AtomicCostEntry>;
 
 /// Why one [`MergedRecord`] didn't produce a row. Not an error: the caller

aqpbm-core/src/lib.rs

@@ -20,7 +20,8 @@ pub use aqpbm_datagen::{
     ZipfParameter, RULE_MONOTONIC_INCREASE, RULE_NONE,
 };
 pub use atomic_costs::{
-    reduce_all, reduce_one, same_cell, AtomicCostEntry, AtomicCostTable, MeasuredAt, SkipReason,
+    reduce_all, reduce_one, same_cell, AtomicCostEntry, AtomicCostTable, MeasuredAt, MeasuredShape,
+    SkipReason,
 };
 pub use benchmark_result::{
     BenchReport, BenchSection, CpuTime, ExternalWorkload, InsertMetrics, LatencySummary,

docs/rqe_sketch_deployment_v1.md

@@ -76,7 +76,11 @@ from sketch-bench runs, gives for each configuration:
 - CPU per insert;
 - CPU per query of one instance;
 - CPU per pairwise merge; and
-- measured accuracy values.
+- measured accuracy values, which only exact accumulators use (see
+  [eligibility](#candidate-deployments)).
+
+Each metric's facts also carry a fitted `data_shape` per grouping, which keys
+the saturation curves.
 
 Families are named by sketch-bench variant, not algorithm, because variants
 of the same algorithm can have different costs.
@@ -199,11 +203,37 @@ An RAQE `r_i` is eligible for a candidate `D` when:
 
 1. `cap_i = D.capability`, `metric_i = D.metric` and `G_i = D.G`.
 2. `D.x % D.y = 0`, `S_i % D.x = 0`, and `T_i % D.y = 0`.
-3. `D.configuration` has a measured `accuracy_metric_i` that clears `tol_i` in
-   `direction_i` at both measured merge counts bracketing the query's
-   `S_i / D.x` windows (count 1 is the single-instance accuracy; beyond the
-   largest measured count, the largest). Merge accuracy is not monotone in the
-   count, so neither bracket alone is conservative.
+3. The accuracy of `D` for `r_i` in `accuracy_metric_i` clears `tol_i` in
+   `direction_i`.
+
+Exact accumulators take that accuracy from the cost table, at both measured
+merge counts bracketing the query's `S_i / D.x` windows (count 1 is the
+single-instance accuracy; beyond the largest measured count, the largest).
+Merge accuracy is not monotone in the count, so the worse bracket is used.
+Sketches take it
+from the saturation study's error-vs-N curves (`SaturationCurves`, #156),
+read at the number of items one group receives over the RAQE's whole
+lookback: series per group times scrapes per lookback.
+
+- The curve is the configuration's at the metric's fitted `data_shape` for
+  `G` (zipf θ and keys `K`, or tail index `a` for quantiles). Between grid
+  points, the worst bracketing point; outside the grid, or with no fit, no
+  accuracy.
+- Between checkpoints, the worse neighbour. Below the first checkpoint, no
+  accuracy. Past the last, the plateau if the point saturated, else none.
+- The curve's `error_metric` must be `accuracy_metric_i`.
+- The window size doesn't matter: a merged answer reads the curve at the
+  lookback's item count, like a single sketch. KLL and top-k merge lossily;
+  their merge penalty is #158.
+
+The cost table's sketch accuracies are not read, but they are one point on
+each curve: a row's `measured_at` (items, Zipf θ and population, or Pareto
+`a`) is a grid point's key. `SaturationCurves::check_cost_table` reads the
+curve there and reports every row that disagrees (beyond 3 seed standard
+errors plus 5%), whose configuration is not in the grid, or whose table lacks
+the curve's `error_metric`. Rows the study has no point for (KLL: the table
+measures Zipf ranks, the study Pareto values) are reported as unchecked.
+`small_problem` refuses to plan when any row disagrees.
 
 For lower-is-better metrics, passing means `measured <= tol_i`. For
 higher-is-better metrics, passing means `measured >= tol_i`. Direction is
@@ -435,11 +465,8 @@ each phase's CPU and memory, together with the selected deployment mapping.
 
 ## v1 scope and TODOs
 
-- **Accuracy after merging:** measured at 4, 16, 64, 256 and 1024 merged
-  shards of one fixed stream (eligibility rule 3). Queries folding more than
-  1024 windows read the 1024 measurement, and the stream size is the
-  benchmark's, not `λ · S_i / card(G)`; the size sweep in sketch-bench#147
-  covers that.
+- **Accuracy after merging:** every merge reads the curve at the merged item
+  count. KLL and top-k don't model their merge penalty yet (#158).
 - **Query-result sharing:** v1 charges every RAQE its own query and merge CPU.
   Revisit when RAQE semantics and execution timing identify safe reuse cases.
 - **Latency SLAs:** the MILP takes optional per-RAQE latency bounds; the

rqe-optimizer/examples/small_problem.rs

@@ -13,6 +13,11 @@
 //! cardinalities and scrape intervals) lives in the two tables below
 //! (`facts`, `raqes`) and nowhere else in this file.
 //!
+//! Sketch accuracy comes from the saturation study's curves, not the cost
+//! table (#156): pass `--saturation-dir DIR` (required), the directory
+//! holding `out_grid_1e7_cost/` and `out_1e9/` from
+//! `scripts/study_saturation.py`.
+//!
 //! Run: `scripts/export_rqe_optimizer_costs.sh DIR` once, then pass
 //! `--cost-dir DIR` (required) to this example. Do not run this example
 //! with no mode flag on a large workload: eager mode retains every feasible
@@ -39,9 +44,10 @@ use rqe_optimizer::candidates::{
 use rqe_optimizer::enumerate::{brute_force, for_each_mapping, for_each_mapping_while, unservable};
 use rqe_optimizer::milp::{minimize, Objective};
 use rqe_optimizer::pareto::{pareto_front, ParetoFront};
+use rqe_optimizer::saturation::{DataShape, SaturationCurves};
 use rqe_optimizer::{
-    validate_facts, AccuracyDirection, AtomicCostTable, Capability, LabelSet, MetricFacts, Raqe,
-    WorkloadFacts,
+    validate_facts, AccuracyDirection, AtomicCostTable, Capability, Deployment, LabelSet,
+    MetricFacts, Raqe, WorkloadFacts,
 };
 
 /// Accuracy metric keys the real comparators actually report (checked
@@ -68,19 +74,34 @@ fn label_set(names: &[&str]) -> LabelSet {
     names.iter().map(|s| s.to_string()).collect()
 }
 
-fn load_cost_table() -> AtomicCostTable {
+/// The directory after `flag`; panics with `what` when it's absent.
+fn required_dir(flag: &str, what: &str) -> String {
     let args: Vec<_> = std::env::args().collect();
-    let cost_dir = args
-        .iter()
-        .position(|arg| arg == "--cost-dir")
+    args.iter()
+        .position(|arg| arg == flag)
         .map(|index| {
             args.get(index + 1)
-                .unwrap_or_else(|| panic!("--cost-dir requires a directory"))
+                .unwrap_or_else(|| panic!("{flag} requires a directory"))
+                .clone()
         })
-        .unwrap_or_else(|| {
-            panic!("--cost-dir DIR is required (a directory written by scripts/export_rqe_optimizer_costs.sh)")
-        });
-    let path = std::path::Path::new(cost_dir).join(COST_TABLE_FILE);
+        .unwrap_or_else(|| panic!("{flag} DIR is required ({what})"))
+}
+
+fn load_saturation_curves() -> SaturationCurves {
+    let dir = required_dir(
+        "--saturation-dir",
+        "holding out_grid_1e7_cost/ and out_1e9/ from scripts/study_saturation.py",
+    );
+    SaturationCurves::load(std::path::Path::new(&dir))
+        .unwrap_or_else(|e| panic!("couldn't load saturation curves: {e}"))
+}
+
+fn load_cost_table() -> AtomicCostTable {
+    let cost_dir = required_dir(
+        "--cost-dir",
+        "a directory written by scripts/export_rqe_optimizer_costs.sh",
+    );
+    let path = std::path::Path::new(&cost_dir).join(COST_TABLE_FILE);
     let raw = std::fs::read_to_string(&path).unwrap_or_else(|e| {
         panic!(
             "couldn't read {} ({e}) -- run \
@@ -94,7 +115,9 @@ fn load_cost_table() -> AtomicCostTable {
 
 /// Both metrics: 5 services × 10 endpoints × 600 pods = 30,000 series
 /// scraped every 15 s, so `λ` = 2,000 samples/sec each. Durations range from
-/// 1 ms to 60 s, which sizes DDSketch.
+/// 1 ms to 60 s, which sizes DDSketch. The data
+/// shapes stand in for fits over a real dataset; the worst group's keys
+/// exceed the average group's pods.
 fn facts() -> WorkloadFacts {
     let http_metric_facts = |value_range| MetricFacts {
         labels: label_set(&["service", "endpoint", "pod"]),
@@ -106,6 +129,11 @@ fn facts() -> WorkloadFacts {
         ]
         .into(),
         value_range,
+        data_shape: [
+            (label_set(&["service"]), data_shape(10_000.0)),
+            (label_set(&["service", "endpoint"]), data_shape(1_000.0)),
+        ]
+        .into(),
     };
     [
         (REQUESTS.to_string(), http_metric_facts(None)),
@@ -114,6 +142,14 @@ fn facts() -> WorkloadFacts {
     .into()
 }
 
+fn data_shape(distinct_keys: f64) -> DataShape {
+    DataShape {
+        zipf_s: 1.1,
+        distinct_keys,
+        tail_index: 1.5,
+    }
+}
+
 fn raqes() -> Vec<Raqe> {
     let se = label_set(&["service", "endpoint"]);
     let s = label_set(&["service"]);
@@ -249,7 +285,7 @@ fn print_mapping(
     mapping_number: u64,
     mapping: &[usize],
     raqes: &[Raqe],
-    deployments: &[rqe_optimizer::Deployment],
+    deployments: &[Deployment],
 ) {
     println!("mapping {mapping_number}:");
     for (raqe, &deployment_index) in raqes.iter().zip(mapping) {
@@ -357,6 +393,7 @@ fn reject_unknown_args() {
         "--print-first",
         "--progress-every",
         "--sample-mappings",
+        "--saturation-dir",
         "--w-cpu",
         "--w-mem",
     ];
@@ -376,16 +413,29 @@ fn main() {
     let mut raqes = raqes();
     apply_latency_slas(&mut raqes);
     let cost_table = load_cost_table();
+    let curves = load_saturation_curves();
+    let check = curves.check_cost_table(&cost_table);
+    for row in &check.unchecked {
+        eprintln!("cost table vs. saturation curves, unchecked: {row}");
+    }
+    if !check.mismatched.is_empty() {
+        panic!(
+            "the cost table disagrees with the saturation curves: {:#?}",
+            check.mismatched
+        );
+    }
     let facts = facts();
     if let Err(problems) = validate_facts(&raqes, &facts) {
         panic!("invalid workload facts: {problems:#?}");
     }
+    let accuracy = |raqe: &Raqe, deployment: &Deployment| curves.accuracy(raqe, deployment, &facts);
 
     let allow_undeployable = std::env::args().any(|arg| arg == "--allow-undeployable-families");
     let unpruned_count = candidates_only.then(|| {
         build_all_candidates_unpruned(&raqes, &cost_table, &facts, allow_undeployable).len()
     });
-    let deployments = build_all_candidates(&raqes, &cost_table, &facts, allow_undeployable);
+    let deployments =
+        build_all_candidates(&raqes, &cost_table, &facts, allow_undeployable, &accuracy);
     println!(
         "{} RAQEs, {} candidate deployments (from {} real cost-table rows)",
         raqes.len(),
@@ -401,7 +451,7 @@ fn main() {
         );
         let mut possible_mappings = Some(1_u64);
         for raqe in &raqes {
-            let eligible = eligible_deployments_for(raqe, &deployments, &facts);
+            let eligible = eligible_deployments_for(raqe, &deployments, &facts, &accuracy);
             println!("  {}: {} eligible deployments", raqe.id, eligible.len());
             possible_mappings = possible_mappings.and_then(|count| {
                 count.checked_mul(eligible.len().try_into().expect("usize fits in u64"))
@@ -422,7 +472,7 @@ fn main() {
         return;
     }
 
-    let missing = unservable(&raqes, &deployments, &facts);
+    let missing = unservable(&raqes, &deployments, &facts, &accuracy);
     if !missing.is_empty() {
         println!("unservable (no eligible deployment): {missing:?}");
         return;
@@ -431,7 +481,7 @@ fn main() {
     let sample_mappings = positive_integer_flag("--sample-mappings", 0);
     if sample_mappings > 0 {
         let mut printed = 0;
-        let result = for_each_mapping_while(&raqes, &deployments, &facts, |mapping| {
+        let result = for_each_mapping_while(&raqes, &deployments, &facts, &accuracy, |mapping| {
             printed += 1;
             print_mapping(printed, mapping, &raqes, &deployments);
             printed < sample_mappings
@@ -453,7 +503,7 @@ fn main() {
             w_cpu: weight_flag("--w-cpu", 1.0),
             w_mem: weight_flag("--w-mem", 0.0),
         };
-        let solution = minimize(&raqes, &deployments, &facts, objective)
+        let solution = minimize(&raqes, &deployments, &facts, objective, &accuracy)
             .expect("small_problem MILP should be feasible");
         println!(
             "MILP solution for {objective:?}: {:.3e}",
@@ -487,7 +537,7 @@ fn main() {
         let print_first = positive_integer_flag("--print-first", 0);
         let started = Instant::now();
         let mut processed = 0_u64;
-        let mapping_count = for_each_mapping(&raqes, &deployments, &facts, |mapping| {
+        let mapping_count = for_each_mapping(&raqes, &deployments, &facts, &accuracy, |mapping| {
             front.consider(mapping, score(&raqes, &deployments, mapping, &facts));
             processed += 1;
             if processed <= print_first {
@@ -514,7 +564,7 @@ fn main() {
         "WARNING: eager mode will retain every feasible mapping. For larger workloads, use --milp, \
          --candidates-only, or --sample-mappings N instead."
     );
-    let mappings = brute_force(&raqes, &deployments, &facts);
+    let mappings = brute_force(&raqes, &deployments, &facts, &accuracy);
     println!("{} feasible full mappings", mappings.len());
 
     let plan_costs: Vec<_> = mappings

rqe-optimizer/src/autosketch.rs

@@ -149,8 +149,7 @@ pub fn search(
             continue;
         }
         probes.push(index);
-        let feasible = accuracy(raqe, &costs[index])
-            .is_some_and(|value| value.is_finite() && raqe.accuracy_ok(value));
+        let feasible = raqe.meets_sla(accuracy(raqe, &costs[index]));
         evaluated.insert(index, feasible);
         if feasible && best.is_none_or(|b| cheaper(index, b)) {
             best = Some(index);
@@ -341,7 +340,7 @@ mod tests {
     use crate::analytical_cost_model::{score, PhaseCost};
     use crate::candidates::is_eligible;
     use crate::test_support::{facts, METRIC};
-    use crate::{AccuracyDirection, Capability, LabelSet};
+    use crate::{table_accuracy_at, AccuracyDirection, Capability, LabelSet};
     use std::collections::BTreeSet;
 
     const ERR: &str = "err";
@@ -392,7 +391,7 @@ mod tests {
 
     fn cheapest_feasible(raqe: &Raqe, costs: &[AtomicCostEntry]) -> usize {
         (0..costs.len())
-            .filter(|&i| raqe.accuracy_ok_for(&costs[i], 1))
+            .filter(|&i| raqe.meets_sla(table_accuracy_at(raqe, &costs[i], 1)))
             .min_by(|&a, &b| compare_resources(&costs[a], &costs[b]))
             .unwrap()
     }
@@ -460,7 +459,11 @@ mod tests {
         assert_eq!(window_adapter(&raqes[0]), (3_600_000, 60_000));
         assert_eq!(window_adapter(&raqes[1]), (3_600_000, 40_000));
         for (r, d) in raqes.iter().zip(&plan.deployments) {
-            assert!(is_eligible(r, d, &facts(1, 1)), "{} not eligible", r.id);
+            assert!(
+                is_eligible(r, d, &facts(1, 1), &crate::table_accuracy),
+                "{} not eligible",
+                r.id
+            );
         }
     }