diff --git a/docs/rqe_sketch_deployment_v1.md b/docs/rqe_sketch_deployment_v1.md index 8e1f13ea..3a28de9a 100644 --- a/docs/rqe_sketch_deployment_v1.md +++ b/docs/rqe_sketch_deployment_v1.md @@ -228,9 +228,14 @@ lookback: series per group times scrapes per lookback. accuracy. Past the last, the plateau if the point saturated, else none. - The curve's `error_metric` must be the family's metric; `SaturationCurves::load` refuses a curve for another metric as a stale study. -- 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. +- For sketches that merge exactly, 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 (#131): an answer merged from `m = S_i / D.x` + windows reads the study's merge curve (`saturation_merge_curve.csv`, the + sketch merged from `m` shards of the same items) at the lookback's item + count. Between measured shard counts, the worse; past the largest, the + largest. Without a merge curve, a merged KLL or top-k answer has no + accuracy (#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 @@ -473,8 +478,9 @@ each phase's CPU and memory, together with the selected deployment mapping. ## v1 scope and TODOs -- **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). +- **Accuracy after merging:** KLL and top-k read merge curves up to the + largest measured shard count (64) and N = 1e7; past those, the largest + count, and the plateau if the plain curve saturated (#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/autosketch_vs_asap.rs b/rqe-optimizer/examples/autosketch_vs_asap.rs new file mode 100644 index 00000000..a5f78901 --- /dev/null +++ b/rqe-optimizer/examples/autosketch_vs_asap.rs @@ -0,0 +1,847 @@ +//! AutoSketch vs. the ASAP planner (ProjectASAP/ASAPQuery#777, paper §6.3). +//! +//! One invocation evaluates one workload and writes one JSON file: +//! +//! ```text +//! autosketch_vs_asap traces --dataset alibaba_v2022 --saturation-dir DIR --out FILE [--runs 5] +//! autosketch_vs_asap synthetic --table FILE --target default --saturation-dir DIR --out FILE [--runs 1] +//! +//! Options for both: --slas-ms 0.01,inf (override the SLA grid), --weights +//! cpu,fargate (a subset of the objective weights), --no-chosen (drop +//! per-RQE choices). The synthetic --target is a strictness level (loose, +//! default, strict). +//! ``` +//! +//! Every plan is scored by `analytical_cost_model::score` (#145): mean CPU and +//! memory per phase. ASAP and PerQuery-CostAware minimize +//! `w_cpu · CPU + w_mem · memory_GiB` at each weight setting of #777 §4: +//! CPU only, and Fargate's per-vCPU and per-GB prices. AutoSketch's plan does +//! not depend on the weights and is scored under each. +//! +//! Methods: +//! - **ASAP:** `milp::minimize` over every RQE jointly, with one absolute +//! latency SLA for every RQE, swept over the SLA grid. RQEs that no eligible +//! deployment can serve within an SLA are excluded from every method at +//! that SLA, and listed in the output. +//! - **AutoSketch-Adapted:** `autosketch::plan`, one search and one dedicated +//! deployment per RQE, accuracy only. +//! - **PerQuery-CostAware:** `milp::minimize` on each RQE alone, under the +//! same SLA; the per-RQE deployments are kept separate. +//! +//! `traces` reads `data/autosketch-eval/table.json`; `synthetic` reads one +//! table written by `export_autosketch_eval_table.py --synthetic`. A table +//! gives the workload: RQEs, and each stream's groups, rate and data shape +//! (the family's `grid_param`/`grid_K`, the worst case rounded to the grid). +//! Each stream (`traces`: dataset, query and kind; `synthetic`: the table's +//! `stream`) is one metric, which is where sharing comes from. +//! +//! Costs and accuracy come from `--saturation-dir`, the same inputs the +//! planner reads (#174): the cost table `optimizer_cost/rqe_atomic_costs.json` +//! (one row per config, at the evaluation dataset's shape) and the saturation +//! curves. ASAP reads `SaturationCurves::accuracy` (the curve at the +//! lookback's items; KLL and top-k merged from `L/x` shards read the merge +//! curve, #158); AutoSketch reads `autosketch_accuracy` (one unmerged sketch +//! per query window, no saturation requirement). + +use std::collections::{BTreeMap, BTreeSet}; +use std::time::Instant; + +use rqe_optimizer::analytical_cost_model::{score, PlanCost}; +use rqe_optimizer::autosketch; +use rqe_optimizer::candidates::{build_all_candidates, eligible_deployments_for, is_eligible}; +use rqe_optimizer::milp::{minimize, MilpSolution, Objective}; +use rqe_optimizer::saturation::{DataShape, SaturationCurves, COST_TABLE}; +use rqe_optimizer::{ + validate_facts, AtomicCostEntry, AtomicCostTable, Capability, Deployment, LabelSet, Mapping, + MetricFacts, Millis, Raqe, WorkloadFacts, +}; +use serde_json::{json, Value}; + +const TRACES_TABLE: &str = "rqe-optimizer/data/autosketch-eval/table.json"; +/// Absolute per-RQE latency SLAs in ms (#777 §5), plus no SLA. +const SLAS_MS: [f64; 5] = [0.01, 0.1, 1.0, 10.0, f64::INFINITY]; +/// Objective weights of #777 §4: CPU only, then AWS Fargate (us-east-1, +/// Linux/x86) $/vCPU-hour and $/GB-hour, so the objective reads in $/hour. +const WEIGHTS: [(&str, f64, f64); 2] = [("cpu", 1.0, 0.0), ("fargate", 0.0405, 0.00445)]; +/// Every stream is scraped once a second: windows and slides are whole +/// seconds, and the stream's samples/s become its series count. +const SCRAPE_MS: Millis = 1_000; +const BYTES_PER_GIB: f64 = (1u64 << 30) as f64; +const SEED: u64 = 7; +/// AutoSketch's benchmark time per evaluated configuration, at the paper's +/// rate: 1–2 minutes each (NSDI '24, §5.2 and Exp#9); we take 1 minute. +const PAPER_SECS_PER_PROBE: f64 = 60.0; +/// Items per benchmark run in AutoSketch's benchmark-time lower bound: +/// sketch-bench's measurement size. +const N_BENCH: f64 = 1e8; + +/// (distinct probed configs, lower-bound seconds) for probes given as +/// (config name, insert CPU per item, query-phase CPU): each distinct config +/// is benchmarked once, inserting N_BENCH items and running one query phase. +fn benchmark_lower_bound(probes: &[(String, f64, f64)]) -> (usize, f64) { + let mut seen = BTreeSet::new(); + let mut secs = 0.0; + for (name, insert, phase) in probes { + if seen.insert(name) { + secs += N_BENCH * insert + phase; + } + } + (seen.len(), secs) +} + +/// One workload: RQEs, one metric per stream, the cost entries each stream +/// may use, and the curves every accuracy is read from. +struct Workload { + name: String, + raqes: Vec, + facts: WorkloadFacts, + costs: BTreeMap>, + curves: SaturationCurves, + notes: Vec, +} + +impl Workload { + /// ASAP's accuracy: the planner's own lookup. + fn asap_accuracy(&self, r: &Raqe, d: &Deployment) -> Option { + self.curves.accuracy(r, d, &self.facts) + } + + /// AutoSketch's: one unmerged sketch per query window. + fn autosketch_accuracy(&self, r: &Raqe, c: &AtomicCostEntry) -> Option { + self.curves.autosketch_accuracy(r, c, &self.facts) + } +} + +fn arg(args: &[String], name: &str) -> Option { + args.windows(2) + .find(|pair| pair[0] == name) + .map(|pair| pair[1].clone()) +} + +fn main() { + let args: Vec = std::env::args().collect(); + let out = arg(&args, "--out").expect("--out FILE is required"); + let runs: usize = arg(&args, "--runs").map_or(5, |r| r.parse().unwrap()); + let saturation_dir = arg(&args, "--saturation-dir").expect("--saturation-dir DIR is required"); + let workload = match args.get(1).map(String::as_str) { + Some("traces") => from_table( + TRACES_TABLE, + Some(&arg(&args, "--dataset").expect("--dataset")), + None, + &saturation_dir, + ), + Some("synthetic") => from_table( + &arg(&args, "--table").expect("--table"), + None, + Some(arg(&args, "--target").expect("--target")), + &saturation_dir, + ), + _ => panic!("usage: autosketch_vs_asap (traces|synthetic) ... --out FILE"), + }; + let slas: Vec = arg(&args, "--slas-ms").map_or(SLAS_MS.to_vec(), |list| { + list.split(',') + .map(|x| { + if x == "inf" { + f64::INFINITY + } else { + x.parse().unwrap() + } + }) + .collect() + }); + let weights: Vec<(&str, Objective)> = WEIGHTS + .iter() + .filter(|(name, ..)| { + arg(&args, "--weights").is_none_or(|list| list.split(',').any(|w| w == *name)) + }) + .map(|&(name, w_cpu, w_mem)| (name, Objective::AUCCost { w_cpu, w_mem })) + .collect(); + let mut result = evaluate(&workload, runs, &slas, &weights); + if args.iter().any(|a| a == "--no-chosen") { + for r in result["results"].as_array_mut().unwrap() { + r.as_object_mut().unwrap().remove("chosen"); + } + } + std::fs::write(&out, serde_json::to_string_pretty(&result).unwrap()).unwrap(); + eprintln!("wrote {out}"); +} + +// ---------------------------------------------------------------- workloads + +/// Per-capability accuracy targets for a strictness level of the synthetic +/// workload (#777 §5). Exact accumulators must have zero error. +fn strictness_target(level: &str, capability: Capability) -> f64 { + let column = match level { + "loose" => 0, + "default" => 1, + "strict" => 2, + other => panic!("unknown strictness level {other}"), + }; + let targets = match capability { + Capability::Quantile => [0.02, 0.01, 0.005], + Capability::TopKByValue | Capability::TopKByCount => [0.90, 0.95, 0.99], + Capability::Cardinality => [0.05, 0.02, 0.01], + Capability::Sum + | Capability::Count + | Capability::Min + | Capability::Max + | Capability::RateOrIncrease => [0.0; 3], + }; + targets[column] +} + +/// The table's capability for an RQE. `traces` key queries are group sums, +/// served by exact accumulators; its tables list sketch families only, so +/// those RQEs stay unservable until the table carries an exact row. +fn capability_of(r: &Value) -> Capability { + match (r["capability"].as_str(), r["kind"].as_str().unwrap()) { + (Some("sum"), _) | (Some("freq"), _) | (None, "keys") => Capability::Sum, + (Some("rate"), _) => Capability::RateOrIncrease, + // `topk(k, sum by (...) (...))`: ranked by summed value. + (Some("topk"), _) => Capability::TopKByValue, + (Some("quantile"), _) | (None, "values") => Capability::Quantile, + other => panic!("unknown capability {other:?}"), + } +} + +fn label_set(names: &[&str]) -> LabelSet { + names.iter().map(|s| s.to_string()).collect() +} + +/// One workload of an evaluation table: `traces`'s `dataset`, or a +/// `synthetic` table's only workload. `target` sets every RQE's accuracy +/// target to a strictness level. +fn from_table( + path: &str, + dataset: Option<&str>, + target: Option, + saturation_dir: &str, +) -> Workload { + let table: Value = serde_json::from_str(&std::fs::read_to_string(path).unwrap()).unwrap(); + let dir = std::path::Path::new(saturation_dir); + let curves = SaturationCurves::load(dir) + .unwrap_or_else(|e| panic!("couldn't load saturation curves from {saturation_dir}: {e}")); + let cost_path = dir.join(COST_TABLE); + let cost_table: AtomicCostTable = + serde_json::from_str(&std::fs::read_to_string(&cost_path).unwrap_or_else(|e| { + panic!( + "couldn't read {} ({e}); run study_saturation.py --phase optimizer-cost", + cost_path.display() + ) + })) + .unwrap(); + let workload = table["workloads"] + .as_array() + .unwrap() + .iter() + .find(|w| dataset.is_none_or(|d| w["dataset"] == d)) + .unwrap_or_else(|| panic!("no workload {dataset:?} in {path}")); + let dataset = workload["dataset"].as_str().unwrap(); + let mut raqes = Vec::new(); + let mut facts = WorkloadFacts::new(); + // Queries one evaluation issues per instance, per stream (max over its RQEs). + let mut queries_per_stream: BTreeMap = BTreeMap::new(); + let mut skipped_families = 0; + let mut rounded_up_streams = BTreeSet::new(); + for r in workload["rqes"].as_array().unwrap() { + let id = r["id"].as_str().unwrap().to_string(); + let stream = match r["stream"].as_str() { + Some(stream) => format!("{dataset}/{stream}"), + None => format!( + "{}/{}/{}", + dataset, + r["query_id"].as_str().unwrap(), + r["kind"].as_str().unwrap() + ), + }; + let capability = capability_of(r); + let rate = r["label_set"]["arrival_rate_per_sec"].as_f64().unwrap(); + let groups = r["label_set"]["groups"].as_u64().unwrap(); + let grouping = if groups > 1 { + label_set(&["g"]) + } else { + LabelSet::new() + }; + // One metric per stream, labels {g, x}: `g` is the grouping (card = + // groups), and with one scrape a second `{g, x}` has one series per + // sample/s. A stream's RQEs differ only in range, so take the + // largest per-RQE rate (conservative). Rates below `groups` + // samples/s are rounded up so every group has a series. + let series = (rate * SCRAPE_MS as f64 / 1000.0).round() as u64; + if series < groups { + rounded_up_streams.insert(stream.clone()); + } + let metric = facts.entry(stream.clone()).or_insert_with(|| MetricFacts { + labels: label_set(&["g", "x"]), + scrape_interval_ms: SCRAPE_MS, + cardinality: [ + (label_set(&["g", "x"]), 0), + (label_set(&["g"]), groups), + (LabelSet::new(), 1), + ] + .into(), + value_range: None, + data_shape: BTreeMap::new(), + }); + let total = metric.cardinality.get_mut(&label_set(&["g", "x"])).unwrap(); + *total = (*total).max(series).max(groups); + // Queries one evaluation issues per instance (the table's q_r); tables + // written before it existed charge one query phase per evaluation. + let queries = r["queries_per_instance"].as_f64().unwrap_or(1.0); + let q = queries_per_stream.entry(stream.clone()).or_default(); + *q = q.max(queries); + + // The first family of the RQE's capability gives its target and the + // stream's data shape; the table rounds both to the grid. + let mut rqe_target = None; + for family in r["families"].as_array().unwrap() { + let sketch = family["sketch"].as_str().unwrap(); + if !capability.families().contains(&sketch) { + // e.g. `traces`' frequency sketches, or top-k for a sum query. + skipped_families += 1; + continue; + } + if rqe_target.is_none() { + rqe_target = Some(match target.as_deref() { + Some(level) => strictness_target(level, capability), + None => family["target"].as_f64().unwrap(), + }); + } + if let Some(shape) = data_shape(family) { + // Several RQEs of a stream: the harder side of each parameter. + metric + .data_shape + .entry(grouping.clone()) + .and_modify(|old| { + old.zipf_s = old.zipf_s.min(shape.zipf_s); + old.distinct_keys = old.distinct_keys.max(shape.distinct_keys); + old.tail_index = old.tail_index.min(shape.tail_index); + }) + .or_insert(shape); + } + } + raqes.push(Raqe { + id: id.clone(), + capability, + lookback_ms: (r["lookback_secs"].as_f64().unwrap() * 1000.0) as Millis, + interval_ms: (r["interval_secs"].as_f64().unwrap() * 1000.0) as Millis, + metric: stream, + spatial_filter: String::new(), + grouping_labels: grouping, + // An RQE with no family of its capability is unservable; its + // target is never read. + accuracy_sla: rqe_target.unwrap_or(0.0), + latency_sla_ms: None, + }); + } + // Every stream may use every row; one evaluation runs q_r queries per + // instance. + let costs = queries_per_stream + .into_iter() + .map(|(stream, queries)| { + let rows = cost_table + .iter() + .map(|row| AtomicCostEntry { + query_cpu_secs: queries * row.query_cpu_secs, + ..row.clone() + }) + .collect(); + (stream, rows) + }) + .collect(); + let name = match &target { + Some(t) => format!("{dataset}/t{t}"), + None => format!("traces/{dataset}"), + }; + Workload { + name, + raqes, + facts, + costs, + curves, + notes: vec![ + format!("{skipped_families} families skipped: they serve another capability"), + "one metric per stream: groups from the table, samples/s = max over the stream's RQEs, scraped once a second".into(), + format!("streams below one sample/s per group, rounded up: {rounded_up_streams:?}"), + format!("costs: {COST_TABLE} under the saturation dir, one row per config"), + "data shape per stream: the families' grid_param/grid_K, the harder side over its RQEs".into(), + "query_cpu_secs = q_r queries per instance per evaluation".into(), + ], + } +} + +/// A table family's grid point as a curve key: Zipf θ and K, or the Pareto +/// tail index for quantiles. The unused fields are never read. +fn data_shape(family: &Value) -> Option { + let param = family["grid_param"].as_f64()?; + let keys = &family["grid_K"]; + match keys + .as_f64() + .or_else(|| keys.as_str().and_then(|k| k.parse().ok())) + { + Some(keys) => Some(DataShape { + zipf_s: param, + distinct_keys: keys, + tail_index: f64::INFINITY, + }), + None => Some(DataShape { + zipf_s: f64::INFINITY, + distinct_keys: 0.0, + tail_index: param, + }), + } +} + +// --------------------------------------------------------------- evaluation + +fn median(mut v: Vec) -> f64 { + v.sort_by(f64::total_cmp); + v[v.len() / 2] +} + +fn group_by_stream(raqes: &[Raqe]) -> BTreeMap<&str, Vec> { + let mut out: BTreeMap<&str, Vec> = BTreeMap::new(); + for r in raqes { + out.entry(r.metric.as_str()).or_default().push(r.clone()); + } + out +} + +/// ASAP candidates, built per stream from that stream's entries. +fn candidates(w: &Workload, raqes: &[Raqe]) -> Vec { + group_by_stream(raqes) + .into_iter() + .flat_map(|(stream, group)| { + let costs = w.costs.get(stream).map_or(&[][..], Vec::as_slice); + build_all_candidates(&group, costs, &w.facts, false, &|r, d| { + w.asap_accuracy(r, d) + }) + }) + .collect() +} + +/// AutoSketch, searched per stream over that stream's entries. +fn autosketch_plan( + w: &Workload, + raqes: &[Raqe], +) -> Result> { + let mut deployments = Vec::new(); + let mut searches = Vec::new(); + let mut unservable = Vec::new(); + let mut order = Vec::new(); + for (stream, group) in group_by_stream(raqes) { + let costs = w.costs.get(stream).map_or(&[][..], Vec::as_slice); + let oracle = |r: &Raqe, c: &AtomicCostEntry| w.autosketch_accuracy(r, c); + match autosketch::plan(&group, costs, SEED, false, oracle) { + Ok(plan) => { + deployments.extend(plan.deployments); + searches.extend(plan.searches); + order.extend(group.iter().map(|r| r.id.clone())); + } + Err(ids) => unservable.extend(ids), + } + } + if !unservable.is_empty() { + return Err(unservable); + } + // Back to the caller's RQE order. + let position: BTreeMap<&str, usize> = order + .iter() + .enumerate() + .map(|(i, id)| (id.as_str(), i)) + .collect(); + let mapping: Mapping = raqes.iter().map(|r| position[r.id.as_str()]).collect(); + Ok(autosketch::AutoSketchPlan { + deployments, + mapping, + searches, + }) +} + +/// Modeled latency of `r` on `d`, in ms. +fn latency_ms(r: &Raqe, d: &Deployment, facts: &WorkloadFacts) -> f64 { + score( + std::slice::from_ref(r), + std::slice::from_ref(d), + &vec![0], + facts, + ) + .query_latency_ms[0] +} + +fn phase(p: &rqe_optimizer::analytical_cost_model::PhaseCost) -> Value { + json!({"cpu": p.cpu_secs_per_sec, "gib": p.memory_bytes / BYTES_PER_GIB}) +} + +fn summarize( + w: &Workload, + raqes: &[Raqe], + deployments: &[Deployment], + mapping: &Mapping, + cost: &PlanCost, + objective: &Objective, + sla_ms: Option, +) -> Value { + let active: BTreeSet = mapping.iter().copied().collect(); + let violations = cost + .query_latency_ms + .iter() + .filter(|&&l| sla_ms.is_some_and(|sla| l > sla * (1.0 + 1e-9))) + .count(); + let chosen: Vec = raqes + .iter() + .zip(mapping) + .zip(&cost.query_latency_ms) + .map(|((r, &di), &latency)| { + let d = &deployments[di]; + json!({ + "rqe": r.id, + "sketch": d.config.sketch, + "params": d.config.sketch_config["params"], + "window_secs": d.window_ms / 1000, + "slide_secs": d.slide_ms / 1000, + "merged_shards": r.lookback_ms / d.window_ms, + "latency_ms": latency, + "deployment": di, + "asap_accuracy": w.asap_accuracy(r, d), + }) + }) + .collect(); + json!({ + "objective": objective.value(cost), + "cpu": cost.cpu_secs_per_sec(), + "gib": cost.memory_bytes() / BYTES_PER_GIB, + "phases": { + "ingest": phase(&cost.ingest), + "merge": phase(&cost.merge), + "query": phase(&cost.query), + "storage": phase(&cost.storage), + }, + "latency_violations": violations, + "max_latency_ms": cost.query_latency_ms.iter().copied().fold(0.0, f64::max), + "median_latency_ms": (!cost.query_latency_ms.is_empty()) + .then(|| median(cost.query_latency_ms.clone())), + "feasible": violations == 0, + "active_deployments": active.len(), + "chosen": chosen, + }) +} + +/// A MILP solution's active deployments, and which one serves each RQE. +fn planned(s: &MilpSolution) -> (Vec, Mapping) { + ( + s.deployments.iter().map(|p| p.deployment.clone()).collect(), + s.raqes.iter().map(|p| p.deployment).collect(), + ) +} + +fn sla_label(sla_ms: f64) -> Value { + if sla_ms.is_finite() { + json!(sla_ms) + } else { + json!("inf") + } +} + +fn evaluate(w: &Workload, runs: usize, slas: &[f64], weights: &[(&str, Objective)]) -> Value { + let asap_accuracy = |r: &Raqe, d: &Deployment| w.asap_accuracy(r, d); + // Drop RQEs that either method cannot serve, so both plan the same batch. + let all_candidates = candidates(w, &w.raqes); + let asap_unservable: Vec = w + .raqes + .iter() + .filter(|r| { + eligible_deployments_for(r, &all_candidates, &w.facts, &asap_accuracy).is_empty() + }) + .map(|r| r.id.clone()) + .collect(); + let autosketch_unservable = autosketch_plan(w, &w.raqes).err().unwrap_or_default(); + let dropped: BTreeSet = asap_unservable + .iter() + .chain(&autosketch_unservable) + .cloned() + .collect(); + let raqes: Vec = w + .raqes + .iter() + .filter(|r| !dropped.contains(&r.id)) + .cloned() + .collect(); + eprintln!( + "{}: {} RQEs, {} dropped", + w.name, + raqes.len(), + dropped.len() + ); + if let Err(problems) = validate_facts(&raqes, &w.facts) { + panic!("invalid facts: {problems:?}"); + } + + // Fastest latency each RQE can reach on any eligible deployment. An RQE + // over an SLA is excluded from every method at that SLA. + let min_latency: Vec = raqes + .iter() + .map(|r| { + eligible_deployments_for(r, &all_candidates, &w.facts, &asap_accuracy) + .into_iter() + .map(|di| latency_ms(r, &all_candidates[di], &w.facts)) + .fold(f64::INFINITY, f64::min) + }) + .collect(); + + // AutoSketch: one plan, independent of the weights and the SLA. + let mut search_secs = Vec::new(); + let mut plan = None; + for _ in 0..runs { + let started = Instant::now(); + plan = Some(autosketch_plan(w, &raqes).expect("servable RQEs only")); + search_secs.push(started.elapsed().as_secs_f64()); + } + let plan = plan.unwrap(); + let probes: usize = plan.searches.iter().map(|s| s.probes.len()).sum(); + let probed: Vec<(String, f64, f64)> = plan + .searches + .iter() + .flat_map(|s| { + let r = raqes.iter().find(|r| r.id == s.raqe_id).unwrap(); + let costs = &w.costs[&r.metric]; + s.probes.iter().map(move |&p| { + let c = &costs[p]; + // One query of one instance stands in for the query phase. + let name = format!("{}/{}", c.sketch, c.sketch_config["params"]); + (name, c.insert_cpu_secs, c.query_cpu_secs) + }) + }) + .collect(); + let (distinct_probes, bench_secs) = benchmark_lower_bound(&probed); + let paper_secs = distinct_probes as f64 * PAPER_SECS_PER_PROBE; + + // PerQuery-CostAware's candidates: each RQE alone. Solved per SLA below. + let started = Instant::now(); + let singles: Vec> = raqes + .iter() + .map(|r| candidates(w, std::slice::from_ref(r))) + .collect(); + let singles_build_secs = started.elapsed().as_secs_f64(); + + let mut results = Vec::new(); + let mut sanity = Vec::new(); + let mut excluded_by_sla = serde_json::Map::new(); + for &sla in slas { + let sla_ms = sla.is_finite().then_some(sla); + let kept: Vec = (0..raqes.len()) + .filter(|&i| min_latency[i] <= sla) + .collect(); + let excluded: Vec<&str> = (0..raqes.len()) + .filter(|i| !kept.contains(i)) + .map(|i| raqes[i].id.as_str()) + .collect(); + excluded_by_sla.insert( + sla_label(sla).to_string().trim_matches('"').to_string(), + json!(excluded), + ); + if kept.is_empty() { + continue; + } + let sla_raqes: Vec = kept + .iter() + .map(|&i| Raqe { + latency_sla_ms: sla_ms, + ..raqes[i].clone() + }) + .collect(); + let identity: Mapping = (0..sla_raqes.len()).collect(); + + let mut build_secs = Vec::new(); + let mut asap_candidates = Vec::new(); + for _ in 0..runs { + let started = Instant::now(); + asap_candidates = candidates(w, &sla_raqes); + build_secs.push(started.elapsed().as_secs_f64()); + } + let auto_deployments: Vec = kept + .iter() + .map(|&i| plan.deployments[plan.mapping[i]].clone()) + .collect(); + let auto_cost = score(&sla_raqes, &auto_deployments, &identity, &w.facts); + + for (weight_name, objective) in weights { + let mut solve_secs = Vec::new(); + let mut solution = None; + for _ in 0..runs { + let started = Instant::now(); + let solved = minimize( + &sla_raqes, + &asap_candidates, + &w.facts, + *objective, + &asap_accuracy, + ); + solve_secs.push(started.elapsed().as_secs_f64()); + solution = Some(solved); + } + let asap = match solution.unwrap() { + Ok(s) => { + let (deployments, mapping) = planned(&s); + let mut v = summarize( + w, + &sla_raqes, + &deployments, + &mapping, + &s.plan_cost, + objective, + sla_ms, + ); + v["planning_secs"] = + json!(median(build_secs.clone()) + median(solve_secs.clone())); + v["candidate_build_secs"] = json!(median(build_secs.clone())); + v["milp_solve_secs"] = json!(median(solve_secs.clone())); + v["candidates"] = json!(asap_candidates.len()); + v + } + Err(e) => json!({"error": e.to_string()}), + }; + let mut auto = summarize( + w, + &sla_raqes, + &auto_deployments, + &identity, + &auto_cost, + objective, + sla_ms, + ); + auto["planning_secs"] = json!(median(search_secs.clone())); + auto["probes"] = json!(probes); + auto["benchmark_secs_lower_bound_nbench1e8"] = json!(bench_secs); + auto["benchmark_secs_paper_rate"] = json!(paper_secs); + // PerQuery-CostAware: each kept RQE alone under the SLA. + let started = Instant::now(); + let perquery_deployments: Result, String> = kept + .iter() + .zip(&sla_raqes) + .map(|(&i, r)| { + minimize( + std::slice::from_ref(r), + &singles[i], + &w.facts, + *objective, + &asap_accuracy, + ) + .map(|solved| { + let (deployments, mapping) = planned(&solved); + deployments[mapping[0]].clone() + }) + .map_err(|e| format!("{}: {e}", r.id)) + }) + .collect(); + let perquery_secs = started.elapsed().as_secs_f64() + singles_build_secs; + let mut perq = match perquery_deployments { + Ok(deployments) => { + let cost = score(&sla_raqes, &deployments, &identity, &w.facts); + summarize( + w, + &sla_raqes, + &deployments, + &identity, + &cost, + objective, + sla_ms, + ) + } + Err(e) => json!({"error": e}), + }; + perq["planning_secs"] = json!(perquery_secs); + + // ASAP meets every SLA, and costs no more than AutoSketch whenever + // AutoSketch meets it too, nor than PerQuery. + if asap.get("error").is_some() || asap["latency_violations"] != 0 { + sanity.push(json!({ + "check": "asap meets every SLA", "weights": weight_name, + "sla_ms": sla_label(sla), "asap": asap, + })); + } + if let (Some(a), Some(b)) = (asap["objective"].as_f64(), auto["objective"].as_f64()) { + if auto["feasible"] == true && a > b * (1.0 + 1e-6) { + let not_eligible: Vec<&str> = sla_raqes + .iter() + .zip(&auto_deployments) + .filter(|(r, d)| !is_eligible(r, d, &w.facts, &asap_accuracy)) + .map(|(r, _)| r.id.as_str()) + .collect(); + sanity.push(json!({ + "check": "asap <= autosketch", "weights": weight_name, + "sla_ms": sla_label(sla), "asap": a, "autosketch": b, + "autosketch_choices_ineligible_for_asap": not_eligible, + })); + } + } + if perq.get("error").is_some() || perq["latency_violations"] != 0 { + sanity.push(json!({"check": "perquery meets every SLA", "weights": weight_name, "sla_ms": sla_label(sla), "perquery": perq})); + } + if let (Some(a), Some(b)) = (asap["objective"].as_f64(), perq["objective"].as_f64()) { + if a > b * (1.0 + 1e-6) { + sanity.push(json!({"check": "asap <= perquery", "weights": weight_name, "sla_ms": sla_label(sla), "asap": a, "perquery": b})); + } + } + for (method, value) in [("asap", asap), ("autosketch", auto), ("perquery", perq)] { + let mut v = value; + v["method"] = json!(method); + v["weights"] = json!(weight_name); + v["sla_ms"] = sla_label(sla); + v["rqes"] = json!(sla_raqes.len()); + v["excluded_by_sla"] = json!(excluded.len()); + results.push(v); + } + } + } + json!({ + "workload": w.name, + "rqes": raqes.len(), + "dropped_unservable": {"asap": asap_unservable, "autosketch": autosketch_unservable}, + "streams": raqes.iter().map(|r| &r.metric).collect::>().len(), + "runs": runs, + "notes": w.notes, + "sla_grid_ms": slas.iter().map(|&l| sla_label(l)).collect::>(), + "weights": weights.iter().map(|(name, o)| { + let Objective::AUCCost { w_cpu, w_mem } = o; + json!({"name": name, "w_cpu": w_cpu, "w_mem": w_mem}) + }).collect::>(), + "min_latency_ms": raqes.iter().zip(&min_latency).map(|(r, &l)| (r.id.clone(), json!(l))).collect::>(), + "excluded_by_sla": excluded_by_sla, + "autosketch": { + "probes": probes, + "search_secs": median(search_secs), + "benchmark_secs_lower_bound_nbench1e8": bench_secs, + "distinct_probes": distinct_probes, + "benchmark_secs_paper_rate": paper_secs, + }, + "sanity_violations": sanity, + "results": results, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn benchmark_bound_counts_each_config_once_at_n_bench() { + let probes = [ + ("cms/a".to_string(), 2e-8, 0.5), + ("cms/b".to_string(), 1e-8, 0.25), + ("cms/a".to_string(), 2e-8, 0.5), + ]; + let (distinct, secs) = benchmark_lower_bound(&probes); + assert_eq!(distinct, 2); + assert!((secs - (1e8 * 2e-8 + 0.5 + 1e8 * 1e-8 + 0.25)).abs() < 1e-12); + } + + #[test] + fn exact_accumulators_must_have_zero_error_at_every_level() { + for level in ["loose", "default", "strict"] { + assert_eq!(strictness_target(level, Capability::Sum), 0.0); + assert_eq!(strictness_target(level, Capability::RateOrIncrease), 0.0); + } + assert_eq!(strictness_target("strict", Capability::Quantile), 0.005); + } +} diff --git a/rqe-optimizer/src/saturation.rs b/rqe-optimizer/src/saturation.rs index aafbf46a..7017c7da 100644 --- a/rqe-optimizer/src/saturation.rs +++ b/rqe-optimizer/src/saturation.rs @@ -3,10 +3,11 @@ //! 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. +//! RAQE's whole lookback: series per group times scrapes per lookback. For +//! sketches that merge exactly, that holds whether the answer is one sketch +//! or a merge of many smaller-window sketches. KLL and top-k merge lossily +//! (#131): an answer merged from `L / x` windows reads the curve of the sketch +//! merged from that many shards (#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`] @@ -56,6 +57,9 @@ struct GridPoint { n_sat: Option, /// `(n, seed_mean_error, seed_se)`, ascending in `n`. curve: Vec<(f64, f64, f64)>, + /// The same, for the sketch merged from `m` shards of the stream, keyed + /// by `m > 1` (`saturation_merge_curve.csv`, #131, #158). + merged: BTreeMap>, } /// How far a cost-table value may sit outside the curve's range at its N: @@ -112,26 +116,72 @@ impl GridPoint { /// 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); + curve_error_at(&self.curve, self.n_sat, n, direction) + } + + /// The error at `n` items of the sketch merged from `merges` instances: + /// the worse of the measured shard counts either side of `merges` (one + /// is the plain sketch), or the largest measured count past it. `None` + /// with no merge curve. Past the merge curve's last checkpoint, its last + /// value if the plain curve saturated by then. + fn merged_error_at(&self, merges: u64, n: f64, direction: AccuracyDirection) -> Option { + if merges <= 1 { + return self.error_at(n, direction); } + let (&largest, _) = self.merged.last_key_value()?; + let below = self.merged.range(..=merges).next_back().map(|(&m, _)| m); 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)) - } + .merged + .range(merges..) + .next() + .map_or(largest, |(&m, _)| m); + let at = |m: u64| match m { + 1 => self.error_at(n, direction), + m => curve_error_at(&self.merged[&m], self.n_sat, n, direction), + }; + let below = at(below.unwrap_or(1))?; + Some(worse(below, at(above)?, direction)) } } +/// [`GridPoint::error_at`] on any of the point's curves. +fn curve_error_at( + curve: &[(f64, f64, f64)], + n_sat: Option, + n: f64, + direction: AccuracyDirection, +) -> Option { + let &(first_n, ..) = curve.first()?; + let &(last_n, last_error, _) = curve.last()?; + if n < first_n { + return None; + } + // Past the last checkpoint, the plateau, if saturated by then: a merge + // curve stops at 1e7 while its point's n_sat may come from the 1e9 run. + if n >= last_n { + return n_sat.filter(|&sat| sat <= last_n).map(|_| last_error); + } + let above = curve.partition_point(|&(checkpoint, ..)| checkpoint < n); + let (checkpoint, error, _) = curve[above]; + if checkpoint == n { + Some(error) + } else { + Some(worse(error, curve[above - 1].1, direction)) + } +} + +/// KLL and the heap top-k sketches lose accuracy when merged (#131): a +/// merged answer reads their merge curve. The other sketches merge exactly. +fn merges_lossily(sketch: &str) -> bool { + matches!( + sketch, + "kll-percall" + | "cms-heap-topk-fastpath-vector2d" + | "countsketch-heap-topk-fastpath-vector2d" + | "univmon-topk" + ) +} + /// Whether some capability plans `sketch`. fn is_candidate(sketch: &str) -> bool { Capability::ALL @@ -163,8 +213,9 @@ pub struct SaturationCurves { } impl SaturationCurves { - /// Reads `saturation.csv` (for the saturation point) and `saturation_curve.csv` from - /// each of [`RUN_DIRS`] under `dir`. Fails on a candidate family's curve + /// Reads `saturation.csv` (for the saturation point), `saturation_curve.csv` + /// and, where present, `saturation_merge_curve.csv` from each of + /// [`RUN_DIRS`] under `dir`. Fails on a candidate family's curve /// for a metric other than its [`accuracy_key`]: that study is stale. pub fn load(dir: &Path) -> io::Result { type PointKey = (String, String, String, String, String); @@ -192,6 +243,7 @@ impl SaturationCurves { n => Some(parse_number(n)?), }, curve: Vec::new(), + merged: BTreeMap::new(), }; let sketch = row["sketch"].as_str(); if is_candidate(sketch) && point.error_metric != accuracy_key(sketch).0 { @@ -216,11 +268,57 @@ impl SaturationCurves { parse_number(&row["seed_se"])?, )); } - points.extend(run_points); + let merge_path = run_dir.join("saturation_merge_curve.csv"); + if merge_path.exists() { + for row in read_csv(&merge_path)? { + let Some((_, point)) = run_points.get_mut(&key(&row)) else { + return Err(invalid(format!( + "{run}: merge curve row for a point not in saturation.csv: {row:?}" + ))); + }; + let shards: u64 = row["shards"] + .parse() + .map_err(|_| invalid(format!("{run}: shards is not a count: {row:?}")))?; + if shards == 0 { + return Err(invalid(format!("{run}: zero shards: {row:?}"))); + } + if shards > 1 { + point.merged.entry(shards).or_default().push(( + parse_number(&row["n"])?, + parse_number(&row["seed_mean_error"])?, + parse_number(&row["seed_se"])?, + )); + } + } + } + // A later run's point wins, but keeps an earlier run's merge + // curves at shard counts it didn't measure (the 1e9 run measures + // none). + for (key, (sketch, mut point)) in run_points { + if let Some((_, earlier)) = points.remove(&key) { + for (shards, curve) in earlier.merged { + point.merged.entry(shards).or_insert(curve); + } + } + points.insert(key, (sketch, point)); + } + } + // A merged KLL or top-k answer needs a merge curve; without one every + // such deployment would silently have no accuracy. + if let Some(((sketch, config, ..), _)) = points.iter().find(|(_, (sketch, point))| { + is_candidate(sketch) && merges_lossily(sketch) && point.merged.is_empty() + }) { + return Err(invalid(format!( + "{sketch} {config} has no merge curve; rerun the study with \ + --merge-shards-list" + ))); } 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)); + for curve in point.merged.values_mut() { + 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 }) @@ -247,9 +345,14 @@ impl SaturationCurves { let points = self.bracketing_points(&deployment.config, shape)?; let covered = items_per_group(metric_facts, grouping, raqe.lookback_ms); let (_, direction) = accuracy_key(&deployment.config.sketch); + let merges = if merges_lossily(&deployment.config.sketch) { + raqe.lookback_ms / deployment.window_ms + } else { + 1 + }; let mut worst = None; for point in points { - let error = point.error_at(covered, direction)?; + let error = point.merged_error_at(merges, covered, direction)?; worst = Some(worst.map_or(error, |w| worse(w, error, direction))); } worst @@ -481,18 +584,23 @@ mod tests { 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. + /// falls with N; θ = 1.0 is worse, and K = 1e5 never saturates. Merged + /// from 4 shards it loses 0.02 more, from 16 shards 0.1. fn curves() -> SaturationCurves { + let curve = |offset: f64| { + vec![ + (1e3, 1.0 - offset, 0.0), + (1e4, 0.95 - offset, 0.01), + (1e5, 0.9 - offset, 0.0), + ] + }; 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), - ], + curve: curve(offset), + merged: BTreeMap::from([(4, curve(offset + 0.02)), (16, curve(offset + 0.1))]), }; SaturationCurves { points_by_sketch: BTreeMap::from([( @@ -594,18 +702,100 @@ mod tests { } #[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. + fn merged_answers_read_the_merge_curve_at_the_lookback_count() { + // 10 items/s: the 1e4 s lookback covers 1e5 items whatever the + // window; top-k reads the curve of the sketch merged from L/x. + let facts = workload(10, shape(1.2, 1e3)); + let r = topk_raqe(10_000_000); + let at = |window_ms| curves().accuracy(&r, &deployment(TOPK, window_ms), &facts); + // One window: the plain curve. + assert_eq!(at(10_000_000), Some(0.9)); + // 4 windows: the 4-shard curve. + assert!((at(2_500_000).unwrap() - 0.88).abs() < 1e-12); + // 10 windows sit between 4 and 16 shards: the worse, 16. + assert!((at(1_000_000).unwrap() - 0.8).abs() < 1e-12); + // 100 windows: past the largest measured count, that count. + assert!((at(100_000).unwrap() - 0.8).abs() < 1e-12); + } + + /// Past a merge curve's last checkpoint, its plateau only if the point + /// saturated by then: n_sat may come from a longer plain run. + #[test] + fn a_merge_curve_extrapolates_only_from_a_saturation_it_covers() { + let mut curves = curves(); + let point = &mut curves.points_by_sketch.get_mut(TOPK).unwrap()[1]; + let higher = AccuracyDirection::HigherIsBetter; + assert!((point.merged_error_at(4, 1e9, higher).unwrap() - 0.88).abs() < 1e-12); + point.n_sat = Some(1e8); + assert_eq!(point.merged_error_at(4, 1e9, higher), None); + } + + /// Without a merge curve, a merged top-k answer has no known accuracy; + /// one window still reads the plain curve. + #[test] + fn a_lossy_merge_without_a_merge_curve_is_unknown() { + let mut curves = curves(); + for point in curves.points_by_sketch.get_mut(TOPK).unwrap() { + point.merged.clear(); + } 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) - ); + assert_eq!( + curves.accuracy(&r, &deployment(TOPK, 1_000_000), &facts), + None + ); + assert_eq!( + curves.accuracy(&r, &deployment(TOPK, 10_000_000), &facts), + Some(0.9) + ); + } + + #[test] + fn load_reads_merge_curves_beyond_one_shard() { + let dir = std::env::temp_dir().join(format!("rqe-merge-{}", std::process::id())); + let header = "family,sketch,config,dist,param,cardinality"; + let point = "quantile,kll-percall,k=200,pareto,2.0,"; + for run in RUN_DIRS { + std::fs::create_dir_all(dir.join(run)).unwrap(); + std::fs::write( + dir.join(run).join("saturation.csv"), + format!( + "{header},n_sat,final_error,error_metric\n{point},1000,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.01,0\n"), + ) + .unwrap(); } + // One shard is the plain curve. The later run measured only 16 + // shards: it keeps the earlier run's 4. + let merge = |rows: &str| format!("{header},n,shards,seed_mean_error,seed_se\n{rows}"); + let merge_path = |run| dir.join(run).join("saturation_merge_curve.csv"); + std::fs::write( + merge_path(RUN_DIRS[0]), + merge(&format!("{point},1000,1,0.01,0\n{point},1000,4,0.02,0\n")), + ) + .unwrap(); + std::fs::write( + merge_path(RUN_DIRS[1]), + merge(&format!("{point},1000,16,0.03,0\n")), + ) + .unwrap(); + let loaded = SaturationCurves::load(&dir); + // A merged KLL with no merge curve anywhere is refused. + std::fs::remove_file(merge_path(RUN_DIRS[0])).unwrap(); + std::fs::write(merge_path(RUN_DIRS[1]), merge("")).unwrap(); + let missing = SaturationCurves::load(&dir); + std::fs::remove_dir_all(&dir).unwrap(); + let point = &loaded.unwrap().points_by_sketch["kll-percall"][0]; + assert_eq!( + point.merged, + BTreeMap::from([(4, vec![(1e3, 0.02, 0.0)]), (16, vec![(1e3, 0.03, 0.0)])]) + ); + assert!(missing.unwrap_err().to_string().contains("no merge curve")); } #[test] @@ -711,6 +901,11 @@ mod tests { format!("{header},n,seed_mean_error,seed_se\n{point},1000,0.05,0\n{point},{last},0.01,0\n"), ) .unwrap(); + std::fs::write( + dir.join(run).join("saturation_merge_curve.csv"), + format!("{header},n,shards,seed_mean_error,seed_se\n{point},1000,4,0.06,0\n"), + ) + .unwrap(); } let loaded = SaturationCurves::load(&dir).unwrap(); std::fs::remove_dir_all(&dir).unwrap(); diff --git a/scripts/export_autosketch_eval_table.py b/scripts/export_autosketch_eval_table.py index 08092466..00833112 100644 --- a/scripts/export_autosketch_eval_table.py +++ b/scripts/export_autosketch_eval_table.py @@ -24,6 +24,14 @@ BOOM rows have no timestamps, window or N in skew_summary.csv; boom_query() derives them from the 20 equal chunks per series and records the assumptions. + +--synthetic writes the synthetic PromQL workload instead (plan section 6, +"Synthetic workload"): one table per workload-grid point under --out (a +directory), from synthetic_queries(), plus plan.tsv. These tables hold the +workload only: RQEs, streams, targets and each family's data shape. The +evaluation reads costs and accuracy from the saturation study, like the +planner (sketch-bench #174), so --synthetic takes no curves or costs. Sum and +rate queries name exact accumulators. """ import argparse @@ -32,6 +40,7 @@ import json import math import os +import random import re import subprocess import sys @@ -41,7 +50,7 @@ from recommend_config import ( # noqa: E402 DEFAULT_TARGETS, FAMILIES, KIND_FAMILIES, MERGE_INEXACT, cost_of, error_at, grid_point, load_merge_curves, load_saturation, nominal_size, num, read_summary) -from study_saturation import load_curves # noqa: E402 +from study_saturation import COST_KEYS, COST_PARETO_ALPHA, COST_THETA, load_curves # noqa: E402 SHARD_COUNTS = [1, 4, 16, 64] CAPABILITY = {"cms": "freq", "countsketch": "freq", "topk": "topk", @@ -146,6 +155,187 @@ def queries(path): return out +# synthetic workload (ProjectASAP/ASAPQuery#777 section 6, "Data model and +# scale"): one metric with C series (label_0, the top-k key) in J jobs, each +# series scraped every 5 ms. The data is fixed; the grid varies the queries. +# The data is the cost table's (study_saturation.py --phase optimizer-cost): +# Zipf θ over C keys, and quantiles on Pareto a, so costs, curves and the +# evaluation share one dataset. +SYNTHETIC_SERIES = COST_KEYS +SYNTHETIC_JOBS = 10 +SYNTHETIC_SAMPLES_PER_SEC = 200 +SYNTHETIC_THETA = COST_THETA +SYNTHETIC_ALPHA = COST_PARETO_ALPHA +SYNTHETIC_WINDOWS = "15m,1h,6h,24h" +SYNTHETIC_INTERVAL = 60 +SYNTHETIC_QUANTILES = ["0.5", "0.75", "0.9", "0.95", "0.99"] +SYNTHETIC_FAMILIES = {"sum": ["exact-sum"], "rate": ["exact-increase"], "topk": ["topk"], + "quantile": ["kll", "dd"]} +EXACT_SKETCH = {"exact-sum": "exact-sum", "exact-increase": "exact-increase"} +WINDOW_SECS = {"1m": 60, "5m": 300, "10m": 600, "15m": 900, "1h": 3600, "6h": 21600, + "24h": 86400} +DASHBOARD_QUANTILES = ["0.5", "0.9", "0.99"] +DASHBOARD_FIXED_WINDOWS = ["15m", "1h"] # D4 and D5 +# Shared replicas: each replica draws windows, quantiles and an interval. +SHARED_WINDOWS = 3 +SHARED_QUANTILES = 3 +SHARED_INTERVALS = [10, 60, 300] +SHARED_SEED = 7 +TEMPLATE_SETS = ["dashboard", "all"] +# Workload grid (#777 section 6): the default, then one dimension at a time. +SYNTHETIC_DEFAULT = {"templates": "dashboard", "shared": 1, "target": "default"} +SYNTHETIC_GRID = { + "templates": TEMPLATE_SETS, + "shared": [1, 8, 64], + "target": ["loose", "default", "strict"], +} + + +def synthetic_plan(): + """The default point and every dimension varied alone, each a full dict.""" + points = [dict(SYNTHETIC_DEFAULT)] + for dim, values in SYNTHETIC_GRID.items(): + points += [{**SYNTHETIC_DEFAULT, dim: v} for v in values] + unique = [] + for p in points: + if p not in unique: + unique.append(p) + return unique + + +def point_id(point): + """`templates=dashboard/shared=8`: the table's name segments.""" + return f"templates={point['templates']}/shared={point['shared']}" + + +def synthetic_queries(windows=SYNTHETIC_WINDOWS, interval=SYNTHETIC_INTERVAL, + templates="dashboard", quantiles=None, dataset=None, + range_with_interval=False, seen=None): + """The `synthetic` queries as read_summary-shaped query-ranges, one per RQE. + + Streams: `series/*` hold one instance per series (C groups), `job/*` one per + job (J groups), and `label_0/*` one top-k sketch over all C series as keys. + `*/dist` streams feed quantile sketches, `*/value` sums, `*/increment` + increases. Spatial templates repeat every 1 s and read the last second + (S = T = 1 s); temporal ones repeat every `interval` seconds over each + lookback in `windows`. RQEs that repeat another's query and range (template + 10, D5) are kept once. + """ + rate = SYNTHETIC_SERIES * SYNTHETIC_SAMPLES_PER_SEC # samples/s over all series + dataset = dataset or "synthetic/" + point_id({"templates": templates, "shared": 1}) + ranges = [(w, WINDOW_SECS[w]) for w in windows.split(",")] + seen = set() if seen is None else seen + out = [] + + def label(rng): + return f"{rng}@{interval}s" if range_with_interval else rng + + def add(query_id, stream, capability, rng, s, t, groups, keys=None): + if (query_id, rng) in seen: + return + seen.add((query_id, rng)) + out.append({ + "dataset": dataset, "query_id": query_id, + "kind": "values" if capability == "quantile" else "keys", + "range": rng, "range_s": s, "step_s": t, "theta": SYNTHETIC_THETA, "K": keys, + "min_N": rate * s / groups, "max_N": rate * s / groups, + "alpha_rank": SYNTHETIC_ALPHA, "alpha_memory": SYNTHETIC_ALPHA, "tail_class": "", + "families": SYNTHETIC_FAMILIES[capability], "capability": capability, + "stream": stream, "groups": groups, "arrival_rate": rate, + "assumptions": [], + **DEFAULT_TARGETS, + }) + + C, J = SYNTHETIC_SERIES, SYNTHETIC_JOBS + if templates == "dashboard": + quantiles = quantiles or DASHBOARD_QUANTILES + for rng, s in ranges: + for q in quantiles: + add(f"quantile_over_time_p{q}", "series/dist", "quantile", label(rng), s, + interval, C) + add("sum_by_job_rate", "job/increment", "rate", label(rng), s, interval, J) + for q in quantiles: + add(f"quantile_by_job_p{q}", "job/dist", "quantile", "1s", 1, 1, J) + for rng, s in [(w, WINDOW_SECS[w]) for w in DASHBOARD_FIXED_WINDOWS]: + if (rng, s) not in ranges: + continue + add("topk32_sum_by_label0_rate", "label_0/increment", "topk", label(rng), s, + interval, 1, C) + for q in ["0.99", "0.5"]: # D5, the same RQEs as D1's + add(f"quantile_over_time_p{q}", "series/dist", "quantile", label(rng), s, + interval, C) + return out + quantiles = quantiles or SYNTHETIC_QUANTILES + add("sum_by_job", "job/value", "sum", "1s", 1, 1, J) + for q in quantiles: + add(f"quantile_by_job_p{q}", "job/dist", "quantile", "1s", 1, 1, J) + for rng, s in ranges: + lbl = label(rng) + add("topk32_sum_by_label0_sum_over_time", "label_0/value", "topk", lbl, s, interval, + 1, C) + add("sum_over_time", "series/value", "sum", lbl, s, interval, C) + for q in quantiles: + add(f"quantile_over_time_p{q}", "series/dist", "quantile", lbl, s, interval, C) + add("rate", "series/increment", "rate", lbl, s, interval, C) + add("sum_by_job_rate", "job/increment", "rate", lbl, s, interval, J) + add("sum_by_job_sum_over_time", "job/value", "sum", lbl, s, interval, J) + add("topk32_sum_by_label0_rate", "label_0/increment", "topk", lbl, s, interval, 1, C) + for q in ["0.9", "0.5"]: # template 10, the same RQEs as template 5's + add(f"quantile_over_time_p{q}", "series/dist", "quantile", lbl, s, interval, C) + return out + + +def shared_queries(point): + """`point["shared"]` replicas of the point's template set on the same streams. + Replica i draws SHARED_WINDOWS of the windows, SHARED_QUANTILES of the + set's quantiles and an interval from SHARED_INTERVALS (seeded); identical + RQEs across replicas are kept once.""" + dataset = "synthetic/" + point_id(point) + windows = SYNTHETIC_WINDOWS.split(",") + pool = DASHBOARD_QUANTILES if point["templates"] == "dashboard" else SYNTHETIC_QUANTILES + seen, out = set(), [] + for i in range(point["shared"]): + rng = random.Random(f"{SHARED_SEED}/{i}") + picked = rng.sample(windows, min(SHARED_WINDOWS, len(windows))) + quantiles = sorted(rng.sample(pool, min(SHARED_QUANTILES, len(pool))), key=float) + interval = rng.choice(SHARED_INTERVALS) + out += synthetic_queries(",".join(sorted(picked, key=WINDOW_SECS.get)), interval, + point["templates"], quantiles, dataset, True, seen) + return out + + +def synthetic_entry(q): + """The table entry for one `synthetic` query-range: the workload only. + Each family names its sketch, target and data shape (θ and K, or the + Pareto a for quantiles); exact accumulators need no shape.""" + families = [] + for family in q["families"]: + if family in EXACT_SKETCH: + families.append({"family": family, "sketch": EXACT_SKETCH[family], + "target": 0.0, "grid_param": None, "grid_K": None}) + continue + sketch, target_col, _ = FAMILIES[family] + quantile = q["capability"] == "quantile" + families.append({ + "family": family, "sketch": sketch, "target": q[target_col], + "grid_param": SYNTHETIC_ALPHA if quantile else SYNTHETIC_THETA, + "grid_K": None if quantile else q["K"], + }) + return { + "id": f"{q['dataset']}/{q['query_id']}/{q['kind']}/{q['range']}", + "dataset": q["dataset"], "query_id": q["query_id"], "kind": q["kind"], + "range": q["range"], "instant": False, + "lookback_secs": q["range_s"], "interval_secs": q["step_s"], + "min_N": q["min_N"], "max_N": q["max_N"], + "queries_per_instance": queries_per_instance(q), + "stream": q["stream"], "capability": q["capability"], + "label_set": {"groups": q["groups"], "arrival_rate_per_sec": q["arrival_rate"], + "keys_per_window": q["K"]}, + "assumptions": q["assumptions"], + "families": families, + } + + def rqe_entry(q, curves, curves_big, sat, sat_big, merge): """The table entry for one skew_summary query-range.""" instant = not q["range_s"] @@ -182,6 +372,9 @@ def rqe_entry(q, curves, curves_big, sat, sat_big, merge): "min_N": q["min_N"], "max_N": q["max_N"], # One sketch per query: group-by values are the sketch's keys, not # separate instances (README, "Label sets"). + # Queries one evaluation issues per sketch instance: a frequency query + # reads every key (K point queries); top-k and a quantile issue one. + "queries_per_instance": queries_per_instance(q), "label_set": {"groups": 1, "arrival_rate_per_sec": q["max_N"] / lookback if lookback else None, "keys_per_window": q["K"]}, @@ -191,6 +384,14 @@ def rqe_entry(q, curves, curves_big, sat, sat_big, merge): } +def queries_per_instance(q): + """Queries one evaluation issues per instance: a `traces` key query reads + every key (K point queries); an exact group value, a top-k list and a + quantile are one query each.""" + capability = q.get("capability") or ("freq" if q["kind"] == "keys" else "quantile") + return q["K"] if capability == "freq" else 1 + + def sha256(path): h = hashlib.sha256() with open(path, "rb") as f: @@ -199,15 +400,39 @@ def sha256(path): return h.hexdigest() -def build(args): - curves = load_curves(args.curves) - curves_big = load_curves(args.curves_big) if args.curves_big else {} - sat = load_saturation(args.saturation) - sat_big = load_saturation(args.saturation_big) - merge = {} +def load_inputs(args): + """Curves, saturation rows and merge curves; later files win per point.""" + curves, sat, merge = {}, {}, {} + for path in args.curves: + curves.update(load_curves(path)) + for path in args.saturation: + sat.update(load_saturation(path)) for path in args.merge_curves or []: for key, per_m in load_merge_curves(path).items(): merge.setdefault(key, {}).update(per_m) + curves_big = load_curves(args.curves_big) if args.curves_big else {} + return curves, curves_big, sat, load_saturation(args.saturation_big), merge + + +def build_synthetic(args): + """({file name: table}, plan rows) for the workload-grid points of + `synthetic_plan()`: one table per distinct table point, and one plan row + (table, target, replicas, result name) per point.""" + out, plan = {}, [] + for point in synthetic_plan(): + name = "synthetic-" + point_id(point).replace("/", "-").replace("=", "") + ".json" + if name not in out: + qs = (shared_queries(point) if point["shared"] > 1 else + synthetic_queries(templates=point["templates"], + dataset="synthetic/" + point_id(point))) + out[name] = table(args, {qs[0]["dataset"]: [synthetic_entry(q) for q in qs]}, []) + result = name[:-len(".json")] + f"-t{point['target']}.json" + plan.append((name, point["target"], result)) + return out, plan + + +def build(args): + curves, curves_big, sat, sat_big, merge = load_inputs(args) workloads = {} for q in queries(args.summary): if not q["step_s"]: @@ -216,8 +441,12 @@ def build(args): workloads.setdefault(q["dataset"], []).append(entry) ids = [r["id"] for rqes in workloads.values() for r in rqes] assert len(ids) == len(set(ids)), "RQE ids must be unique" - inputs = [p for p in [args.summary, args.curves, args.curves_big, args.saturation, + inputs = [p for p in [args.summary, *args.curves, args.curves_big, *args.saturation, args.saturation_big, *(args.merge_curves or [])] if p] + return table(args, workloads, inputs) + + +def table(args, workloads, inputs): return { "schema_version": 1, "plan": "ProjectASAP/ASAPQuery#777", @@ -232,10 +461,13 @@ def build(args): def main(argv=None): parser = argparse.ArgumentParser( description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) - parser.add_argument("--summary", required=True, help="ASAPQuery skew_summary.csv") - parser.add_argument("--curves", required=True) + parser.add_argument("--summary", help="ASAPQuery skew_summary.csv") + parser.add_argument("--synthetic", action="store_true", + help="write the `synthetic` workload-grid tables and plan.tsv into the " + "--out directory instead") + parser.add_argument("--curves", nargs="+", help="traces only") parser.add_argument("--curves-big") - parser.add_argument("--saturation", required=True) + parser.add_argument("--saturation", nargs="+", help="traces only") parser.add_argument("--saturation-big") parser.add_argument("--merge-curves", nargs="*", help="saturation_merge_curve.csv files") parser.add_argument("--revision", help="sketch-bench commit of the curves (default: HEAD)") @@ -244,12 +476,27 @@ def main(argv=None): if args.revision is None: args.revision = subprocess.run(["git", "rev-parse", "HEAD"], capture_output=True, text=True, check=False).stdout.strip() - table = build(args) + if args.synthetic: + os.makedirs(args.out, exist_ok=True) + tables, plan = build_synthetic(args) + for name, t in tables.items(): + with open(os.path.join(args.out, name), "w") as f: + json.dump(t, f, sort_keys=True) + # One line per run: table, accuracy target, result file name. + with open(os.path.join(args.out, "plan.tsv"), "w") as f: + for name, target, result in plan: + f.write(f"{name}\t{target}\t{result}\n") + print(f"Done. {len(tables)} `synthetic` tables and {len(plan)} planned runs in " + f"{args.out}", file=sys.stderr) + return 0 + if not (args.summary and args.curves and args.saturation): + parser.error("--summary, --curves and --saturation are required without --synthetic") + result = build(args) os.makedirs(os.path.dirname(args.out) or ".", exist_ok=True) with open(args.out, "w") as f: - json.dump(table, f, indent=1, sort_keys=True) + json.dump(result, f, indent=1, sort_keys=True) f.write("\n") - n = sum(len(w["rqes"]) for w in table["workloads"]) + n = sum(len(w["rqes"]) for w in result["workloads"]) print(f"Done. {n} RQEs in {args.out}", file=sys.stderr) return 0 diff --git a/scripts/plot_autosketch_vs_asap.py b/scripts/plot_autosketch_vs_asap.py new file mode 100644 index 00000000..895495dd --- /dev/null +++ b/scripts/plot_autosketch_vs_asap.py @@ -0,0 +1,134 @@ +#!/usr/bin/env python3 +"""Plot the AutoSketch vs. ASAP evaluation on the `traces` workloads +(ProjectASAP/ASAPQuery#777) from the JSON files of +`rqe-optimizer/examples/autosketch_vs_asap.rs`. + +Writes fig1_objective.png, fig4_objective_vs_sla.png and summary.md next to +the inputs. + +Usage: scripts/plot_autosketch_vs_asap.py DIR +""" + +import json +import pathlib +import sys + +import matplotlib + +matplotlib.use("Agg") +import matplotlib.pyplot as plt # noqa: E402 + +METHODS = ["asap", "perquery", "autosketch"] +LABELS = {"asap": "ASAP", "perquery": "PerQuery-CostAware", "autosketch": "AutoSketch-Adapted"} +COLORS = {"asap": "#1f77b4", "perquery": "#ff7f0e", "autosketch": "#2ca02c"} +WEIGHT_UNITS = {"cpu": "vCPU", "fargate": "$/hour"} + + +def sla_text(sla): + return "no SLA" if sla == "inf" else f"{sla:g} ms" + + +def load(directory): + runs = {} + for path in sorted(pathlib.Path(directory).glob("traces-*.json")): + data = json.loads(path.read_text()) + runs[data["workload"].split("/")[-1]] = data + return runs + + +def pick(run, method, weights, sla): + for r in run["results"]: + if r["method"] == method and r["weights"] == weights and str(r["sla_ms"]) == str(sla): + return r + return None + + +def objective(r): + return None if r is None or "objective" not in r else r["objective"] + + +def fig_objective(runs, out): + """Objective per dataset and method, no SLA, one panel per weight setting.""" + weights = [w["name"] for w in next(iter(runs.values()))["weights"]] + fig, axes = plt.subplots(1, len(weights), figsize=(5 * len(weights), 3.5), squeeze=False) + for ax, wname in zip(axes[0], weights): + names = list(runs) + for i, method in enumerate(METHODS): + ys = [objective(pick(runs[n], method, wname, "inf")) or 0 for n in names] + ax.bar([x + i * 0.27 for x in range(len(names))], ys, 0.27, + label=LABELS[method], color=COLORS[method]) + ax.set_xticks([x + 0.27 for x in range(len(names))], names, fontsize=8) + ax.set_yscale("log") + ax.set_ylabel(f"objective ({WEIGHT_UNITS.get(wname, wname)})") + ax.set_title(f"weights: {wname}", fontsize=9) + axes[0][0].legend(fontsize=7) + fig.tight_layout() + fig.savefig(out / "fig1_objective.png", dpi=150) + plt.close(fig) + + +def fig_objective_vs_sla(runs, out): + """Objective vs. the latency SLA, CPU-only weights, one panel per dataset.""" + fig, axes = plt.subplots(1, len(runs), figsize=(4 * len(runs), 3.2), squeeze=False) + for ax, (name, run) in zip(axes[0], runs.items()): + slas = run["sla_grid_ms"] + xs = list(range(len(slas))) + for method in METHODS: + points = [(x, objective(pick(run, method, "cpu", s))) for x, s in zip(xs, slas)] + points = [(x, y) for x, y in points if y is not None] + if points: + ax.plot(*zip(*points), marker="o", label=LABELS[method], color=COLORS[method]) + ax.set_xticks(xs, [sla_text(s) for s in slas], fontsize=7) + ax.set_yscale("log") + ax.set_title(name, fontsize=9) + ax.set_ylabel("CPU (vCPU)") + axes[0][0].legend(fontsize=7) + fig.tight_layout() + fig.savefig(out / "fig4_objective_vs_sla.png", dpi=150) + plt.close(fig) + + +def fmt(x, digits=3): + return "—" if x is None else f"{x:.{digits}g}" + + +def summary(runs, out): + lines = ["# AutoSketch vs. ASAP on the trace workloads\n"] + for name, run in runs.items(): + lines.append(f"## {name}\n") + lines.append(f"{run['rqes']} RQEs; dropped as unservable: " + f"{len(set(run['dropped_unservable']['asap']) | set(run['dropped_unservable']['autosketch']))}. " + f"Sanity violations: {len(run['sanity_violations'])}.\n") + for w in run["weights"]: + for sla in run["sla_grid_ms"]: + rows = [pick(run, m, w["name"], sla) for m in METHODS] + if not any(rows): + continue + base = objective(rows[0]) + lines.append(f"Weights {w['name']}, {sla_text(sla)}:\n") + lines.append("| method | objective | vs ASAP | CPU (vCPU) | GiB | deployments | " + "max latency (ms) | planning (s) |") + lines.append("|---|---|---|---|---|---|---|---|") + for method, r in zip(METHODS, rows): + if r is None or "error" in r: + lines.append(f"| {LABELS[method]} | {r and r.get('error')} |||||||") + continue + ratio = r["objective"] / base if base else None + lines.append( + f"| {LABELS[method]} | {fmt(r['objective'])} | {fmt(ratio)}× | " + f"{fmt(r['cpu'])} | {fmt(r['gib'])} | {r['active_deployments']} | " + f"{fmt(r['max_latency_ms'])} | {fmt(r['planning_secs'])} |") + lines.append("") + (out / "summary.md").write_text("\n".join(lines)) + + +def main(): + directory = pathlib.Path(sys.argv[1]) + runs = load(directory) + fig_objective(runs, directory) + fig_objective_vs_sla(runs, directory) + summary(runs, directory) + + +if __name__ == "__main__": + main() diff --git a/scripts/plot_autosketch_vs_asap_synthetic.py b/scripts/plot_autosketch_vs_asap_synthetic.py new file mode 100644 index 00000000..9df1108a --- /dev/null +++ b/scripts/plot_autosketch_vs_asap_synthetic.py @@ -0,0 +1,167 @@ +#!/usr/bin/env python3 +"""Plot the synthetic AutoSketch vs. ASAP evaluation (ProjectASAP/ASAPQuery#777, +section 7) from the JSON files of `autosketch_vs_asap synthetic`. + +Writes, next to the inputs: + fig_objective_vs_latency.png default workload: objective vs. achieved max + latency over the SLA grid, one panel per weights + fig_objective_by_dimension.png objective with no SLA, each grid dimension + varied alone (templates, shared replicas, + strictness) + fig_planning_time.png planning time vs. RQEs (shared replicas) + summary.md every workload, method, weights and SLA + +Usage: scripts/plot_autosketch_vs_asap_synthetic.py DIR +""" + +import json +import pathlib +import re +import sys + +import matplotlib + +matplotlib.use("Agg") +import matplotlib.pyplot as plt # noqa: E402 + +METHODS = ["asap", "perquery", "autosketch"] +LABELS = {"asap": "ASAP", "perquery": "PerQuery-CostAware", "autosketch": "AutoSketch-Adapted"} +COLORS = {"asap": "#1f77b4", "perquery": "#ff7f0e", "autosketch": "#2ca02c"} +UNITS = {"cpu": "vCPU", "fargate": "$/hour"} +DEFAULT = ("dashboard", 1, "default") + + +def point(data): + """(templates, shared, target) of a result's workload name.""" + m = re.search(r"templates=(\w+)/shared=(\d+)/t(\w+)", data["workload"]) + return m.group(1), int(m.group(2)), m.group(3) + + +def load(directory): + return {point(d): d for d in (json.loads(p.read_text()) + for p in sorted(pathlib.Path(directory).glob("synthetic-*.json")))} + + +def pick(data, method, weights, sla): + for r in data["results"]: + if r["method"] == method and r["weights"] == weights and str(r["sla_ms"]) == str(sla): + return r if "objective" in r else None + return None + + +def weight_names(runs): + return [w["name"] for w in next(iter(runs.values()))["weights"]] + + +def fig_objective_vs_latency(runs, out): + data = runs.get(DEFAULT) + if data is None: + return + names = weight_names(runs) + fig, axes = plt.subplots(1, len(names), figsize=(5 * len(names), 3.6), squeeze=False) + for ax, w in zip(axes[0], names): + for m in METHODS: + pts = [(r["max_latency_ms"], r["objective"]) for s in data["sla_grid_ms"] + if (r := pick(data, m, w, s))] + if pts: + ax.plot(*zip(*sorted(pts)), marker="o", label=LABELS[m], color=COLORS[m]) + ax.set_xscale("log") + ax.set_yscale("log") + ax.set_xlabel("max estimated latency (ms)") + ax.set_ylabel(f"objective ({UNITS.get(w, w)})") + ax.set_title(f"dashboard, default strictness, weights {w}", fontsize=9) + axes[0][0].legend(fontsize=7) + fig.tight_layout() + fig.savefig(out / "fig_objective_vs_latency.png", dpi=150) + plt.close(fig) + + +def fig_objective_by_dimension(runs, out): + dims = [("templates", 0, ["dashboard", "all"]), ("shared", 1, [1, 8, 64]), + ("strictness", 2, ["loose", "default", "strict"])] + fig, axes = plt.subplots(1, len(dims), figsize=(4.5 * len(dims), 3.4), squeeze=False) + for ax, (name, index, values) in zip(axes[0], dims): + xs = list(range(len(values))) + for i, m in enumerate(METHODS): + ys = [] + for v in values: + key = list(DEFAULT) + key[index] = v + data = runs.get(tuple(key)) + r = data and pick(data, m, "cpu", "inf") + ys.append(r["objective"] if r else 0) + ax.bar([x + i * 0.27 for x in xs], ys, 0.27, label=LABELS[m], color=COLORS[m]) + ax.set_xticks([x + 0.27 for x in xs], [str(v) for v in values]) + ax.set_yscale("log") + ax.set_title(name, fontsize=9) + ax.set_ylabel("CPU (vCPU), no SLA") + axes[0][0].legend(fontsize=7) + fig.tight_layout() + fig.savefig(out / "fig_objective_by_dimension.png", dpi=150) + plt.close(fig) + + +def fig_planning_time(runs, out): + fig, ax = plt.subplots(figsize=(4.5, 3.4)) + for m in METHODS: + pts = [] + for r in (1, 8, 64): + data = runs.get(("dashboard", r, "default")) + res = data and pick(data, m, "cpu", "inf") + if res: + pts.append((data["rqes"], res["planning_secs"])) + if pts: + ax.plot(*zip(*pts), marker="o", label=LABELS[m], color=COLORS[m]) + ax.set_xscale("log") + ax.set_yscale("log") + ax.set_xlabel("RQEs") + ax.set_ylabel("planning time (s)") + ax.legend(fontsize=7) + fig.tight_layout() + fig.savefig(out / "fig_planning_time.png", dpi=150) + plt.close(fig) + + +def fmt(x, digits=3): + return "—" if x is None else f"{x:.{digits}g}" + + +def summary(runs, out): + lines = ["# Synthetic workload: AutoSketch vs. ASAP\n"] + for key, data in sorted(runs.items(), key=lambda kv: str(kv[0])): + lines.append(f"## templates={key[0]}, shared={key[1]}, strictness={key[2]}\n") + lines.append(f"{data['rqes']} RQEs on {data['streams']} streams; sanity violations: " + f"{len(data['sanity_violations'])}. AutoSketch probes: " + f"{data['autosketch']['probes']}, benchmark time ≥ " + f"{fmt(data['autosketch']['benchmark_secs_lower_bound_nbench1e8'])} s.\n") + lines.append("| weights | SLA (ms) | RQEs | method | objective | vs ASAP | CPU (vCPU) | " + "GiB | deployments | max latency (ms) | planning (s) |") + lines.append("|---|---|---|---|---|---|---|---|---|---|---|") + for w in weight_names(runs): + for sla in data["sla_grid_ms"]: + base = pick(data, "asap", w, sla) + for m in METHODS: + r = pick(data, m, w, sla) + if r is None: + continue + ratio = r["objective"] / base["objective"] if base and base["objective"] else None + lines.append( + f"| {w} | {sla} | {r['rqes']} | {LABELS[m]} | {fmt(r['objective'])} | " + f"{fmt(ratio)}× | {fmt(r['cpu'])} | {fmt(r['gib'])} | " + f"{r['active_deployments']} | {fmt(r['max_latency_ms'])} | " + f"{fmt(r['planning_secs'])} |") + lines.append("") + (out / "summary.md").write_text("\n".join(lines)) + + +def main(): + out = pathlib.Path(sys.argv[1]) + runs = load(out) + fig_objective_vs_latency(runs, out) + fig_objective_by_dimension(runs, out) + fig_planning_time(runs, out) + summary(runs, out) + + +if __name__ == "__main__": + main() diff --git a/scripts/run_autosketch_vs_asap.sh b/scripts/run_autosketch_vs_asap.sh new file mode 100755 index 00000000..478ef465 --- /dev/null +++ b/scripts/run_autosketch_vs_asap.sh @@ -0,0 +1,27 @@ +#!/usr/bin/env bash +# Run the AutoSketch vs. ASAP evaluation on the `traces` workloads +# (ProjectASAP/ASAPQuery#777) and plot it. +# +# Usage: scripts/run_autosketch_vs_asap.sh SATURATION_DIR +# +# `traces` reads rqe-optimizer/data/autosketch-eval/table.json for the +# workload. SATURATION_DIR holds the saturation curves (out_grid_1e7_cost/, +# out_1e9/) and the cost table (optimizer_cost/) from +# scripts/study_saturation.py, the same inputs the planner reads. + +set -euo pipefail + +SATURATION_DIR=${1:?usage: $0 SATURATION_DIR} + +OUT=rqe-optimizer/results/autosketch-vs-asap +BIN=./target/release/examples/autosketch_vs_asap +mkdir -p "$OUT" + +cargo build --release -p rqe-optimizer --example autosketch_vs_asap + +for dataset in alibaba_v2022 google_2011 boom; do + "$BIN" traces --dataset "$dataset" --saturation-dir "$SATURATION_DIR" \ + --out "$OUT/traces-$dataset.json" --runs 5 +done + +python3 scripts/plot_autosketch_vs_asap.py "$OUT" diff --git a/scripts/run_autosketch_vs_asap_synthetic.sh b/scripts/run_autosketch_vs_asap_synthetic.sh new file mode 100755 index 00000000..9fc72f39 --- /dev/null +++ b/scripts/run_autosketch_vs_asap_synthetic.sh @@ -0,0 +1,57 @@ +#!/usr/bin/env bash +# Run the synthetic workload grid of the AutoSketch vs. ASAP evaluation +# (ProjectASAP/ASAPQuery#777, section 6 "Workload grid") and plot it. +# +# Usage: +# SATURATION_DIR=DIR scripts/run_autosketch_vs_asap_synthetic.sh TABLES_DIR run +# scripts/run_autosketch_vs_asap_synthetic.sh TABLES_DIR plot +# +# TABLES_DIR holds the tables and plan.tsv of +# `export_autosketch_eval_table.py --synthetic` (the workload). SATURATION_DIR +# holds the curves and the cost table of scripts/study_saturation.py, the +# same inputs the planner reads. `run` evaluates every plan.tsv +# row, JOBS at a time, at every weight setting and SLA of #777; a run over +# JOB_TIMEOUT seconds is stopped and listed in failed.txt. RUNS timing runs +# each. The machine is recorded in machine-.txt. + +set -euo pipefail + +TABLES=${1:?usage: run_autosketch_vs_asap_synthetic.sh TABLES_DIR (run|plot)} +MODE=${2:?usage: run_autosketch_vs_asap_synthetic.sh TABLES_DIR (run|plot)} +OUT=${OUT:-rqe-optimizer/results/autosketch-vs-asap-synthetic} +JOBS=${JOBS:-4} +RUNS=${RUNS:-3} +JOB_TIMEOUT=${JOB_TIMEOUT:-14400} +BIN=${CARGO_TARGET_DIR:-./target}/release/examples/autosketch_vs_asap +mkdir -p "$OUT" + +case "$MODE" in +run) + : "${SATURATION_DIR:?set SATURATION_DIR to the saturation study output directory}" + export SATURATION_DIR JOB_TIMEOUT RUNS OUT + cargo build --release -p rqe-optimizer --example autosketch_vs_asap + { + echo "host: $(hostname)" + echo "cpu: $(lscpu | sed -n 's/^Model name: *//p')" + echo "nproc: $(nproc)" + echo "started: $(date -u +%FT%TZ)" + echo "jobs: $JOBS" + } > "$OUT/machine-$(hostname -s).txt" + : > "$OUT/failed.txt" + while IFS=$'\t' read -r table target result; do + [[ -s "$OUT/$result" ]] || echo "$TABLES/$table $target $OUT/$result" + done < "$TABLES/plan.tsv" | + xargs -P "$JOBS" -L 1 sh -c \ + 'timeout "$JOB_TIMEOUT" "$0" synthetic --table "$1" --target "$2" --runs "$RUNS" \ + --saturation-dir "$SATURATION_DIR" --out "$3" 2>> "$3.log" \ + || echo "$3" >> "$OUT/failed.txt"' \ + "$BIN" + ;; +plot) + python3 scripts/plot_autosketch_vs_asap_synthetic.py "$OUT" + ;; +*) + echo "unknown mode $MODE" >&2 + exit 2 + ;; +esac diff --git a/scripts/test_export_autosketch_eval_table.py b/scripts/test_export_autosketch_eval_table.py index 983d797f..2fd582be 100644 --- a/scripts/test_export_autosketch_eval_table.py +++ b/scripts/test_export_autosketch_eval_table.py @@ -13,7 +13,10 @@ sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) -from export_autosketch_eval_table import boom_query, main # noqa: E402 +from study_saturation import COST_KEYS, COST_PARETO_ALPHA, COST_THETA # noqa: E402 +from export_autosketch_eval_table import ( # noqa: E402 + SYNTHETIC_DEFAULT, SYNTHETIC_GRID, boom_query, main, queries_per_instance, shared_queries, + synthetic_plan, synthetic_queries) from study_saturation import CURVE_COLUMNS, SUMMARY_COLUMNS # noqa: E402 SUMMARY_HEADER = [ @@ -87,22 +90,98 @@ def test_inexact_merge_reads_the_merged_curve_or_flags_its_absence(self): self.assertEqual(kll["asap_error_by_m"]["16"], kll["autosketch_error"]) self.assertIn("single-sketch curve optimistic", kll["asap_flag_by_m"]["16"]) - def test_saturated_compares_max_N_with_n_sat(self): - configs, _ = self.export() - self.assertTrue(configs["freq"]["saturated"]) - self.assertFalse(configs["p99"]["saturated"]) - - def test_autosketch_reads_the_curve_even_when_unsaturated(self): - configs, _ = self.export() - # max_N 5000 lies between 1000 (0.2) and 10000 (0.1), log-linear: 0.130. - self.assertAlmostEqual(configs["p99"]["autosketch_error"], 0.2 - 0.1 * 0.69897, 4) - - def test_rqe_carries_window_and_label_set(self): + def test_a_key_query_reads_every_key(self): _, rqes = self.export() - r = rqes["freq"] - self.assertEqual((r["lookback_secs"], r["interval_secs"]), (3600, 60)) - self.assertEqual(r["label_set"]["groups"], 1) - self.assertAlmostEqual(r["label_set"]["arrival_rate_per_sec"], 5000 / 3600) + # A key query reads every key (worst_K); a value query issues one. + self.assertEqual(rqes["freq"]["queries_per_instance"], 1000) + self.assertEqual(rqes["p99"]["queries_per_instance"], 1) + + def test_synthetic_mode_writes_one_table_per_grid_point_and_a_plan(self): + out = self.path("synthetic") + main(["--synthetic", "--revision", "test", "--out", out]) + with open(os.path.join(out, "plan.tsv")) as f: + plan = [line.rstrip("\n").split("\t") for line in f] + self.assertEqual(len(plan), len(synthetic_plan())) + self.assertEqual(len(os.listdir(out)), len({row[0] for row in plan}) + 1) + default = "synthetic-templatesdashboard-shared1.json" + self.assertIn([default, "default", default[:-5] + "-tdefault.json"], plan) + with open(os.path.join(out, "synthetic-templatesall-shared1.json")) as f: + rqes = {r["query_id"] + "/" + r["range"]: r + for r in json.load(f)["workloads"][0]["rqes"]} + # The tables hold the workload only: sums name the exact accumulator + # with no data shape; sketches carry the cost table's shape. + family = rqes["sum_by_job/1s"]["families"][0] + self.assertEqual((family["sketch"], family["target"], family["grid_param"]), + ("exact-sum", 0.0, None)) + self.assertNotIn("configs", family) + self.assertEqual(rqes["sum_by_job/1s"]["label_set"]["groups"], 10) + top = rqes["topk32_sum_by_label0_rate/15m"]["families"][0] + self.assertEqual((top["grid_param"], top["grid_K"]), (COST_THETA, COST_KEYS)) + kll, dd = rqes["quantile_over_time_p0.5/15m"]["families"] + self.assertEqual((kll["sketch"], kll["grid_param"], kll["grid_K"]), + ("kll-percall", COST_PARETO_ALPHA, None)) + self.assertEqual(dd["sketch"], "dd") + + +class SyntheticQueriesTest(unittest.TestCase): + def test_ten_templates_give_50_distinct_rqes(self): + qs = synthetic_queries(templates="all") + # 1 + 5 spatial; per window 1 + 1 + 5 + 1 + 1 + 1 + 1 temporal, and + # template 10's operands repeat template 5's. + self.assertEqual(len(qs), 6 + 11 * 4) + streams = {(q["stream"], q["capability"], q["groups"]) for q in qs} + self.assertEqual(streams, { + ("job/value", "sum", 10), ("job/increment", "rate", 10), + ("job/dist", "quantile", 10), ("series/value", "sum", 10_000), + ("series/increment", "rate", 10_000), ("series/dist", "quantile", 10_000), + ("label_0/value", "topk", 1), ("label_0/increment", "topk", 1)}) + + def test_items_per_instance_follow_the_data_model(self): + qs = {(q["query_id"], q["range"]): q for q in synthetic_queries(templates="all")} + # 1e4 series x 200 samples/s = 2e6 samples/s. + self.assertEqual(qs[("quantile_by_job_p0.5", "1s")]["max_N"], 2e5) + self.assertEqual(qs[("quantile_over_time_p0.5", "15m")]["max_N"], 200 * 900) + top = qs[("topk32_sum_by_label0_rate", "24h")] + self.assertEqual((top["K"], top["groups"], top["max_N"]), (10_000, 1, 2e6 * 86400)) + + def test_every_template_issues_one_query_per_instance(self): + self.assertEqual({queries_per_instance(q) for q in synthetic_queries(templates="all")}, + {1}) + + +class SyntheticPlanTest(unittest.TestCase): + def test_plan_is_the_default_and_each_dimension_alone(self): + plan = synthetic_plan() + self.assertEqual(plan[0], SYNTHETIC_DEFAULT) + for dim, values in SYNTHETIC_GRID.items(): + for v in values: + self.assertIn({**SYNTHETIC_DEFAULT, dim: v}, plan) + self.assertEqual(len(plan), 1 + sum(len(v) - 1 for v in SYNTHETIC_GRID.values())) + + +class DashboardAndSharedTest(unittest.TestCase): + def test_dashboard_has_21_distinct_rqes(self): + qs = synthetic_queries(templates="dashboard") + count = {} + for q in qs: + name = q["query_id"].split("_p")[0] + count[name] = count.get(name, 0) + 1 + # D1 3 q x 4 windows, D2 3, D3 4, D4 2; D5 repeats D1. + self.assertEqual(count, {"quantile_over_time": 12, "quantile_by_job": 3, + "sum_by_job_rate": 4, "topk32_sum_by_label0_rate": 2}) + self.assertTrue(all(q["step_s"] == 60 for q in qs if q["range"] != "1s")) + + def test_shared_replicas_read_one_stream_and_dedupe(self): + point = {**SYNTHETIC_DEFAULT, "shared": 4} + qs = shared_queries(point) + self.assertEqual(qs, shared_queries(point)) # seeded + ids = [(q["query_id"], q["range"]) for q in qs] + self.assertEqual(len(ids), len(set(ids))) + self.assertTrue(all(q["dataset"].endswith("shared=4") for q in qs)) + self.assertTrue(all("@" in q["range"] for q in qs if q["range"] != "1s")) + sizes = [len(shared_queries({**SYNTHETIC_DEFAULT, "shared": r})) for r in (1, 8, 64)] + self.assertLess(sizes[0], sizes[1]) + self.assertLess(sizes[2], 8 * sizes[1]) class BoomTest(unittest.TestCase):