diff --git a/aqpbm-core/src/atomic_costs.rs b/aqpbm-core/src/atomic_costs.rs index 81367af3..a80b4136 100644 --- a/aqpbm-core/src/atomic_costs.rs +++ b/aqpbm-core/src/atomic_costs.rs @@ -72,6 +72,36 @@ pub struct MeasuredAt { pub distribution: Option, } +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 { + 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; /// Why one [`MergedRecord`] didn't produce a row. Not an error: the caller diff --git a/aqpbm-core/src/lib.rs b/aqpbm-core/src/lib.rs index e75880cd..7485576e 100644 --- a/aqpbm-core/src/lib.rs +++ b/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, diff --git a/docs/rqe_sketch_deployment_v1.md b/docs/rqe_sketch_deployment_v1.md index 32ecedcb..87322771 100644 --- a/docs/rqe_sketch_deployment_v1.md +++ b/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 diff --git a/rqe-optimizer/examples/small_problem.rs b/rqe-optimizer/examples/small_problem.rs index c0fa6928..a7d27c7b 100644 --- a/rqe-optimizer/examples/small_problem.rs +++ b/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 { 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 diff --git a/rqe-optimizer/src/autosketch.rs b/rqe-optimizer/src/autosketch.rs index c63406d5..756ef304 100644 --- a/rqe-optimizer/src/autosketch.rs +++ b/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 + ); } } diff --git a/rqe-optimizer/src/candidates.rs b/rqe-optimizer/src/candidates.rs index fafb1faa..6aa9a2b9 100644 --- a/rqe-optimizer/src/candidates.rs +++ b/rqe-optimizer/src/candidates.rs @@ -2,8 +2,8 @@ use crate::analytical_cost_model; use crate::{ - AtomicCostEntry, Capability, Deployment, LabelSet, MetricFacts, Millis, Raqe, WorkloadFacts, - KEY_TRACKER_FAMILY, + Accuracy, AtomicCostEntry, Capability, Deployment, LabelSet, MetricFacts, Millis, Raqe, + WorkloadFacts, KEY_TRACKER_FAMILY, }; use std::collections::{BTreeMap, BTreeSet}; @@ -138,11 +138,13 @@ pub fn build_all_candidates( costs: &[AtomicCostEntry], facts: &WorkloadFacts, allow_undeployable_families: bool, + accuracy: &Accuracy, ) -> Vec { prune_dominated_candidates( raqes, facts, build_all_candidates_unpruned(raqes, costs, facts, allow_undeployable_families), + accuracy, ) } @@ -189,10 +191,11 @@ pub fn prune_dominated_candidates( raqes: &[Raqe], facts: &WorkloadFacts, candidates: Vec, + accuracy: &Accuracy, ) -> Vec { let costs: Vec = candidates .iter() - .map(|candidate| CandidateCosts::new(candidate, raqes, facts)) + .map(|candidate| CandidateCosts::new(candidate, raqes, facts, accuracy)) .collect(); candidates @@ -231,12 +234,17 @@ struct CandidateCosts { } impl CandidateCosts { - fn new(candidate: &Deployment, raqes: &[Raqe], facts: &WorkloadFacts) -> Self { + fn new( + candidate: &Deployment, + raqes: &[Raqe], + facts: &WorkloadFacts, + accuracy: &Accuracy, + ) -> Self { let ingest = analytical_cost_model::ingest(candidate, facts); let per_raqe = raqes .iter() .map(|raqe| { - is_eligible(raqe, candidate, facts).then(|| { + is_eligible(raqe, candidate, facts, accuracy).then(|| { [ analytical_cost_model::query_latency_ms(raqe, candidate, facts), analytical_cost_model::merge(raqe, candidate, facts).memory_bytes, @@ -330,18 +338,19 @@ pub fn eligible_deployments_for( r: &Raqe, deployments: &[Deployment], facts: &WorkloadFacts, + accuracy: &Accuracy, ) -> Vec { deployments .iter() .enumerate() - .filter(|(_, d)| is_eligible(r, d, facts)) + .filter(|(_, d)| is_eligible(r, d, facts, accuracy)) .map(|(i, _)| i) .collect() } /// Whether `d` can serve `r`. Every rule lives here, so the MILP, enumeration /// and candidate pruning agree on what is valid. -pub fn is_eligible(r: &Raqe, d: &Deployment, facts: &WorkloadFacts) -> bool { +pub fn is_eligible(r: &Raqe, d: &Deployment, facts: &WorkloadFacts, accuracy: &Accuracy) -> bool { let properties = d.properties(); r.capability == d.capability && r.metric == d.metric @@ -356,14 +365,14 @@ pub fn is_eligible(r: &Raqe, d: &Deployment, facts: &WorkloadFacts) -> bool { && properties.needs_delta_set_key_tracker == d.key_tracker.is_some() && (!properties.one_fixed_size_sketch_for_all_groups || measured_at_group_count(&d.config, facts[&d.metric].cardinality[&d.grouping_labels])) - && r.accuracy_ok_for(&d.config, r.lookback_ms / d.window_ms) + && r.meets_sla(accuracy(r, d)) } #[cfg(test)] mod tests { use super::*; use crate::test_support::{facts, METRIC}; - use crate::{AccuracyDirection, LabelSet}; + use crate::{table_accuracy, AccuracyDirection, LabelSet}; use std::collections::BTreeMap; fn raqe(id: &str, lookback: Millis, interval: Millis) -> Raqe { Raqe { @@ -411,21 +420,57 @@ mod tests { }; let r = raqe("a", 60_000, 60_000); // One window reads query_accuracy; 2 reads 1 and 4; 4 reads 4 alone. - assert!(is_eligible(&r, &at_window(60_000), &facts(1, 1))); - assert!(is_eligible(&r, &at_window(30_000), &facts(1, 1))); - assert!(is_eligible(&r, &at_window(15_000), &facts(1, 1))); + assert!(is_eligible( + &r, + &at_window(60_000), + &facts(1, 1), + &table_accuracy + )); + assert!(is_eligible( + &r, + &at_window(30_000), + &facts(1, 1), + &table_accuracy + )); + assert!(is_eligible( + &r, + &at_window(15_000), + &facts(1, 1), + &table_accuracy + )); // 12 and 20 merges bracket the bad count 16 from either side; 60 // reads 16 and 64, so it fails too even though 64 alone passes. - assert!(!is_eligible(&r, &at_window(5_000), &facts(1, 1))); - assert!(!is_eligible(&r, &at_window(3_000), &facts(1, 1))); - assert!(!is_eligible(&r, &at_window(1_000), &facts(1, 1))); + assert!(!is_eligible( + &r, + &at_window(5_000), + &facts(1, 1), + &table_accuracy + )); + assert!(!is_eligible( + &r, + &at_window(3_000), + &facts(1, 1), + &table_accuracy + )); + assert!(!is_eligible( + &r, + &at_window(1_000), + &facts(1, 1), + &table_accuracy + )); // Past the largest count reads only 64. - assert!(is_eligible(&r, &at_window(500), &facts(1, 1))); + assert!(is_eligible( + &r, + &at_window(500), + &facts(1, 1), + &table_accuracy + )); } #[test] fn shared_slide_comes_from_subset_gcd_not_all_divisors() { let raqes = vec![raqe("a", 60_000, 20_000), raqe("b", 60_000, 30_000)]; - let candidates = build_all_candidates(&raqes, &[cost()], &facts(1, 1), false); + let candidates = + build_all_candidates(&raqes, &[cost()], &facts(1, 1), false, &table_accuracy); let slides: BTreeSet<_> = candidates .iter() .filter(|d| d.window_ms == 60_000) @@ -457,11 +502,19 @@ mod tests { &[cost()], &facts(1, 1), false, + &table_accuracy, ); - assert!(candidates - .iter() - .all(|d| is_eligible(&unfiltered, d, &facts(1, 1)) - != is_eligible(&filtered, d, &facts(1, 1)))); + assert!(candidates.iter().all(|d| is_eligible( + &unfiltered, + d, + &facts(1, 1), + &table_accuracy + ) != is_eligible( + &filtered, + d, + &facts(1, 1), + &table_accuracy + ))); } #[test] @@ -479,6 +532,7 @@ mod tests { &[named("exact-min"), named("exact-max")], &facts(1, 1), false, + &table_accuracy, ); assert!(!candidates.is_empty()); assert!(candidates.iter().all(|d| d.config.sketch == "exact-min")); @@ -495,8 +549,13 @@ mod tests { ..cost() }; let costs = [named("kll-percall"), named("dd")]; - let candidates = - build_all_candidates(std::slice::from_ref(&r), &costs, &facts(1, 1), false); + let candidates = build_all_candidates( + std::slice::from_ref(&r), + &costs, + &facts(1, 1), + false, + &table_accuracy, + ); assert!(!candidates.is_empty()); assert!(candidates.iter().all(|d| d.config.sketch == "kll-percall")); let all = build_all_candidates_unpruned(&[r], &costs, &facts(1, 1), true); @@ -516,14 +575,15 @@ mod tests { slide_ms: 60_000, key_tracker: None, }; - assert!(is_eligible(&r, &d, &facts(1, 1))); + assert!(is_eligible(&r, &d, &facts(1, 1), &table_accuracy)); assert!(!is_eligible( &r, &Deployment { window_ms: 128_000, ..d.clone() }, - &facts(1, 1) + &facts(1, 1), + &table_accuracy )); assert!(!is_eligible( &r, @@ -531,7 +591,8 @@ mod tests { slide_ms: 70_000, ..d }, - &facts(1, 1) + &facts(1, 1), + &table_accuracy )); } @@ -553,7 +614,12 @@ mod tests { ..coarse.clone() }; - let retained = prune_dominated_candidates(&[r], &facts(1, 1), vec![fine, coarse.clone()]); + let retained = prune_dominated_candidates( + &[r], + &facts(1, 1), + vec![fine, coarse.clone()], + &table_accuracy, + ); assert_eq!(retained, vec![coarse]); } @@ -582,6 +648,7 @@ mod tests { &[r], &facts(1, 1), vec![small_window.clone(), large_window.clone()], + &table_accuracy, ); assert_eq!(retained, vec![large_window]); @@ -621,6 +688,7 @@ mod tests { &[r], &facts(1, 1), vec![whole_window.clone(), panes.clone()], + &table_accuracy, ); assert_eq!(retained, vec![whole_window, panes]); @@ -657,8 +725,12 @@ mod tests { ..direct.clone() }; - let retained = - prune_dominated_candidates(&[r], &facts(1, 1), vec![direct.clone(), halves.clone()]); + let retained = prune_dominated_candidates( + &[r], + &facts(1, 1), + vec![direct.clone(), halves.clone()], + &table_accuracy, + ); assert_eq!(retained, vec![direct, halves]); } @@ -698,6 +770,7 @@ mod tests { &[hydra, tracker.clone(), tracker], &facts(10, 10), true, + &table_accuracy, ); } } diff --git a/rqe-optimizer/src/enumerate.rs b/rqe-optimizer/src/enumerate.rs index 62ff90af..20c2ed53 100644 --- a/rqe-optimizer/src/enumerate.rs +++ b/rqe-optimizer/src/enumerate.rs @@ -10,7 +10,9 @@ //! An ILP would replace this module alone: `Mapping` already encodes //! `x_{i,D}` and `y_D`. -use crate::{candidates::eligible_deployments_for, Deployment, Mapping, Raqe, WorkloadFacts}; +use crate::{ + candidates::eligible_deployments_for, Accuracy, Deployment, Mapping, Raqe, WorkloadFacts, +}; /// RAQEs nothing in `deployments` can serve (§3.4). Surfaced separately from /// `brute_force`: an unservable RAQE is a modeling gap (tolerance too tight, @@ -19,10 +21,11 @@ pub fn unservable( raqes: &[Raqe], deployments: &[Deployment], facts: &WorkloadFacts, + accuracy: &Accuracy, ) -> Vec { raqes .iter() - .filter(|r| eligible_deployments_for(r, deployments, facts).is_empty()) + .filter(|r| eligible_deployments_for(r, deployments, facts, accuracy).is_empty()) .map(|r| r.id.clone()) .collect() } @@ -33,9 +36,10 @@ pub fn brute_force( raqes: &[Raqe], deployments: &[Deployment], facts: &WorkloadFacts, + accuracy: &Accuracy, ) -> Vec { let mut mappings = Vec::new(); - for_each_mapping(raqes, deployments, facts, |mapping| { + for_each_mapping(raqes, deployments, facts, accuracy, |mapping| { mappings.push(mapping.to_vec()) }); mappings @@ -48,9 +52,10 @@ pub fn for_each_mapping( raqes: &[Raqe], deployments: &[Deployment], facts: &WorkloadFacts, + accuracy: &Accuracy, mut visit: impl FnMut(&Mapping), ) -> u64 { - for_each_mapping_while(raqes, deployments, facts, |mapping| { + for_each_mapping_while(raqes, deployments, facts, accuracy, |mapping| { visit(mapping); true }) @@ -72,11 +77,12 @@ pub fn for_each_mapping_while( raqes: &[Raqe], deployments: &[Deployment], facts: &WorkloadFacts, + accuracy: &Accuracy, mut visit: impl FnMut(&Mapping) -> bool, ) -> EnumerationResult { let choices: Vec> = raqes .iter() - .map(|r| eligible_deployments_for(r, deployments, facts)) + .map(|r| eligible_deployments_for(r, deployments, facts, accuracy)) .collect(); if choices.iter().any(|c| c.is_empty()) { @@ -122,7 +128,7 @@ mod tests { use super::*; use crate::candidates; use crate::test_support::{facts, METRIC}; - use crate::{AtomicCostEntry, Capability, Deployment, LabelSet}; + use crate::{table_accuracy, AtomicCostEntry, Capability, Deployment, LabelSet}; use std::collections::BTreeMap; fn cost(sketch: &str) -> AtomicCostEntry { @@ -142,7 +148,7 @@ mod tests { } fn build_all_candidates(raqes: &[Raqe], costs: &[AtomicCostEntry]) -> Vec { - candidates::build_all_candidates(raqes, costs, &facts(1, 1), false) + candidates::build_all_candidates(raqes, costs, &facts(1, 1), false, &table_accuracy) } fn raqe(id: &str, interval: u64, lookback: u64) -> Raqe { @@ -167,8 +173,8 @@ mod tests { let table = vec![cost("cms-heap-topk-fastpath-vector2d")]; let deployments = build_all_candidates(&raqes, &table); - assert!(unservable(&raqes, &deployments, &facts(1, 1)).is_empty()); - let mappings = brute_force(&raqes, &deployments, &facts(1, 1)); + assert!(unservable(&raqes, &deployments, &facts(1, 1), &table_accuracy).is_empty()); + let mappings = brute_force(&raqes, &deployments, &facts(1, 1), &table_accuracy); assert!( mappings.iter().any(|m| m[0] == m[1]), "expected at least one mapping where both identical RAQEs land on the same deployment" @@ -182,21 +188,25 @@ mod tests { let deployments = build_all_candidates(&raqes, &table); assert_eq!( - unservable(&raqes, &deployments, &facts(1, 1)), + unservable(&raqes, &deployments, &facts(1, 1), &table_accuracy), vec!["a".to_string()] ); - assert!(brute_force(&raqes, &deployments, &facts(1, 1)).is_empty()); + assert!(brute_force(&raqes, &deployments, &facts(1, 1), &table_accuracy).is_empty()); } #[test] fn streaming_visits_the_same_mappings_as_eager_enumeration() { let raqes = vec![raqe("a", 60_000, 3_600_000), raqe("b", 60_000, 3_600_000)]; let deployments = build_all_candidates(&raqes, &[cost("cms-heap-topk-fastpath-vector2d")]); - let eager = brute_force(&raqes, &deployments, &facts(1, 1)); + let eager = brute_force(&raqes, &deployments, &facts(1, 1), &table_accuracy); let mut streamed = Vec::new(); - let count = for_each_mapping(&raqes, &deployments, &facts(1, 1), |mapping| { - streamed.push(mapping.clone()) - }); + let count = for_each_mapping( + &raqes, + &deployments, + &facts(1, 1), + &table_accuracy, + |mapping| streamed.push(mapping.clone()), + ); assert_eq!(count as usize, eager.len()); assert_eq!(streamed, eager); } @@ -206,10 +216,16 @@ mod tests { let raqes = vec![raqe("a", 60_000, 3_600_000), raqe("b", 60_000, 3_600_000)]; let deployments = build_all_candidates(&raqes, &[cost("cms-heap-topk-fastpath-vector2d")]); let mut sampled = Vec::new(); - let result = for_each_mapping_while(&raqes, &deployments, &facts(1, 1), |mapping| { - sampled.push(mapping.clone()); - sampled.len() < 2 - }); + let result = for_each_mapping_while( + &raqes, + &deployments, + &facts(1, 1), + &table_accuracy, + |mapping| { + sampled.push(mapping.clone()); + sampled.len() < 2 + }, + ); assert_eq!(result.visited, 2); assert!(!result.completed); diff --git a/rqe-optimizer/src/lib.rs b/rqe-optimizer/src/lib.rs index c41f0d8f..ecfa12cb 100644 --- a/rqe-optimizer/src/lib.rs +++ b/rqe-optimizer/src/lib.rs @@ -15,6 +15,7 @@ pub mod candidates; pub mod enumerate; pub mod milp; pub mod pareto; +pub mod saturation; use std::collections::BTreeMap; use std::collections::BTreeSet; @@ -137,6 +138,9 @@ pub struct FamilyProperties { /// The sketch can't list its groups, so a DeltaSet records which keys /// each window saw, for the query to probe. pub needs_delta_set_key_tracker: bool, + /// Answers without error, so its accuracy comes from the cost table + /// rather than a saturation curve. + pub exact: bool, } /// Panics on a variant with no entry: guessing would misprice it. @@ -145,14 +149,16 @@ pub fn family_properties(variant: &str) -> FamilyProperties { mergeable_across_windows: true, one_fixed_size_sketch_for_all_groups: false, needs_delta_set_key_tracker: false, + exact: false, }; match variant { - "exact-sum" - | "exact-min" - | "exact-max" - | "exact-increase" - | KEY_TRACKER_FAMILY - | "kll-percall" + "exact-sum" | "exact-min" | "exact-max" | "exact-increase" | KEY_TRACKER_FAMILY => { + FamilyProperties { + exact: true, + ..one_sketch_per_group + } + } + "kll-percall" | "dd" | "hll" | "univmon-cardinality" @@ -163,6 +169,7 @@ pub fn family_properties(variant: &str) -> FamilyProperties { mergeable_across_windows: true, one_fixed_size_sketch_for_all_groups: true, needs_delta_set_key_tracker: true, + exact: false, }, _ => panic!("{variant} has no FamilyProperties; add it to family_properties"), } @@ -182,6 +189,9 @@ pub struct MetricFacts { /// `(lo, hi)`: smallest and largest positive sample value, if known. /// Sizes DDSketch; without it, DDSketch keeps the measured memory. pub value_range: Option<(f64, f64)>, + /// Fitted data parameters per grouping `G`, for reading sketch accuracy + /// off the saturation curves. Without one, no sketch serves `G`. + pub data_shape: BTreeMap, } impl MetricFacts { @@ -299,8 +309,10 @@ pub struct Raqe { pub spatial_filter: String, /// `G`: the query's group-by labels. pub grouping_labels: LabelSet, - /// Which `AtomicCostEntry::query_accuracy` key to check. Comparators - /// name their metrics differently per capability, so the RAQE picks. + /// Which accuracy metric to check: a saturation curve's `error_metric`, + /// or an exact accumulator's `AtomicCostEntry::query_accuracy` key. + /// Comparators name their metrics differently per capability, so the + /// RAQE picks. pub accuracy_metric: String, /// `tol_i`: hard constraint. Read as a ceiling or a floor depending on /// [`Raqe::accuracy_direction`]. @@ -321,19 +333,40 @@ impl Raqe { } } - /// Whether `config` measured this RAQE's metric and clears it after - /// `merges` windows are folded per query (`L / x`), at both measured - /// counts bracketing `merges` (error is not monotone in the count). - /// Missing is not passing. - pub fn accuracy_ok_for(&self, config: &AtomicCostEntry, merges: u64) -> bool { - accuracy_at(config, merges).iter().all(|scores| { - scores - .get(&self.accuracy_metric) - .is_some_and(|&v| self.accuracy_ok(v)) - }) + /// Whether `accuracy` clears the SLA. Unknown or non-finite never passes. + /// The MILP's eligibility and the AutoSketch baseline both decide here. + pub fn meets_sla(&self, accuracy: Option) -> bool { + accuracy.is_some_and(|value| value.is_finite() && self.accuracy_ok(value)) } } +/// The accuracy `raqe` gets from a deployment, or `None` when unknown -- +/// unknown never passes. Production uses +/// [`saturation::SaturationCurves::accuracy`]. +pub type Accuracy<'a> = dyn Fn(&Raqe, &Deployment) -> Option + 'a; + +/// The cost table's measured value after the query's `L / x` merges, which is +/// the exact accumulators' accuracy. +pub fn table_accuracy(raqe: &Raqe, deployment: &Deployment) -> Option { + table_accuracy_at( + raqe, + &deployment.config, + raqe.lookback_ms / deployment.window_ms, + ) +} + +/// The worse of `config`'s scores at the two measured counts bracketing +/// `merges` (error is not monotone in the count). Missing at either is `None`. +pub fn table_accuracy_at(raqe: &Raqe, config: &AtomicCostEntry, merges: u64) -> Option { + let [below, above] = + accuracy_at(config, merges).map(|scores| scores.get(&raqe.accuracy_metric).copied()); + let (below, above) = (below?, above?); + Some(match raqe.accuracy_direction { + AccuracyDirection::LowerIsBetter => below.max(above), + AccuracyDirection::HigherIsBetter => below.min(above), + }) +} + /// `config`'s scores at the two measured merge counts bracketing `merges`: /// the largest at or below it (count 1 is `query_accuracy`), and the smallest /// at or above it, else the largest measured. One window, or a row with no @@ -433,6 +466,7 @@ pub(crate) mod test_support { scrape_interval_ms: 1_000, cardinality: BTreeMap::from([(LabelSet::new(), groups), (labels, series)]), value_range: None, + data_shape: BTreeMap::new(), }, )]) } @@ -531,13 +565,14 @@ mod tests { accuracy_direction: AccuracyDirection::HigherIsBetter, ..raqe(60_000, 60_000) }; - assert!(r.accuracy_ok_for(&config, 1)); - assert!(r.accuracy_ok_for(&config, 4)); - assert!(!r.accuracy_ok_for(&config, 8)); // [4, 16]: 16 fails - assert!(!r.accuracy_ok_for(&config, 16)); - assert!(!r.accuracy_ok_for(&config, 32)); // [16, 64] - assert!(r.accuracy_ok_for(&config, 64)); - assert!(r.accuracy_ok_for(&config, 5_000)); // beyond: [64, 64] + let ok = |merges| table_accuracy_at(&r, &config, merges).is_some_and(|v| r.accuracy_ok(v)); + assert!(ok(1)); + assert!(ok(4)); + assert!(!ok(8)); // [4, 16]: 16 fails + assert!(!ok(16)); + assert!(!ok(32)); // [16, 64] + assert!(ok(64)); + assert!(ok(5_000)); // beyond: [64, 64] } #[test] @@ -593,6 +628,7 @@ mod tests { (labels(&["service", "endpoint"]), 50), ]), value_range: None, + data_shape: BTreeMap::new(), }, )]); let r = Raqe { @@ -611,6 +647,7 @@ mod tests { scrape_interval_ms: 0, cardinality: BTreeMap::from([(labels(&["service"]), 0)]), value_range: Some((0.0, 1.0)), + data_shape: BTreeMap::new(), }, )]); let raqes = [ @@ -674,6 +711,7 @@ mod tests { (labels(&["service", "endpoint"]), 50), ]), value_range: None, + data_shape: BTreeMap::new(), }, )]); let r = Raqe { diff --git a/rqe-optimizer/src/milp.rs b/rqe-optimizer/src/milp.rs index de8f82a1..cc986e8c 100644 --- a/rqe-optimizer/src/milp.rs +++ b/rqe-optimizer/src/milp.rs @@ -5,7 +5,7 @@ use crate::analytical_cost_model::{self, score, PhaseCost, PlanCost, BYTES_PER_GIB}; use crate::candidates::eligible_deployments_for; -use crate::{Deployment, Mapping, Raqe, WorkloadFacts}; +use crate::{Accuracy, Deployment, Mapping, Raqe, WorkloadFacts}; use good_lp::{ default_solver, variable, Expression, ProblemVariables, ResolutionError, Solution, SolverModel, Variable, @@ -80,10 +80,11 @@ pub fn minimize( deployments: &[Deployment], facts: &WorkloadFacts, objective: Objective, + accuracy: &Accuracy, ) -> Result { let eligible: Vec> = raqes .iter() - .map(|raqe| eligible_deployments_for(raqe, deployments, facts)) + .map(|raqe| eligible_deployments_for(raqe, deployments, facts, accuracy)) .collect(); assert!( eligible.iter().all(|choices| !choices.is_empty()), @@ -274,7 +275,7 @@ mod tests { use crate::candidates::build_all_candidates; use crate::enumerate::brute_force; use crate::test_support::{self, facts}; - use crate::{AtomicCostEntry, Capability}; + use crate::{table_accuracy, AtomicCostEntry, Capability}; /// Window and slide 60 s, merge 1 CPU-sec. fn deployment(insert_cpu_secs: f64, memory: f64, query_cpu_secs: f64) -> Deployment { @@ -313,12 +314,13 @@ mod tests { objective: Objective, ) { let facts = facts(1, 1); - let best = brute_force(raqes, deployments, &facts) + let best = brute_force(raqes, deployments, &facts, &table_accuracy) .iter() .map(|mapping| objective.value(&score(raqes, deployments, mapping, &facts))) .min_by(f64::total_cmp) .expect("test workload is servable"); - let milp = minimize(raqes, deployments, &facts, objective).expect("feasible MILP"); + let milp = minimize(raqes, deployments, &facts, objective, &table_accuracy) + .expect("feasible MILP"); let got = objective.value(&milp.plan_cost); assert!( (got - best).abs() <= 1e-9 * best, @@ -337,9 +339,15 @@ mod tests { sketch: sketch.into(), ..deployment(1.0, 1.0, 1.0).config }]; - let candidates = build_all_candidates(&raqes, &costs, &facts(1, 1), false); - let milp = minimize(&raqes, &candidates, &facts(1, 1), Objective::default()) - .expect("feasible MILP"); + let candidates = build_all_candidates(&raqes, &costs, &facts(1, 1), false, &table_accuracy); + let milp = minimize( + &raqes, + &candidates, + &facts(1, 1), + Objective::default(), + &table_accuracy, + ) + .expect("feasible MILP"); raqes.len() - milp.deployments.len() } @@ -366,8 +374,14 @@ mod tests { // CPU 1 + 3/60 with latency 3, or 2 + 1/60 with latency 1. let deployments = vec![deployment(1.0, 20.0, 3.0), deployment(2.0, 10.0, 1.0)]; let solve = |raqes: &[Raqe]| { - let milp = minimize(raqes, &deployments, &facts(1, 1), Objective::default()) - .expect("feasible MILP"); + let milp = minimize( + raqes, + &deployments, + &facts(1, 1), + Objective::default(), + &table_accuracy, + ) + .expect("feasible MILP"); chosen(&milp, &deployments) }; @@ -401,8 +415,14 @@ mod tests { deployment(0.05, 2.0 * BYTES_PER_GIB, 0.0), ]; let solve = |objective| { - let milp = - minimize(&raqes, &deployments, &facts(1, 1), objective).expect("feasible MILP"); + let milp = minimize( + &raqes, + &deployments, + &facts(1, 1), + objective, + &table_accuracy, + ) + .expect("feasible MILP"); chosen(&milp, &deployments) }; @@ -441,8 +461,14 @@ mod tests { deployment(tiny, tiny, 1.5e-7), deployment(10.0 * tiny, tiny, 0.5e-7), ]; - let milp = minimize(&raqes, &deployments, &facts(1, 1), Objective::default()) - .expect("feasible MILP"); + let milp = minimize( + &raqes, + &deployments, + &facts(1, 1), + Objective::default(), + &table_accuracy, + ) + .expect("feasible MILP"); assert_eq!(chosen(&milp, &deployments), vec![1]); assert!(milp.plan_cost.query_latency_ms[0] <= 1e-4); } @@ -458,6 +484,7 @@ mod tests { &[unused, shared.clone()], &facts(1, 1), Objective::default(), + &table_accuracy, ) .expect("feasible MILP"); diff --git a/rqe-optimizer/src/saturation.rs b/rqe-optimizer/src/saturation.rs new file mode 100644 index 00000000..e254ad1e --- /dev/null +++ b/rqe-optimizer/src/saturation.rs @@ -0,0 +1,689 @@ +//! Sketch accuracy read from the saturation study's error-vs-N curves +//! (`scripts/study_saturation.py`, #130, #140) at the number of items a +//! deployment's answer covers. Design decisions: #156. +//! +//! The curve is read at the number of items one group receives over the +//! RAQE's whole lookback: series per group times scrapes per lookback. That +//! holds whether the answer is one sketch or a merge of many smaller-window +//! sketches; the window size doesn't matter. +//! KLL and top-k merge lossily (#131); their merge penalty is #158. +//! +//! The cost table's sketch accuracies are not read: each is one point on a +//! curve, at the row's `measured_at`, and [`SaturationCurves::check_cost_table`] +//! checks that the two agree there (#171). + +use std::collections::BTreeMap; +use std::io; +use std::path::Path; + +use crate::autosketch::window_adapter; +use crate::{ + family_properties, table_accuracy, AccuracyDirection, AtomicCostEntry, Deployment, LabelSet, + MetricFacts, Raqe, WorkloadFacts, +}; +use aqpbm_core::MeasuredShape; + +/// The data parameters a saturation curve is keyed by, fitted from the +/// dataset per (metric, grouping) as the worst case over windows and groups. +#[derive(Debug, Clone, Copy, PartialEq)] +pub struct DataShape { + /// Zipf skew θ of the keys inserted into one group's sketch (CMS, top-k, + /// HLL). Describes whichever insertion mode the engine uses: once per + /// sample, or weighted by sample value. + pub zipf_s: f64, + /// `K`: distinct keys per group per window. + pub distinct_keys: f64, + /// Pareto tail index `a` of the values (KLL, DDSketch). + pub tail_index: f64, +} + +/// The two runs' output directories under `--saturation-dir`. A point in +/// both is taken whole from the later one. +const RUN_DIRS: [&str; 2] = ["out_grid_1e7_cost", "out_1e9"]; + +/// One (sketch, config, data shape) point of the study. +#[derive(Debug, Clone, PartialEq)] +struct GridPoint { + params: BTreeMap, + shape: MeasuredShape, + error_metric: String, + /// `None`: still changing at the largest measured N. + n_sat: Option, + /// `(n, seed_mean_error, seed_se)`, ascending in `n`. + curve: Vec<(f64, f64, f64)>, +} + +/// How far a cost-table value may sit outside the curve's range at its N: +/// this many of the curve's seed standard errors, plus [`AGREEMENT_RELATIVE`] +/// of the curve's value, because the table row is one seed of its own. +const AGREEMENT_SES: f64 = 3.0; +const AGREEMENT_RELATIVE: f64 = 0.05; + +impl GridPoint { + /// Zipf θ, or the Pareto tail index for quantile sketches. + fn shape_param(&self) -> f64 { + match self.shape { + MeasuredShape::Zipf { skew, .. } => skew, + MeasuredShape::Pareto { tail_index } => tail_index, + } + } + + /// `K`; `None` for quantile sketches. + fn distinct_keys(&self) -> Option { + match self.shape { + MeasuredShape::Zipf { keys, .. } => Some(keys), + MeasuredShape::Pareto { .. } => None, + } + } + + /// Whether a value measured at `n` agrees with the curve: within + /// tolerance of the checkpoint at `n`, or of the range between the two + /// either side. `None` where the curve can't say (as in + /// [`GridPoint::error_at`]). + fn agrees_at(&self, n: f64, value: f64) -> Option { + let &(first_n, ..) = self.curve.first()?; + let &last = self.curve.last()?; + let neighbours = if n < first_n { + return None; + } else if n >= last.0 { + self.n_sat?; + [last, last] + } else { + let above = self + .curve + .partition_point(|&(checkpoint, ..)| checkpoint < n); + if self.curve[above].0 == n { + [self.curve[above]; 2] + } else { + [self.curve[above - 1], self.curve[above]] + } + }; + let [(_, a, a_se), (_, b, b_se)] = neighbours; + let (low, high) = (a.min(b), a.max(b)); + let tolerance = AGREEMENT_SES * a_se.max(b_se) + AGREEMENT_RELATIVE * high.abs(); + Some(value >= low - tolerance && value <= high + tolerance) + } + + /// Q6 of #156: between checkpoints, the worse neighbour; below the first, + /// unmeasured; past the last, the plateau only if the point saturated. + fn error_at(&self, n: f64, direction: AccuracyDirection) -> Option { + let &(first_n, ..) = self.curve.first()?; + let &(last_n, last_error, _) = self.curve.last()?; + if n < first_n { + return None; + } + if n >= last_n { + return self.n_sat.map(|_| last_error); + } + let above = self + .curve + .partition_point(|&(checkpoint, ..)| checkpoint < n); + let (checkpoint, error, _) = self.curve[above]; + if checkpoint == n { + Some(error) + } else { + Some(worse(error, self.curve[above - 1].1, direction)) + } + } +} + +fn worse(a: f64, b: f64, direction: AccuracyDirection) -> f64 { + match direction { + AccuracyDirection::LowerIsBetter => a.max(b), + AccuracyDirection::HigherIsBetter => a.min(b), + } +} + +/// What [`SaturationCurves::check_cost_table`] found, one line per row. +#[derive(Debug, Default, PartialEq)] +pub struct CostTableCheck { + /// Rows whose accuracy disagrees with the curve at their `measured_at`, + /// whose config the grid lacks, or that lack the curve's metric. + pub mismatched: Vec, + /// Rows with no curve at their measured shape or N, such as KLL: the + /// table measures Zipf ranks, the study Pareto values. + pub unchecked: Vec, +} + +#[derive(Debug, Clone, Default)] +pub struct SaturationCurves { + points_by_sketch: BTreeMap>, +} + +impl SaturationCurves { + /// Reads `saturation.csv` (for the saturation point) and `saturation_curve.csv` from + /// each of [`RUN_DIRS`] under `dir`. + pub fn load(dir: &Path) -> io::Result { + type PointKey = (String, String, String, String, String); + let key = |row: &CsvRow| -> PointKey { + let field = |name: &str| row[name].clone(); + ( + field("sketch"), + field("config"), + field("dist"), + field("param"), + field("cardinality"), + ) + }; + let mut points: BTreeMap = BTreeMap::new(); + for run in RUN_DIRS { + let run_dir = dir.join(run); + let mut run_points = BTreeMap::new(); + for row in read_csv(&run_dir.join("saturation.csv"))? { + let point = GridPoint { + params: parse_config(&row["config"]), + shape: parse_shape(&row)?, + error_metric: row["error_metric"].clone(), + n_sat: match row["n_sat"].as_str() { + "not_saturated" => None, + n => Some(parse_number(n)?), + }, + curve: Vec::new(), + }; + run_points.insert(key(&row), (row["sketch"].clone(), point)); + } + for row in read_csv(&run_dir.join("saturation_curve.csv"))? { + let Some((_, point)) = run_points.get_mut(&key(&row)) else { + return Err(invalid(format!( + "{run}: curve row for a point not in saturation.csv: {row:?}" + ))); + }; + point.curve.push(( + parse_number(&row["n"])?, + parse_number(&row["seed_mean_error"])?, + parse_number(&row["seed_se"])?, + )); + } + points.extend(run_points); + } + let mut points_by_sketch: BTreeMap> = BTreeMap::new(); + for (sketch, mut point) in points.into_values() { + point.curve.sort_by(|a, b| a.0.total_cmp(&b.0)); + points_by_sketch.entry(sketch).or_default().push(point); + } + Ok(Self { points_by_sketch }) + } + + /// The accuracy `raqe` gets from `deployment`, or `None` when it can't be + /// read: no fitted [`DataShape`], a config or shape outside the grid, the + /// wrong metric, or too few items. + /// Exact accumulators keep the cost table's value. `facts` must pass + /// [`crate::validate_facts`]. + pub fn accuracy( + &self, + raqe: &Raqe, + deployment: &Deployment, + facts: &WorkloadFacts, + ) -> Option { + if family_properties(&deployment.config.sketch).exact { + return table_accuracy(raqe, deployment); + } + let metric_facts = &facts[&deployment.metric]; + let grouping = &deployment.grouping_labels; + let shape = metric_facts.data_shape.get(grouping)?; + let points = self.bracketing_points(&deployment.config, shape)?; + let covered = items_per_group(metric_facts, grouping, raqe.lookback_ms); + let mut worst = None; + for point in points { + if point.error_metric != raqe.accuracy_metric { + return None; + } + let error = point.error_at(covered, raqe.accuracy_direction)?; + worst = Some(worst.map_or(error, |w| worse(w, error, raqe.accuracy_direction))); + } + worst + } + + /// The oracle for [`crate::autosketch::plan`]: AutoSketch keeps one + /// unmerged sketch per query window. + pub fn autosketch_accuracy( + &self, + raqe: &Raqe, + config: &AtomicCostEntry, + facts: &WorkloadFacts, + ) -> Option { + let (window_ms, slide_ms) = window_adapter(raqe); + let deployment = Deployment { + capability: raqe.capability, + metric: raqe.metric.clone(), + spatial_filter: raqe.spatial_filter.clone(), + grouping_labels: raqe.grouping_labels.clone(), + config: config.clone(), + window_ms, + slide_ms, + key_tracker: None, + }; + self.accuracy(raqe, &deployment, facts) + } + + /// Checks the cost table against the curves: a sketch row's + /// `query_accuracy` is the curve of its config and + /// `measured_at.data_shape()`, read at `measured_at.items_per_instance`. + /// Rows of sketches the study didn't run (exact accumulators among them) + /// are skipped. + pub fn check_cost_table(&self, costs: &[AtomicCostEntry]) -> CostTableCheck { + let mut check = CostTableCheck::default(); + for row in costs { + let Some(points) = self.points_by_sketch.get(&row.sketch) else { + continue; + }; + let name = format!("{} {}", row.sketch, row.sketch_config["params"]); + let params = config_params(row); + let mut same_config = points + .iter() + .filter(|point| Some(&point.params) == params.as_ref()) + .peekable(); + if same_config.peek().is_none() { + check + .mismatched + .push(format!("{name}: config not in the saturation grid")); + continue; + } + let Some(measured_at) = &row.measured_at else { + check.unchecked.push(format!("{name}: no measured_at")); + continue; + }; + let shape = measured_at.data_shape(); + let Some(point) = same_config.find(|point| Some(point.shape) == shape) else { + check + .unchecked + .push(format!("{name}: no grid point at {shape:?}")); + continue; + }; + let metric = &point.error_metric; + let Some(&value) = row.query_accuracy.get(metric) else { + check + .mismatched + .push(format!("{name}: the table has no {metric}")); + continue; + }; + let n = measured_at.items_per_instance as f64; + match point.agrees_at(n, value) { + None => check + .unchecked + .push(format!("{name}: N = {n} is outside the curve")), + Some(false) => check.mismatched.push(format!( + "{name}: {metric} = {value} at N = {n}, curve {:?}", + point.curve + )), + Some(true) => {} + } + } + check + } + + /// Every grid point of `config` bracketing `shape` (Q8 of #156), or + /// `None` when `config` isn't in the grid, `shape` lies outside it, or a + /// bracketing point is missing. + fn bracketing_points( + &self, + config: &AtomicCostEntry, + shape: &DataShape, + ) -> Option> { + let params = config_params(config)?; + let points: Vec<&GridPoint> = self + .points_by_sketch + .get(&config.sketch)? + .iter() + .filter(|point| point.params == params) + .collect(); + let quantile = points.first()?.distinct_keys().is_none(); + let shape_params = if quantile { + bracket(points.iter().map(|p| p.shape_param()), shape.tail_index)? + } else { + bracket(points.iter().map(|p| p.shape_param()), shape.zipf_s)? + }; + let keys = if quantile { + vec![None] + } else { + let keys = points.iter().filter_map(|p| p.distinct_keys()); + bracket(keys, shape.distinct_keys)? + .into_iter() + .map(Some) + .collect() + }; + let bracketing: Vec<&GridPoint> = points + .into_iter() + .filter(|p| { + shape_params.contains(&p.shape_param()) && keys.contains(&p.distinct_keys()) + }) + .collect(); + (bracketing.len() == shape_params.len() * keys.len()).then_some(bracketing) + } +} + +/// Items one group receives in `window_ms`: series per group times scrapes +/// in the window. +fn items_per_group(facts: &MetricFacts, grouping: &LabelSet, window_ms: u64) -> f64 { + let series = facts.cardinality[&facts.labels] as f64; + let groups = facts.cardinality[grouping] as f64; + series / groups * window_ms as f64 / facts.scrape_interval_ms as f64 +} + +/// The grid values equal to `target`, else the two either side of it; +/// `None` outside the grid. +fn bracket(values: impl Iterator, target: f64) -> Option> { + let mut values: Vec = values.collect(); + values.sort_by(f64::total_cmp); + values.dedup(); + if values.contains(&target) { + return Some(vec![target]); + } + let above = values.partition_point(|&v| v < target); + (above > 0 && above < values.len()).then(|| vec![values[above - 1], values[above]]) +} + +/// `{"params": {"rows": 3, "cols": 1024}}` → `{rows: 3, cols: 1024}`. +fn config_params(config: &AtomicCostEntry) -> Option> { + config.sketch_config["params"] + .as_object()? + .iter() + .map(|(name, value)| Some((name.clone(), value.as_f64()?))) + .collect() +} + +/// `"rows=3 cols=1024"` → `{rows: 3, cols: 1024}`; the grid's config strings. +fn parse_config(config: &str) -> BTreeMap { + config + .split_whitespace() + .filter_map(|pair| { + let (name, value) = pair.split_once('=')?; + Some((name.to_string(), value.parse().ok()?)) + }) + .collect() +} + +/// A grid row's `dist`, `param` and `cardinality` as a [`MeasuredShape`]. +fn parse_shape(row: &CsvRow) -> io::Result { + let param = parse_number(&row["param"])?; + match row["dist"].as_str() { + "zipf" => Ok(MeasuredShape::Zipf { + skew: param, + keys: parse_number(&row["cardinality"])?, + }), + "pareto" => Ok(MeasuredShape::Pareto { tail_index: param }), + dist => Err(invalid(format!("unknown dist {dist:?}"))), + } +} + +fn parse_number(text: &str) -> io::Result { + text.parse() + .map_err(|_| invalid(format!("not a number: {text:?}"))) +} + +fn invalid(message: String) -> io::Error { + io::Error::new(io::ErrorKind::InvalidData, message) +} + +type CsvRow = BTreeMap; + +// ponytail: plain comma split; the study's CSVs quote nothing (configs hold +// spaces, never commas). Use the csv crate if that changes. +fn read_csv(path: &Path) -> io::Result> { + let text = std::fs::read_to_string(path) + .map_err(|e| io::Error::new(e.kind(), format!("{}: {e}", path.display())))?; + let mut lines = text.lines(); + let header: Vec<&str> = lines.next().unwrap_or_default().split(',').collect(); + Ok(lines + .filter(|line| !line.is_empty()) + .map(|line| { + header + .iter() + .zip(line.split(',')) + .map(|(name, value)| (name.to_string(), value.to_string())) + .collect() + }) + .collect()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::test_support::{facts, raqe, METRIC}; + use crate::Capability; + + const TOPK: &str = "cms-heap-topk-fastpath-vector2d"; + + /// Top-k rows=3 cols=1024 at θ ∈ {1.0, 1.2}, K ∈ {1e3, 1e5}. Precision + /// falls with N; θ = 1.0 is worse, and K = 1e5 never saturates. + fn curves() -> SaturationCurves { + let point = |theta: f64, keys: f64, offset: f64, n_sat: Option| GridPoint { + params: parse_config("rows=3 cols=1024"), + shape: MeasuredShape::Zipf { skew: theta, keys }, + error_metric: "precision_at_k".into(), + n_sat, + curve: vec![ + (1e3, 1.0 - offset, 0.0), + (1e4, 0.95 - offset, 0.01), + (1e5, 0.9 - offset, 0.0), + ], + }; + SaturationCurves { + points_by_sketch: BTreeMap::from([( + TOPK.to_string(), + vec![ + point(1.0, 1e3, 0.05, Some(1e4)), + point(1.2, 1e3, 0.0, Some(1e4)), + point(1.0, 1e5, 0.1, None), + point(1.2, 1e5, 0.08, None), + ], + )]), + } + } + + /// One group of `series` series scraped every second, shaped `shape`. + fn workload(series: u64, shape: DataShape) -> WorkloadFacts { + let mut facts = facts(1, series); + let metric = facts.get_mut(METRIC).unwrap(); + metric.data_shape.insert(LabelSet::new(), shape); + facts + } + + fn shape(zipf_s: f64, distinct_keys: f64) -> DataShape { + DataShape { + zipf_s, + distinct_keys, + tail_index: 2.0, + } + } + + fn topk_raqe(lookback_ms: u64) -> Raqe { + Raqe { + capability: Capability::TopKByValue, + accuracy_metric: "precision_at_k".into(), + accuracy_direction: AccuracyDirection::HigherIsBetter, + ..raqe(lookback_ms, lookback_ms) + } + } + + fn deployment(sketch: &str, window_ms: u64) -> Deployment { + Deployment { + capability: Capability::TopKByValue, + metric: METRIC.into(), + spatial_filter: String::new(), + grouping_labels: LabelSet::new(), + config: AtomicCostEntry { + sketch: sketch.into(), + sketch_config: serde_json::json!({"params": {"rows": 3, "cols": 1024}}), + mem_bytes_per_instance: 1.0, + insert_cpu_secs: 1.0, + merge_cpu_secs: 1.0, + query_cpu_secs: 1.0, + query_accuracy: BTreeMap::from([("precision_at_k".into(), 1.0)]), + merge_accuracy: BTreeMap::new(), + measured_at: None, + }, + window_ms, + slide_ms: window_ms, + key_tracker: None, + } + } + + #[test] + fn reads_the_worse_neighbour_between_checkpoints_and_the_plateau_past_the_last() { + let point = &curves().points_by_sketch[TOPK][1]; + let higher = AccuracyDirection::HigherIsBetter; + assert_eq!(point.error_at(1e4, higher), Some(0.95)); + assert_eq!(point.error_at(3e4, higher), Some(0.9)); + assert_eq!(point.error_at(5e2, higher), None); + assert_eq!(point.error_at(1e9, higher), Some(0.9)); + let unsaturated = &curves().points_by_sketch[TOPK][2]; + assert_eq!(unsaturated.error_at(1e9, higher), None); + } + + #[test] + fn takes_the_worst_bracketing_grid_point() { + // 10 series × 1000 s = 1e4 items; θ = 1.1 brackets 1.0 and 1.2. + let facts = workload(10, shape(1.1, 1e3)); + let r = topk_raqe(1_000_000); + let value = curves().accuracy(&r, &deployment(TOPK, 1_000_000), &facts); + assert_eq!( + value, + Some(0.95 - 0.05), + "θ = 1.0 at 1e4 is worse than θ = 1.2" + ); + } + + #[test] + fn a_shape_outside_the_grid_or_without_a_fit_has_no_accuracy() { + let r = topk_raqe(1_000_000); + let d = deployment(TOPK, 1_000_000); + assert_eq!( + curves().accuracy(&r, &d, &workload(10, shape(0.5, 1e3))), + None + ); + assert_eq!( + curves().accuracy(&r, &d, &workload(10, shape(1.0, 1e7))), + None + ); + assert_eq!(curves().accuracy(&r, &d, &facts(1, 10)), None); + } + + #[test] + fn a_metric_the_curve_did_not_measure_has_no_accuracy() { + let facts = workload(10, shape(1.2, 1e3)); + let r = Raqe { + accuracy_metric: "recall_at_k".into(), + ..topk_raqe(1_000_000) + }; + assert_eq!( + curves().accuracy(&r, &deployment(TOPK, 1_000_000), &facts), + None + ); + } + + #[test] + fn merged_answers_read_the_curve_at_the_lookback_count_whatever_the_window() { + // 10 items/s: a 100 s window holds 1e3 items, below the 1e4 + // saturation point, but the 1e4 s lookback covers 1e5 items whatever + // the window. + let facts = workload(10, shape(1.2, 1e3)); + let r = topk_raqe(10_000_000); + for window_ms in [100_000, 1_000_000, 10_000_000] { + assert_eq!( + curves().accuracy(&r, &deployment(TOPK, window_ms), &facts), + Some(0.9) + ); + } + } + + #[test] + fn exact_accumulators_keep_the_cost_table_value() { + let r = topk_raqe(1_000_000); + let value = curves().accuracy(&r, &deployment("exact-sum", 1_000_000), &facts(1, 10)); + assert_eq!(value, Some(1.0)); + } + + /// A top-k row measured at θ = 1.2, K = 1e3 and `items`, scoring + /// `precision`. + fn measured_row(cols: u64, items: u64, precision: f64) -> AtomicCostEntry { + use aqpbm_core::{DataDistribution, MeasuredAt, ZipfParameter}; + AtomicCostEntry { + sketch_config: serde_json::json!({"params": {"rows": 3, "cols": cols}}), + query_accuracy: BTreeMap::from([("precision_at_k".into(), precision)]), + measured_at: Some(MeasuredAt { + items_per_instance: items, + keys_per_instance: Some(1_000), + value_range: None, + merge_operand_items: None, + distribution: Some(DataDistribution::Zipf(ZipfParameter { + skewness: 1.2, + population_size: 1_000, + seed: 1, + })), + }), + ..deployment(TOPK, 1).config + } + } + + #[test] + fn the_cost_table_is_a_point_on_the_curve() { + let check = |rows: &[AtomicCostEntry]| curves().check_cost_table(rows); + // θ = 1.2, K = 1e3 reads 0.95 ± 3 × 0.01 + 5% at N = 1e4. + assert_eq!( + check(&[measured_row(1024, 10_000, 0.93)]), + CostTableCheck::default() + ); + assert_eq!( + check(&[measured_row(1024, 10_000, 0.8)]).mismatched.len(), + 1 + ); + // Between checkpoints: the range of the two either side. + assert!(check(&[measured_row(1024, 30_000, 0.92)]) + .mismatched + .is_empty()); + // A config off the grid is a mismatch; a shape or N the study didn't + // run is unchecked. + assert_eq!( + check(&[measured_row(2048, 10_000, 0.95)]).mismatched.len(), + 1 + ); + assert_eq!(check(&[measured_row(1024, 100, 0.95)]).unchecked.len(), 1); + let mut pareto = measured_row(1024, 10_000, 0.95); + pareto.measured_at.as_mut().unwrap().distribution = None; + assert_eq!(check(&[pareto]).unchecked.len(), 1); + // A table without the curve's metric is a mismatch. + let mut other_metric = measured_row(1024, 10_000, 0.95); + other_metric.query_accuracy = BTreeMap::from([("recall_at_k".into(), 0.95)]); + assert_eq!(check(&[other_metric]).mismatched.len(), 1); + // Sketches the study didn't run, exact ones among them, are skipped. + let exact = AtomicCostEntry { + sketch: "exact-sum".into(), + ..measured_row(1024, 10_000, 0.0) + }; + assert_eq!(check(&[exact]), CostTableCheck::default()); + } + + #[test] + fn load_takes_a_point_in_both_runs_from_the_1e9_run() { + let dir = std::env::temp_dir().join(format!("rqe-saturation-{}", std::process::id())); + let header = "family,sketch,config,dist,param,cardinality"; + for (run, n_sat, last) in [ + (RUN_DIRS[0], "not_saturated", "1e7"), + (RUN_DIRS[1], "1e8", "1e9"), + ] { + std::fs::create_dir_all(dir.join(run)).unwrap(); + let point = "quantile,kll-percall,k=200,pareto,2.0,"; + std::fs::write( + dir.join(run).join("saturation.csv"), + format!( + "{header},n_sat,final_error,error_metric\n{point},{n_sat},0.01,mean_rank_err\n" + ), + ) + .unwrap(); + std::fs::write( + dir.join(run).join("saturation_curve.csv"), + format!("{header},n,seed_mean_error,seed_se\n{point},1000,0.05,0\n{point},{last},0.01,0\n"), + ) + .unwrap(); + } + let loaded = SaturationCurves::load(&dir).unwrap(); + std::fs::remove_dir_all(&dir).unwrap(); + let points = &loaded.points_by_sketch["kll-percall"]; + assert_eq!(points.len(), 1); + assert_eq!(points[0].n_sat, Some(1e8)); + assert_eq!(points[0].curve, vec![(1e3, 0.05, 0.0), (1e9, 0.01, 0.0)]); + assert_eq!(points[0].shape, MeasuredShape::Pareto { tail_index: 2.0 }); + assert_eq!(points[0].params, BTreeMap::from([("k".to_string(), 200.0)])); + } +} diff --git a/scripts/export_rqe_optimizer_costs.sh b/scripts/export_rqe_optimizer_costs.sh index e83b7f1f..9d2cb55e 100755 --- a/scripts/export_rqe_optimizer_costs.sh +++ b/scripts/export_rqe_optimizer_costs.sh @@ -68,11 +68,14 @@ MERGE_SHARDS=16 MERGE_ACCURACY_SHARDS=(4 16 64 256 1024) # The swept parameters, every one of them. Exact accumulators take none. +# Sketch configs are points of the saturation grid (scripts/study_saturation.py): +# the optimizer reads their accuracy from its curves, and a config with no +# curve is never eligible (#156). HLL_PRECISIONS=(12 14) -KLL_KS=(200 500) +KLL_KS=(200 800) DD_ALPHAS=(0.01 0.02) CMS_HEAP_ROWS=(3 5) -CMS_HEAP_COLS=(2048) +CMS_HEAP_COLS=(1024 4096) UNIVMON_CONFIGS=( "heap_size=1000 sketch_row=5 sketch_col=2048 layer_size=8" "heap_size=500 sketch_row=3 sketch_col=1024 layer_size=6"