From 8b59466c087d4f7a77bbe76bd8ea4363a2aee323 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 6 Oct 2026 23:39:33 +0000 Subject: [PATCH 1/3] rqe-optimizer: evaluate AutoSketch vs ASAP on the trace workloads Runner for ASAPQuery#777's comparison on main's planner: one metric per stream of the evaluation table, ASAP by milp::minimize, PerQuery by minimize per RQE, AutoSketch by autosketch::plan, all scored by analytical_cost_model::score (#145) at #777's weights (CPU only; Fargate $/vCPU-h and $/GB-h) and SLA grid {0.01, 0.1, 1, 10} ms and none. Each RQE's accuracy rides in the cost entries: acc@ (with merged values in merge_accuracy, #154) for ASAP, as@ for AutoSketch. Co-Authored-By: Claude Opus 5.5 --- rqe-optimizer/examples/autosketch_vs_asap.rs | 823 +++++++++++++++++++ scripts/plot_autosketch_vs_asap.py | 134 +++ scripts/run_autosketch_vs_asap.sh | 21 + 3 files changed, 978 insertions(+) create mode 100644 rqe-optimizer/examples/autosketch_vs_asap.rs create mode 100644 scripts/plot_autosketch_vs_asap.py create mode 100755 scripts/run_autosketch_vs_asap.sh diff --git a/rqe-optimizer/examples/autosketch_vs_asap.rs b/rqe-optimizer/examples/autosketch_vs_asap.rs new file mode 100644 index 00000000..8712f691 --- /dev/null +++ b/rqe-optimizer/examples/autosketch_vs_asap.rs @@ -0,0 +1,823 @@ +//! 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 --out FILE [--runs 5] +//! autosketch_vs_asap synthetic --table FILE --target default --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`. Each stream +//! of a table (`traces`: dataset, query and kind; `synthetic`: the table's +//! `stream`) is one metric, which is where sharing comes from. Each RQE's +//! accuracy is carried in its stream's cost entries: `acc@` for ASAP, +//! with the value after merging `m` shards in `merge_accuracy[m]`, and +//! `as@` (one sketch at the query window's N, no saturation +//! requirement) for AutoSketch. + +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, Objective}; +use rqe_optimizer::{ + validate_facts, AccuracyDirection, AtomicCostEntry, 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, and the cost entries each +/// stream may use. `flags` maps `(stream, sketch, config)` to the table's +/// flags for the chosen-config report. +struct Workload { + name: String, + raqes: Vec, + facts: WorkloadFacts, + costs: BTreeMap>, + flags: BTreeMap<(String, String, String), Value>, + /// The benchmark's whole query phase (seconds) per (sketch, config), for + /// the AutoSketch benchmark-time lower bound. + query_phase_secs: BTreeMap<(String, String), f64>, + notes: Vec, +} + +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 workload = match args.get(1).map(String::as_str) { + Some("traces") => from_table( + TRACES_TABLE, + Some(&arg(&args, "--dataset").expect("--dataset")), + None, + ), + Some("synthetic") => from_table( + &arg(&args, "--table").expect("--table"), + None, + Some(arg(&args, "--target").expect("--target")), + ), + _ => 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::TopK => [0.90, 0.95, 0.99], + Capability::Cardinality => [0.05, 0.02, 0.01], + Capability::SumOrCount | 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::SumOrCount, + (Some("rate"), _) => Capability::RateOrIncrease, + (Some("topk"), _) => Capability::TopK, + (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) -> Workload { + let table: Value = serde_json::from_str(&std::fs::read_to_string(path).unwrap()).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(); + // (stream, sketch, config) -> entry under construction. + let mut entries: BTreeMap<(String, String, String), AtomicCostEntry> = BTreeMap::new(); + let mut flags: BTreeMap<(String, String, String), Value> = BTreeMap::new(); + let mut query_phase_secs: BTreeMap<(String, String), f64> = 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(); + // 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, + }); + 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 mut rqe_target = None; + let mut direction = AccuracyDirection::LowerIsBetter; + for family in r["families"].as_array().unwrap() { + let sketch = family["sketch"].as_str().unwrap().to_string(); + if !capability.families().contains(&sketch.as_str()) { + // e.g. `traces`' frequency sketches, or top-k for a sum query. + skipped_families += 1; + continue; + } + let higher = family["higher_is_better"].as_bool().unwrap(); + direction = if higher { + AccuracyDirection::HigherIsBetter + } else { + AccuracyDirection::LowerIsBetter + }; + let t = match target.as_deref() { + Some(level) => strictness_target(level, capability), + None => family["target"].as_f64().unwrap(), + }; + assert!(rqe_target.is_none_or(|old| old == t), "one target per RQE"); + rqe_target = Some(t); + for c in family["configs"].as_array().unwrap() { + let config = c["config"].as_str().unwrap().to_string(); + let key = (stream.clone(), sketch.clone(), config.clone()); + let num = |field: &str| c[field].as_f64(); + let (Some(mem), Some(insert_ns), Some(merge_us), Some(phase_us), Some(query_us)) = ( + num("memory_bytes"), + num("insert_ns_per_item"), + num("merge_us_per_fold"), + num("query_phase_us"), + num("query_us_per_query").or(num("query_phase_us")), + ) else { + continue; // no measured cost: not a candidate + }; + let phase = query_phase_secs + .entry((sketch.clone(), c["params"].to_string())) + .or_default(); + *phase = phase.max(phase_us * 1e-6); + let entry = entries.entry(key.clone()).or_insert(AtomicCostEntry { + sketch: sketch.clone(), + sketch_config: json!({"algorithm": sketch, "params": c["params"]}), + mem_bytes_per_instance: 0.0, + insert_cpu_secs: 0.0, + merge_cpu_secs: 0.0, + query_cpu_secs: 0.0, + query_accuracy: BTreeMap::new(), + merge_accuracy: BTreeMap::new(), + measured_at: None, + }); + // A stream's RQEs may sit at different grid points; keep the + // most expensive measurement (conservative). + entry.mem_bytes_per_instance = entry.mem_bytes_per_instance.max(mem); + entry.insert_cpu_secs = entry.insert_cpu_secs.max(insert_ns * 1e-9); + entry.merge_cpu_secs = entry.merge_cpu_secs.max(merge_us * 1e-6); + // One evaluation's query CPU per instance: q_r queries. + entry.query_cpu_secs = entry.query_cpu_secs.max(queries * query_us * 1e-6); + if let Some(v) = c["autosketch_error"].as_f64() { + entry.query_accuracy.insert(format!("as@{id}"), v); + } + for (m, v) in c["asap_error_by_m"].as_object().unwrap() { + let Some(v) = v.as_f64() else { continue }; + let m: u64 = m.parse().unwrap(); + if m == 1 { + entry.query_accuracy.insert(format!("acc@{id}"), v); + } else { + entry + .merge_accuracy + .entry(m) + .or_default() + .insert(format!("acc@{id}"), v); + } + } + flags + .entry(key) + .or_insert_with(|| json!({})) + .as_object_mut() + .unwrap() + .insert( + id.clone(), + json!({ + "saturated": c["saturated"], + "not_saturated_in_grid": c["not_saturated_in_grid"], + "autosketch_flag": c["autosketch_flag"], + "asap_flag_by_m": c["asap_flag_by_m"], + "cost_flag": c["cost_flag"], + }), + ); + } + } + let grouping = if groups > 1 { + label_set(&["g"]) + } else { + LabelSet::new() + }; + 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, + accuracy_metric: format!("acc@{id}"), + // An RQE with no family of its capability is unservable; its + // target is never read. + accuracy_sla: rqe_target.unwrap_or(0.0), + accuracy_direction: direction, + latency_sla_ms: None, + }); + } + let mut costs: BTreeMap> = BTreeMap::new(); + for ((stream, _, _), entry) in entries { + costs.entry(stream).or_default().push(entry); + } + let name = match &target { + Some(t) => format!("{dataset}/t{t}"), + None => format!("traces/{dataset}"), + }; + Workload { + name, + raqes, + facts, + costs, + flags, + query_phase_secs, + 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:?}"), + "per-config costs of a stream = max over its RQEs' grid points".into(), + "query_cpu_secs = q_r queries per instance per evaluation".into(), + ], + } +} + +// --------------------------------------------------------------- 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) + }) + .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| c.query_accuracy.get(&format!("as@{}", r.id)).copied(); + 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]; + let flags = w.flags.iter().find_map(|((stream, sketch, c), f)| { + (*stream == d.metric + && *sketch == d.config.sketch + && params_match(c, &d.config.sketch_config["params"])) + .then(|| f[&r.id].clone()) + }); + 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, + "flags": flags, + }) + }) + .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, + }) +} + +/// The table's config strings ("rows=3 cols=256") against a params object. +fn params_match(config: &str, params: &Value) -> bool { + config.split_whitespace().all(|kv| { + let (k, v) = kv.split_once('=').unwrap(); + params[k].as_f64() == v.parse::().ok() + }) +} + +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 { + // 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).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) + .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]; + let key = (c.sketch.clone(), c.sketch_config["params"].to_string()); + let phase = w.query_phase_secs.get(&key).copied().unwrap_or(0.0); + (format!("{}/{}", key.0, key.1), c.insert_cpu_secs, phase) + }) + }) + .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); + solve_secs.push(started.elapsed().as_secs_f64()); + solution = Some(solved); + } + let asap = match solution.unwrap() { + Ok(s) => { + let mut v = summarize( + w, + &sla_raqes, + &asap_candidates, + &s.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) + .map(|solved| singles[i][solved.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)) + .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::SumOrCount), 0.0); + assert_eq!(strictness_target(level, Capability::RateOrIncrease), 0.0); + } + assert_eq!(strictness_target("strict", Capability::Quantile), 0.005); + } +} 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/run_autosketch_vs_asap.sh b/scripts/run_autosketch_vs_asap.sh new file mode 100755 index 00000000..44b244e2 --- /dev/null +++ b/scripts/run_autosketch_vs_asap.sh @@ -0,0 +1,21 @@ +#!/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 +# +# `traces` reads rqe-optimizer/data/autosketch-eval/table.json. + +set -euo pipefail + +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" --out "$OUT/traces-$dataset.json" --runs 5 +done + +python3 scripts/plot_autosketch_vs_asap.py "$OUT" From 3bd0b7d103bb8bbd52d6d4e7a470785ad61af308 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 6 Oct 2026 23:43:04 +0000 Subject: [PATCH 2/3] Evaluate the synthetic PromQL workload of ASAPQuery#777 export_autosketch_eval_table.py --synthetic writes #777's synthetic workload: fixed data (1e4 series = label_0 keys, 10 jobs, 200 samples/s per series, Zipf theta 1.1), windows {15m, 1h, 6h, 24h}, the 10 templates (50 distinct RQEs) and the dashboard set (21), top-k at k = 32. Sum and rate are served by exact accumulators priced from --exact-costs (the cost table's exact-sum / exact-increase rows), with zero error at every merge count. The grid is the default point plus each dimension alone: template set, shared replicas {1, 8, 64}, strictness. Quantile curves stay on Pareto a = 2 until the targeted Zipf-rank saturation run. run_autosketch_vs_asap_synthetic.sh runs plan.tsv; the plot script writes the objective-vs-latency, per-dimension and planning-time figures and a summary. Co-Authored-By: Claude Opus 5.5 --- scripts/export_autosketch_eval_table.py | 354 +++++++++++++++++-- scripts/plot_autosketch_vs_asap_synthetic.py | 167 +++++++++ scripts/run_autosketch_vs_asap_synthetic.sh | 52 +++ scripts/test_export_autosketch_eval_table.py | 115 +++++- 4 files changed, 664 insertions(+), 24 deletions(-) create mode 100644 scripts/plot_autosketch_vs_asap_synthetic.py create mode 100755 scripts/run_autosketch_vs_asap_synthetic.sh diff --git a/scripts/export_autosketch_eval_table.py b/scripts/export_autosketch_eval_table.py index 08092466..497b9724 100644 --- a/scripts/export_autosketch_eval_table.py +++ b/scripts/export_autosketch_eval_table.py @@ -24,6 +24,11 @@ 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. Sum and rate queries are +served by exact accumulators whose costs come from --exact-costs. """ import argparse @@ -32,6 +37,7 @@ import json import math import os +import random import re import subprocess import sys @@ -41,7 +47,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 load_curves, point_key, query_secs # noqa: E402 SHARD_COUNTS = [1, 4, 16, 64] CAPABILITY = {"cms": "freq", "countsketch": "freq", "topk": "topk", @@ -65,7 +71,40 @@ def saturation_of(srow, n): return n_sat, n_sat is not None and n >= n_sat -def config_row(q, family, key, cost_key, curves, curves_big, sat, sat_big, merge): +def record_key(record): + """A cost record's point key: its "key" field, or for records written + without one (#130's cost phase), its sketch, params and distribution.""" + if "key" in record: + return point_key(*record["key"]) + dist = record["workload"]["synthetic"]["description"]["column_spec"][0]["distribution"] + params = record["sketch_config"]["params"] + config = " ".join(f"{k}={v:g}" if isinstance(v, float) else f"{k}={v}" + for k, v in sorted(params.items(), key=lambda kv: CONFIG_ORDER.index(kv[0]))) + if dist["kind"] == "pareto": + return point_key(record["sketch"], config, "pareto", dist["alpha"], "") + return point_key(record["sketch"], config, dist["kind"], dist["skewness"], + dist["population_size"]) + + +# Parameter order in saturation.csv's config strings ("rows=3 cols=1024"). +CONFIG_ORDER = ["rows", "cols", "k", "alpha", "lg_k"] + + +def per_query_us(per_query, cost_key): + """CPU per query (µs): the cost record's query phase over its query count + (study_saturation.query_secs). A rows=5 config without a record takes + 5/3 of the rows=3 one, as cost_of does.""" + if cost_key in per_query: + return per_query[cost_key] * 1e6 + sketch, config = cost_key[0], cost_key[1] + if config.startswith("rows=5 "): + rows3 = (sketch, config.replace("rows=5 ", "rows=3 "), *cost_key[2:]) + if rows3 in per_query: + return per_query[rows3] * 1e6 * 5.0 / 3.0 + return None + + +def config_row(q, family, key, cost_key, curves, curves_big, sat, sat_big, merge, per_query): """One config's lookups for query-range q, or None without a curve.""" curve = curves.get(key) if not curve: @@ -95,6 +134,7 @@ def config_row(q, family, key, cost_key, curves, curves_big, sat, sat_big, merge "config": key[1], "params": params_of(key[1]), "memory_bytes": memory, "insert_ns_per_item": insert_ns, "merge_us_per_fold": merge_us, "query_phase_us": query_us, + "query_us_per_query": per_query_us(per_query, cost_key), "cost_flag": cost_flag, "n_sat": n_sat, "not_saturated_in_grid": bool(srow) and srow["n_sat"] == "not_saturated", "saturated": saturated, @@ -146,13 +186,193 @@ def queries(path): return out -def rqe_entry(q, curves, curves_big, sat, sat_big, merge): - """The table entry for one skew_summary query-range.""" +# 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. +SYNTHETIC_SERIES = 10_000 +SYNTHETIC_JOBS = 10 +SYNTHETIC_SAMPLES_PER_SEC = 200 +SYNTHETIC_THETA = 1.1 +# Quantile curves are measured on Pareto data; #777 measures quantiles on the +# Zipf ranks, which needs its own saturation run. Until then, Pareto a = 2. +SYNTHETIC_ALPHA = 2.0 +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": ["quantile curves on Pareto a = 2, not the Zipf ranks of #777"] + if capability == "quantile" else [], + **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 exact_family(q, family, exact_costs): + """An exact accumulator's family for q: one config, zero error at every m, + costs per group from the cost table (`--exact-costs`).""" + sketch = EXACT_SKETCH[family] + row = exact_costs.get(sketch) + if row is None: + return None + zero = {str(m): 0.0 for m in SHARD_COUNTS} + config = { + "config": "", "params": {}, + "memory_bytes": row["mem_bytes_per_instance"], + "insert_ns_per_item": row["insert_cpu_secs"] * 1e9, + "merge_us_per_fold": row["merge_cpu_secs"] * 1e6, + "query_phase_us": row["query_cpu_secs"] * 1e6, + "query_us_per_query": row["query_cpu_secs"] * 1e6, + "cost_flag": "", "n_sat": None, "not_saturated_in_grid": False, "saturated": True, + "autosketch_error": 0.0, "autosketch_flag": "", + "asap_error_by_m": zero, "asap_flag_by_m": {m: "" for m in zero}, + } + return { + "family": family, "capability": q["capability"], "sketch": sketch, + "error_metric": "relative_error", "higher_is_better": False, "target": 0.0, + "grid_param": None, "grid_K": None, "flags": [], "configs": [config], + } + + +def rqe_entry(q, curves, curves_big, sat, sat_big, merge, per_query, exact_costs=None): + """The table entry for one skew_summary (or `synthetic`) query-range.""" instant = not q["range_s"] # An instant query reads one step's samples (max_N counts one step window). lookback = q["step_s"] if instant else q["range_s"] families = [] - for family in KIND_FAMILIES[q["kind"]]: + for family in q.get("families", KIND_FAMILIES[q["kind"]]): + if family in EXACT_SKETCH: + exact = exact_family(q, family, exact_costs or {}) + if exact: + families.append(exact) + continue sketch, target_col, higher_better = FAMILIES[family] points = [k for k in curves if k[0] == sketch] if not points: @@ -162,7 +382,8 @@ def rqe_entry(q, curves, curves_big, sat, sat_big, merge): for config in sorted({k[1] for k in points}, key=nominal_size): key = (sketch, config, points[0][2], float(param), card) cost_key = (sketch, config, key[2], float(cost_param), card) - row = config_row(q, family, key, cost_key, curves, curves_big, sat, sat_big, merge) + row = config_row(q, family, key, cost_key, curves, curves_big, sat, sat_big, merge, + per_query) if row: rows.append(row) if not rows: @@ -174,7 +395,7 @@ def rqe_entry(q, curves, curves_big, sat, sat_big, merge): "target": q[target_col], "grid_param": param, "grid_K": card, "flags": [x for x in common if x], "configs": rows, }) - return { + entry = { "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": instant, @@ -182,6 +403,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"]}, @@ -189,6 +413,19 @@ def rqe_entry(q, curves, curves_big, sat, sat_big, merge): ["instant: lookback = one step (max_N counts one step window)"] if instant else []), "families": families, } + if "stream" in q: + entry.update(stream=q["stream"], capability=q["capability"], + label_set={"groups": q["groups"], "arrival_rate_per_sec": q["arrival_rate"], + "keys_per_window": q["K"]}) + return entry + + +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): @@ -199,25 +436,75 @@ 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, merge curves and CPU per query (from the raw + cost records); later files win per point.""" + curves, sat, merge, per_query = {}, {}, {}, {} + 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) + for path in args.cost_records or []: + with open(path) as f: + for line in f: + if line.strip(): + record = json.loads(line) + per_query[record_key(record)] = query_secs(record) + curves_big = load_curves(args.curves_big) if args.curves_big else {} + return curves, curves_big, sat, load_saturation(args.saturation_big), merge, per_query + + +def load_exact_costs(path): + """{sketch: row} for the exact accumulators of a cost table + (`export_rqe_optimizer_costs.sh`'s rqe_atomic_costs.json).""" + if not path: + return {} + with open(path) as f: + return {r["sketch"]: r for r in json.load(f) if r["sketch"] in EXACT_SKETCH.values()} + + +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.""" + curves, curves_big, sat, sat_big, merge, per_query = load_inputs(args) + exact_costs = load_exact_costs(args.exact_costs) + assert exact_costs, "--exact-costs must hold exact-sum and exact-increase rows" + 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))) + rqes = [rqe_entry(q, curves, curves_big, sat, sat_big, merge, per_query, exact_costs) + for q in qs] + out[name] = table(args, {qs[0]["dataset"]: rqes}) + 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, per_query = load_inputs(args) workloads = {} for q in queries(args.summary): if not q["step_s"]: continue - entry = rqe_entry(q, curves, curves_big, sat, sat_big, merge) + entry = rqe_entry(q, curves, curves_big, sat, sat_big, merge, per_query) 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, - args.saturation_big, *(args.merge_curves or [])] if p] + return table(args, workloads) + + +def table(args, workloads): + inputs = [p for p in [args.summary, *args.curves, args.curves_big, *args.saturation, + args.saturation_big, *(args.merge_curves or []), + *(args.cost_records or [])] if p] return { "schema_version": 1, "plan": "ProjectASAP/ASAPQuery#777", @@ -232,24 +519,47 @@ 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", required=True, nargs="+") parser.add_argument("--curves-big") - parser.add_argument("--saturation", required=True) + parser.add_argument("--saturation", required=True, nargs="+") parser.add_argument("--saturation-big") parser.add_argument("--merge-curves", nargs="*", help="saturation_merge_curve.csv files") + parser.add_argument("--exact-costs", + help="rqe_atomic_costs.json with the exact-sum and exact-increase rows " + "(--synthetic)") + parser.add_argument("--cost-records", nargs="*", + help="raw saturation_cost.jsonl files, for CPU per query") parser.add_argument("--revision", help="sketch-bench commit of the curves (default: HEAD)") parser.add_argument("--out", required=True) args = parser.parse_args(argv) 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: + parser.error("--summary is 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_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_synthetic.sh b/scripts/run_autosketch_vs_asap_synthetic.sh new file mode 100755 index 00000000..51d8e0c1 --- /dev/null +++ b/scripts/run_autosketch_vs_asap_synthetic.sh @@ -0,0 +1,52 @@ +#!/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: +# 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`. `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) + 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"'" \ + --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..6e090194 100644 --- a/scripts/test_export_autosketch_eval_table.py +++ b/scripts/test_export_autosketch_eval_table.py @@ -13,7 +13,9 @@ sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) -from export_autosketch_eval_table import boom_query, main # 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 = [ @@ -67,9 +69,19 @@ def tearDown(self): def export(self): out = self.path("table.json") + # A cost record for the CMS point: 2 runs averaging 6 ms of query CPU + # (4 user + 2 sys), over 0.5 s of wall time at 4000 queries/s, i.e. + # 2000 queries: 3 µs of CPU per query. + with open(self.path("cost.jsonl"), "w") as f: + f.write(json.dumps({ + "key": [CMS, "rows=3 cols=256", "zipf", 1.0, "1000"], + "query_cpu_time_ms": {"user_ms": {"mean": 4.0}, "sys_ms": {"mean": 2.0}}, + "query_wall_time_ms": {"mean": 500.0}, + "query_throughput_items_per_sec": {"mean": 4000.0}, + }) + "\n") main(["--summary", self.path("summary.csv"), "--curves", self.path("curve.csv"), "--saturation", self.path("sat.csv"), "--merge-curves", self.path("merge.csv"), - "--revision", "test", "--out", out]) + "--cost-records", self.path("cost.jsonl"), "--revision", "test", "--out", out]) with open(out) as f: table = json.load(f) rqes = {r["query_id"]: r for w in table["workloads"] for r in w["rqes"]} @@ -87,6 +99,16 @@ 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_per_query_cpu_is_the_query_phase_over_its_query_count(self): + configs, rqes = self.export() + self.assertAlmostEqual(configs["freq"]["query_us_per_query"], 3.0) + self.assertEqual(configs["freq"]["query_phase_us"], 2000.0) + # No cost record for the KLL point: no per-query CPU. + self.assertIsNone(configs["p99"]["query_us_per_query"]) + # 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_saturated_compares_max_N_with_n_sat(self): configs, _ = self.export() self.assertTrue(configs["freq"]["saturated"]) @@ -105,6 +127,95 @@ def test_rqe_carries_window_and_label_set(self): self.assertAlmostEqual(r["label_set"]["arrival_rate_per_sec"], 5000 / 3600) + def test_synthetic_mode_writes_one_table_per_grid_point_and_a_plan(self): + out = self.path("synthetic") + with open(self.path("exact.json"), "w") as f: + json.dump([{"sketch": s, "sketch_config": None, "mem_bytes_per_instance": 128.0, + "insert_cpu_secs": 1e-7, "merge_cpu_secs": 2e-7, "query_cpu_secs": 4e-7, + "query_accuracy": {}} for s in ("exact-sum", "exact-increase")], f) + main(["--synthetic", "--curves", self.path("curve.csv"), "--saturation", + self.path("sat.csv"), "--exact-costs", self.path("exact.json"), + "--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"]} + # Sums are served by the exact accumulator, priced from --exact-costs, + # with zero error at every merge count. + family = rqes["sum_by_job/1s"]["families"][0] + self.assertEqual((family["sketch"], family["target"]), ("exact-sum", 0.0)) + config = family["configs"][0] + self.assertEqual(config["memory_bytes"], 128.0) + self.assertEqual(set(config["asap_error_by_m"].values()), {0.0}) + self.assertEqual(rqes["sum_by_job/1s"]["label_set"]["groups"], 10) + + +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): def test_chunk_defines_n_and_window(self): row = dict.fromkeys(SUMMARY_HEADER, "") From a6ed461d7587eb7c4b96d4e7f06071a2d98dd9d4 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 6 Oct 2026 23:43:21 +0000 Subject: [PATCH 3/3] Synthetic workload results: reproduction steps and ASAP's profiling time The two cost models are gone (#145's weighted objective replaces them, in the runner of the PRs below). This keeps ASAP's one-time profiling time (#777 section 7) and documents how the reduced grid is run; results are committed after the full run. Co-Authored-By: Claude Opus 5.5 --- rqe-optimizer/data/profiling-time.json | 78 +++++++++++++++++++ .../autosketch-vs-asap-synthetic/README.md | 33 ++++++++ scripts/profiling_time.py | 75 ++++++++++++++++++ 3 files changed, 186 insertions(+) create mode 100644 rqe-optimizer/data/profiling-time.json create mode 100644 rqe-optimizer/results/autosketch-vs-asap-synthetic/README.md create mode 100644 scripts/profiling_time.py diff --git a/rqe-optimizer/data/profiling-time.json b/rqe-optimizer/data/profiling-time.json new file mode 100644 index 00000000..480425f9 --- /dev/null +++ b/rqe-optimizer/data/profiling-time.json @@ -0,0 +1,78 @@ +{ + "method": "first-to-last record timestamp per run (see scripts/profiling_time.py)", + "runs": { + "#130 grid accuracy": { + "file": "out_grid_1e7/saturation_accuracy.jsonl", + "records": 30384, + "first": "2026-09-29T20:40:43Z", + "last": "2026-09-29T21:18:55Z", + "wall_secs": 2292.0 + }, + "#130 grid cost": { + "file": "out_grid_1e7_cost/saturation_cost.jsonl", + "records": 343, + "first": "2026-09-30T16:49:46Z", + "last": "2026-09-30T19:37:05Z", + "wall_secs": 10039.0 + }, + "#130 1e9 subset": { + "file": "out_1e9/saturation_accuracy.jsonl", + "records": 13875, + "first": "2026-09-29T21:24:47Z", + "last": "2026-09-30T15:50:56Z", + "wall_secs": 66369.0 + }, + "#131 merge (first config)": { + "file": "out_merge/saturation_accuracy.jsonl", + "records": 8160, + "first": "2026-09-29T21:40:14Z", + "last": "2026-09-29T21:52:21Z", + "wall_secs": 727.0 + }, + "#131 merge top-k grid": { + "file": "out_merge_topk/saturation_accuracy.jsonl", + "records": 34272, + "first": "2026-10-02T02:33:13Z", + "last": "2026-10-02T03:04:25Z", + "wall_secs": 1872.0 + }, + "#131 merge quantile grid": { + "file": "out_merge_quantile/saturation_accuracy.jsonl", + "records": 19040, + "first": "2026-10-02T03:04:26Z", + "last": "2026-10-02T03:06:14Z", + "wall_secs": 108.0 + }, + "#140 accuracy grid": { + "file": "grid/saturation_accuracy.jsonl", + "records": 30240, + "first": "2026-10-05T03:24:53Z", + "last": "2026-10-05T05:45:15Z", + "wall_secs": 8422.0 + }, + "#140 merge top-k": { + "file": "merge_topk/saturation_accuracy.jsonl", + "records": 32640, + "first": "2026-10-05T03:24:53Z", + "last": "2026-10-05T05:11:07Z", + "wall_secs": 6374.0 + }, + "#140 cost (178 points, N 1e8, serial)": { + "file": "grid/saturation_cost.serial-178.jsonl", + "records": 178, + "first": "2026-10-05T05:45:53Z", + "last": "2026-10-05T12:33:48Z", + "wall_secs": 24475.0 + }, + "#140 cost (62 top-k points, N 1e7, serial)": { + "file": "grid/saturation_cost.serial-1e7-62.jsonl", + "records": 62, + "first": "2026-10-05T20:40:49Z", + "last": "2026-10-05T22:45:37Z", + "wall_secs": 7488.0 + } + }, + "total_wall_secs": 128166.0, + "missing": [], + "note": "Wall spans on the machines that produced each run, with that run's own --jobs. #140's discarded 12-way parallel cost records and its stopped serial N=1e8 rerun are not counted: they did not produce the tables." +} diff --git a/rqe-optimizer/results/autosketch-vs-asap-synthetic/README.md b/rqe-optimizer/results/autosketch-vs-asap-synthetic/README.md new file mode 100644 index 00000000..9f90346d --- /dev/null +++ b/rqe-optimizer/results/autosketch-vs-asap-synthetic/README.md @@ -0,0 +1,33 @@ +# Synthetic workload: AutoSketch vs. ASAP + +Results of ProjectASAP/ASAPQuery#777's synthetic workload grid (§6 "Workload +grid"): the dashboard default, the 10 templates, shared replicas {8, 64} and +strictness {loose, strict}, each solved by ASAP, PerQuery-CostAware and +AutoSketch-Adapted at both weight settings (CPU only; Fargate prices) and +the SLA grid {0.01, 0.1, 1, 10} ms and none. + +**Pending:** the full run waits for the targeted saturation run at the +workload's data shape (θ = 1.1, K = 1e4, quantiles on the Zipf ranks; +#777 §6) and a cost re-export (#157). Results are committed after it. + +## Reproduce + +```sh +# Tables and plan.tsv (inputs as in rqe-optimizer/data/autosketch-eval/README.md, +# plus #140's synthetic_cardinalities curves and the cost table's exact rows). +python3 scripts/export_autosketch_eval_table.py --synthetic \ + --curves out_grid_1e7_cost/saturation_curve.csv synthetic_cardinalities/saturation_curve.csv \ + --curves-big out_1e9/saturation_curve.csv \ + --saturation out_grid_1e7_cost/saturation.csv synthetic_cardinalities/saturation.csv \ + --saturation-big out_1e9/saturation.csv \ + --merge-curves out_merge_topk/saturation_merge_curve.csv \ + out_merge_quantile/saturation_merge_curve.csv \ + synthetic_cardinalities/saturation_merge_curve.csv \ + --cost-records out_grid_1e7_cost/saturation_cost.jsonl \ + --exact-costs COSTS/rqe_atomic_costs.json \ + --out TABLES +scripts/run_autosketch_vs_asap_synthetic.sh TABLES run +scripts/run_autosketch_vs_asap_synthetic.sh TABLES plot +# ASAP's one-time profiling time (#777 §7) +scripts/profiling_time.py rqe-optimizer/data/profiling-time.json NAME=FILE.jsonl ... +``` diff --git a/scripts/profiling_time.py b/scripts/profiling_time.py new file mode 100644 index 00000000..5f182be8 --- /dev/null +++ b/scripts/profiling_time.py @@ -0,0 +1,75 @@ +#!/usr/bin/env python3 +"""Wall time of the sketch-bench runs behind the evaluation tables: ASAP's +one-time profiling cost (ProjectASAP/ASAPQuery#777 section 7). + +Usage: scripts/profiling_time.py OUT.json NAME=FILE.jsonl [NAME=FILE.jsonl ...] + +Each FILE is a raw approxbench record stream (saturation_accuracy.jsonl, +saturation_cost.jsonl). A run's wall time is the span between its first and +last record timestamp (any `timestamp` or `*_timestamp` field), which +undercounts by the duration of its last measurement and overcounts by any +pause between a stopped run and its resumption. A file without timestamps is +reported as missing rather than estimated. +""" + +import json +import re +import sys +from datetime import datetime + +STAMP = re.compile(r"\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d") + + +def stamps(obj, out): + if isinstance(obj, dict): + for key, value in obj.items(): + if key == "timestamp" or key.endswith("_timestamp"): + if isinstance(value, str) and STAMP.match(value): + out.append(datetime.fromisoformat(value[:19])) + else: + stamps(value, out) + elif isinstance(obj, list): + for value in obj: + stamps(value, out) + + +def span(path): + found = [] + records = 0 + with open(path) as f: + for line in f: + if line.strip(): + records += 1 + stamps(json.loads(line), found) + if not found: + return {"file": path, "records": records, "wall_secs": None, + "note": "no timestamps: wall time missing"} + return {"file": path, "records": records, "first": min(found).isoformat() + "Z", + "last": max(found).isoformat() + "Z", + "wall_secs": (max(found) - min(found)).total_seconds()} + + +def main(): + out = sys.argv[1] + runs = {} + for arg in sys.argv[2:]: + name, path = arg.split("=", 1) + runs[name] = span(path) + known = [r["wall_secs"] for r in runs.values() if r["wall_secs"] is not None] + result = { + "method": "first-to-last record timestamp per run (see scripts/profiling_time.py)", + "runs": runs, + "total_wall_secs": sum(known), + "missing": [n for n, r in runs.items() if r["wall_secs"] is None], + } + with open(out, "w") as f: + json.dump(result, f, indent=1) + f.write("\n") + for name, r in runs.items(): + print(f"{name}: {r['wall_secs']} s over {r['records']} records", file=sys.stderr) + print(f"total {result['total_wall_secs'] / 3600:.2f} h; missing {result['missing']}", + file=sys.stderr) + + +if __name__ == "__main__": + main()