From f3a7df3ecb010ce90f171cac0cfc048777465753 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 7 Sep 2026 10:31:30 -0400 Subject: [PATCH 01/18] feat(planner): load profiled atomic costs --- .design_docs/cost-optimizer-decision-map.md | 147 ++++++++++ .../optimizer-v1-implementation-plan.md | 15 +- CONTEXT.md | 24 ++ asap-planner-rs/src/bin/candidate_gen_dump.rs | 19 +- asap-planner-rs/src/bin/optimizer_cli.rs | 20 +- asap-planner-rs/src/optimizer/atomic_costs.rs | 267 +++++++++++++++++- asap-planner-rs/src/optimizer/greedy.rs | 1 + asap-planner-rs/src/optimizer/mod.rs | 4 +- 8 files changed, 471 insertions(+), 26 deletions(-) create mode 100644 .design_docs/cost-optimizer-decision-map.md create mode 100644 CONTEXT.md diff --git a/.design_docs/cost-optimizer-decision-map.md b/.design_docs/cost-optimizer-decision-map.md new file mode 100644 index 00000000..58a0764e --- /dev/null +++ b/.design_docs/cost-optimizer-decision-map.md @@ -0,0 +1,147 @@ +# Cost-based optimizer decision map + +Goal: turn a PromQL workload with query frequencies and dataset-conditioned +sketch measurements into an offline deployment plan, then collect evidence that +the selected plan improves the stated objective without violating accuracy. + +## Progress log + +- 2026-09-07: Located the existing offline planner entry point: + `asap-optimizer-cli`. `Controller::generate()` still takes the hardcoded + planner path, so production wiring is intentionally not the first milestone. +- 2026-09-07: Verified a wire-contract gap: sketch-bench emits a versioned, + workload-profiled atomic-cost document, while ASAPQuery currently loads a + legacy flat entry array. A profile-aware loader and explicit selector are the + first integration change. +- 2026-09-07: Distinguished grouping labels from sketch keys. Grouping labels + determine how many sketch instances the planner deploys; sketch-key + properties belong to a measured benchmark profile. +- 2026-09-07: Chosen first vertical slice: KLL backing PromQL + `quantile_over_time`. The next unresolved concrete inputs are a source + metric, KLL input value/key column, trace time slice, and grouping labels. +- 2026-09-07: Candidate external workload specs already present in + `sketch-bench`: Alibaba microservices CPU (`cpu_utilization`, grouped by + `msname`, one-minute slice); Google task CPU (`cpu_rate`, grouped by + `machine_id`, three-minute slice); and Datadog/BOOM (`target`, scalar, + one-minute slice). Existing ASAPQuery Prometheus replay configurations also + exist for Alibaba node CPU and Google CPU. Recommended first case: Google + CPU, because it has both a sketch-bench workload spec and a matching + Prometheus replay configuration with explicit labels. +- 2026-09-07: Verified the selected Google trace is available locally: + `google-cluster-data/ClusterData2011/clusterdata-2011-2/task_usage/part-00262-of-00500.csv.gz` + (92 MB). The external loader supports numeric f64 grouped workloads, and the + selected spec uses `cpu_rate`, grouped by `machine_id`, over a three-minute + time slice. `kll-percall` is the correct benchmark variant because the + ASAPQuery runtime invokes KLL `quantile()` per query. Remaining prerequisites: + (1) a matching `quantile_over_time` workload and series-inventory CSV, + (2) profile-aware atomic-cost loading/selection in ASAPQuery, and + (3) explicit accuracy, arrival-rate, and exact-baseline assumptions. +- 2026-09-07: Inspected sketch-bench PR #124. It is the correct producer-side + contract: a versioned document, one profile per exact external workload, + required `query_accuracy`, and no cross-profile merging. Decision: do not + add an ASAPQuery-specific `logical_metric` field to PR #124. Its + `value_column` identifies the physical trace column measured; a Prometheus + metric name is a query-engine identity that may rename or transform that + column. ASAPQuery owns the explicit mapping from a queried metric to a + benchmark profile. Do not add KLL key columns for this slice: KLL consumes + numeric values and keyed external workloads are intentionally unsupported. + ASAPQuery should mirror the versioned types, select exactly one external + profile from an explicit selector, and pass only its entries to the existing + candidate resolver. +- 2026-09-07: Corrected an earlier naming example after inspecting collector + code. `google_mean_cpu_usage_rate_0` is an older experiment-specific name; + the existing evaluation mapper emits `google_cluster_2019_cpu_rate` from + raw `cpu_rate`, and creates a `_q_kll` alias only to route a second copy of + the same samples through KLL. These are naming/routing conventions, not + distinct source values. For the first slice, use one canonical raw-stream + identity throughout (recommended Prometheus-safe name: + `google_task_usage_cpu_rate`), and do not require a separate logical-metric + field in the sketch-bench document. +- 2026-09-07: Validated the PR-123/PR-124 dependency analysis. PR 123 changes + the underlying benchmark-record schema; PR 124 is the atomic-cost consumer + break and must be handled via its independent document schema version. Two + implementation corrections: `query_accuracy` is a required map of named + metrics, not one `f64`; and workload selection can happen once at the + ASAPQuery load boundary, returning the selected profile's existing flat + `AtomicCostTable`. This avoids threading workload identity through every + candidate resolver while still rejecting zero or ambiguous matches before + any candidate lookup. +- 2026-09-07: Implemented the ASAPQuery consumer seam on branch + `feat/profiled-atomic-cost-loader`. It mirrors PR 124's versioned document, + external workload, required accuracy map, strict schema validation, and + exact single-profile selection. The two offline optimizer CLIs now require a + JSON `profiles[].workload` selector whenever `--atomic-costs` is supplied; + selected entries preserve the existing flat resolver interface. Focused + loader and full planner tests pass (216 tests plus doc-tests); the latter was + run outside the sandbox because its existing ClickHouse mock opens a local + listener. +- 2026-09-07: Review follow-up: centralized CLI profile loading behind one + optimizer module interface and added explicit zero-match coverage. Synthetic + profile descriptions remain opaque JSON because sketch-bench's data-generator + schema is independently versioned; document, profile, entry, and external + workload fields remain strict at this consumer boundary. + +## #1: What identifies a benchmark cost profile? + +Type: Discuss + +### Question + +Which dataset properties are expected to change insert, merge, query, memory, +or accuracy enough that they must select a distinct atomic-cost profile? Decide +the first paper-scale profile matrix and the semantics for choosing a profile at +planning time. + +### Answer + +Open. `sketch-bench` already emits `AtomicCostDocument { schema_version, +profiles: [{ workload, entries }] }`; the optimizer currently reads the older +flat entry array and therefore has no profile-selection rule. The available +observability corpus is under `../benchmarks/metrics_observability/data` from +the workspace root (Datadog/BOOM and Alibaba traces). Candidate framing from +the user: `(dataset_name, metric_name, keying/aggregated-label names, +time-range)`. The remaining decision is whether labels and range identify an +atomic measurement or instead parameterize the structural cost formula. + +Clarification: **grouping labels** are the PromQL `GROUP BY` labels and +partition the metric stream into separate sketch instances. A **sketch key** +is the value or label tuple inserted into keyed sketches such as HLL, CMS, and +Hydra. Grouping labels and their observed distinct-group count are planner +context; the sketch-key distribution/cardinality (and, where relevant, encoded +key size) belong in the benchmark profile. + +For the first vertical slice, scope this decision to one KLL +`quantile_over_time` workload. Do not define the full cross-dataset matrix yet. + +## #2: What is the minimum credible empirical planning loop? + +Blocked by: #1 +Type: Prototype + +### Question + +What end-to-end experiment should prove that measured costs, rather than +hand-tuned constants, change a planner decision appropriately for a fixed +PromQL workload and series inventory? + +### Answer + +Open. The existing offline `asap-optimizer-cli` is the intended harness after +the document-loader/profile-selection gap is closed. + +## #3: What feasibility evidence constrains optimization? + +Blocked by: #1, #2 +Type: Discuss + +### Question + +For each query/sketch/config/dataset case, which accuracy metric and threshold +make a candidate eligible, and how will exact-query cost and arrival rate be +measured rather than assumed? + +### Answer + +Open. `sketch-bench` already retains capability-specific `query_accuracy`, but +the ASAPQuery greedy optimizer does not use it; arrival rate (`rho`) and exact +query cost are currently placeholders. diff --git a/.design_docs/optimizer-v1-implementation-plan.md b/.design_docs/optimizer-v1-implementation-plan.md index 2fde701f..658b172e 100644 --- a/.design_docs/optimizer-v1-implementation-plan.md +++ b/.design_docs/optimizer-v1-implementation-plan.md @@ -322,7 +322,8 @@ cargo run -p asap_planner --bin asap-optimizer-cli -- \ --dataset \ --data-ingestion-interval-ms 60000 \ [--rho 1.0] \ - [--atomic-costs ] + [--atomic-costs \ + --atomic-cost-workload ] ``` Takes the same `ControllerConfig` YAML format as `asap-planner --input_config`. @@ -332,7 +333,11 @@ group count; no live Prometheus connection is needed. A `metrics:` hints block, present, is checked against the dataset and mismatches fail loudly. Prints deployed streaming configs and query configs to stdout. `--rho` is the placeholder arrival rate (see TODOs below — not real yet). `--atomic-costs` is -optional; omit it and ordinary unbenchmarked candidates use the flat stub, while +optional; when supplied it requires `--atomic-cost-workload`, a JSON file +containing the exact `profiles[].workload` value from that benchmark artifact. +The loader validates the document schema and rejects a selector that matches +zero or multiple profiles; it never mixes entries across workloads. Omit both +flags and ordinary unbenchmarked candidates use the flat stub, while CMS-with-heap candidates warn and are dropped until a matching reference row is available. ### Running with real sketch-bench costs @@ -352,13 +357,15 @@ CMS-with-heap candidates warn and are dropped until a matching reference row is # See what each candidate would cost, before selection: cargo run -p asap_planner --bin candidate-gen-dump -- \ --input_config workload.yaml --data-ingestion-interval-ms 60000 \ - --atomic-costs path/to/atomic_costs.json + --atomic-costs path/to/atomic_costs.json \ + --atomic-cost-workload path/to/workload-selector.json # Run the actual optimizer: cargo run -p asap_planner --bin asap-optimizer-cli -- \ --input_config workload.yaml --data-ingestion-interval-ms 60000 \ --dataset path/to/series-inventory.csv \ - --atomic-costs path/to/atomic_costs.json + --atomic-costs path/to/atomic_costs.json \ + --atomic-cost-workload path/to/workload-selector.json ``` `candidate-gen-dump`'s output labels each resolved params row `[real]` or `[stub]`; candidates diff --git a/CONTEXT.md b/CONTEXT.md new file mode 100644 index 00000000..a83c52ec --- /dev/null +++ b/CONTEXT.md @@ -0,0 +1,24 @@ +# ASAPQuery + +Terms used by the cost-based sketch optimizer and its benchmarking pipeline. + +## Language + +**Grouping labels**: +PromQL `GROUP BY` labels that partition a metric stream into label groups; a +non-subpopulation sketch is instantiated once per distinct group. +_Avoid_: keying labels + +**Sketch key**: +The value or label tuple inserted into and queried from a keyed sketch such as +HLL, CMS, or Hydra. Its distribution, cardinality, and encoded size may affect +the measured atomic cost and accuracy. +_Avoid_: grouping label, partition key + +**Benchmark profile**: +A reproducible description of the input trace slice used to measure an +atomic-cost table for a sketch family and configuration. + +**Planner context**: +Query- and deployment-specific structural inputs, such as label-group count +and retained-window count, that scale atomic costs into a plan cost. diff --git a/asap-planner-rs/src/bin/candidate_gen_dump.rs b/asap-planner-rs/src/bin/candidate_gen_dump.rs index a900125f..b0af6864 100644 --- a/asap-planner-rs/src/bin/candidate_gen_dump.rs +++ b/asap-planner-rs/src/bin/candidate_gen_dump.rs @@ -6,8 +6,8 @@ use std::path::PathBuf; use asap_planner::{ optimizer::{ - enumerate_candidates, extract_aqes, load_atomic_cost_table, resolve_atomic_costs, - AtomicCostTable, AtomicCosts, CandidateConfig, RQE, + enumerate_candidates, extract_aqes, load_optional_selected_atomic_cost_table, + resolve_atomic_costs, AtomicCostTable, AtomicCosts, CandidateConfig, RQE, }, ControllerConfig, }; @@ -28,21 +28,26 @@ struct Args { #[arg(long = "data-ingestion-interval-ms")] scrape_interval_ms: u64, - /// Path to sketch-bench's exported atomic-cost table (see ASAPQuery#524). + /// Path to sketch-bench's versioned atomic-cost document. /// When given, each params row also prints its resolved AtomicCosts -- /// real (from the table) or the flat stub (unbenchmarked family, or this /// exact param point missing from the table) -- labeled which. #[arg(long = "atomic-costs")] atomic_costs: Option, + + /// JSON `profiles[].workload` value selecting exactly one measured profile. + #[arg(long = "atomic-cost-workload", requires = "atomic_costs")] + atomic_cost_workload: Option, } fn main() -> anyhow::Result<()> { let args = Args::parse(); - let atomic_cost_table = match &args.atomic_costs { - Some(path) => load_atomic_cost_table(path)?, - None => AtomicCostTable::default(), - }; + let atomic_cost_table = load_optional_selected_atomic_cost_table( + args.atomic_costs.as_deref(), + args.atomic_cost_workload.as_deref(), + )? + .unwrap_or_default(); let yaml_str = std::fs::read_to_string(&args.input_config)?; let config: ControllerConfig = serde_yaml::from_str(&yaml_str)?; diff --git a/asap-planner-rs/src/bin/optimizer_cli.rs b/asap-planner-rs/src/bin/optimizer_cli.rs index eda78f66..b6521b70 100644 --- a/asap-planner-rs/src/bin/optimizer_cli.rs +++ b/asap-planner-rs/src/bin/optimizer_cli.rs @@ -7,7 +7,7 @@ use std::path::PathBuf; use asap_planner::optimizer::{ - load_atomic_cost_table, run_greedy_pipeline, AtomicCostTable, SeriesDataset, + load_optional_selected_atomic_cost_table, run_greedy_pipeline, AtomicCostTable, SeriesDataset, }; use asap_planner::ControllerConfig; use clap::Parser; @@ -35,14 +35,21 @@ struct Args { #[arg(long = "rho", default_value = "1.0", value_parser = parse_positive_finite)] rho: f64, - /// Path to the atomic-cost table sketch-bench's `atomic-costs` subcommand - /// exports (see ASAPQuery#524, sketch-bench#30). Omitted: every + /// Path to the versioned atomic-cost document sketch-bench's `atomic-costs` + /// subcommand exports. Requires --atomic-cost-workload to select exactly + /// one measured workload profile. Omitted: every /// benchmarked-family candidate (CMS/HLL/KLL) is dropped, since there is /// no data to cost it at — only trivial accumulators and EXACT remain /// selectable. #[arg(long = "atomic-costs")] atomic_costs: Option, + /// JSON `profiles[].workload` value copied from the sketch-bench atomic-cost + /// document. This makes the empirical workload profile explicit and avoids + /// mixing costs from different traces or time windows. + #[arg(long = "atomic-cost-workload", requires = "atomic_costs")] + atomic_cost_workload: Option, + #[arg(short, long, action = clap::ArgAction::Count)] verbose: u8, } @@ -70,8 +77,11 @@ fn main() -> anyhow::Result<()> { let config: ControllerConfig = serde_yaml::from_str(&yaml_str)?; let dataset = SeriesDataset::from_path(&args.dataset)?; - let atomic_cost_table = match &args.atomic_costs { - Some(path) => load_atomic_cost_table(path)?, + let atomic_cost_table = match load_optional_selected_atomic_cost_table( + args.atomic_costs.as_deref(), + args.atomic_cost_workload.as_deref(), + )? { + Some(table) => table, None => { tracing::warn!( "no --atomic-costs supplied; CMS/HLL/KLL candidates will never be selected" diff --git a/asap-planner-rs/src/optimizer/atomic_costs.rs b/asap-planner-rs/src/optimizer/atomic_costs.rs index b7e0e8f0..43f15a76 100644 --- a/asap-planner-rs/src/optimizer/atomic_costs.rs +++ b/asap-planner-rs/src/optimizer/atomic_costs.rs @@ -8,7 +8,7 @@ //! hand; `atomic_cost_entry_deserializes_sketch_benchs_documented_shape` //! below is a canary for drift. -use std::collections::HashMap; +use std::collections::{BTreeMap, HashMap}; use std::path::Path; use promql_utilities::query_logics::enums::AggregationType; @@ -22,8 +22,58 @@ use super::constants::{ use super::cost_model::AtomicCosts; const CMS_HEAP_BENCHMARK: &str = "cms-heap-topk-regularpath-vector2d"; +pub const ATOMIC_COST_SCHEMA_VERSION: u32 = 1; + +/// Versioned atomic-cost document emitted by `approxbench atomic-costs`. +/// +/// A profile is deliberately selected before candidate resolution: costs from +/// different input workloads must never be mixed by a flat lookup. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct AtomicCostDocument { + pub schema_version: u32, + pub profiles: Vec, +} #[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct AtomicCostProfile { + pub workload: WorkloadDescription, + pub entries: Vec, +} + +/// The provenance of the input data on which atomic costs were measured. +/// Synthetic descriptions remain opaque because their generator schema evolves +/// independently; external profiles are represented explicitly for selection. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum WorkloadDescription { + Synthetic { description: Value }, + External(ExternalWorkload), +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ExternalWorkload { + pub source: String, + pub dataset: String, + pub mode: String, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub key_columns: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub group_columns: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub variate: Option, + pub value_column: String, + pub window_start_ns: i64, + pub window_end_ns: i64, + pub records_loaded: u64, + pub source_timestamp_unit: String, + pub timestamp_unit: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] pub struct AtomicCostEntry { pub sketch: String, pub sketch_config: Value, @@ -31,16 +81,99 @@ pub struct AtomicCostEntry { pub insert_cpu_secs: f64, pub merge_cpu_secs: f64, pub query_cpu_secs: f64, + pub query_accuracy: BTreeMap, } pub type AtomicCostTable = Vec; -/// Read an `AtomicCostTable` exported by `sketch-bench atomic-costs`. -pub fn load_atomic_cost_table(path: &Path) -> anyhow::Result { +/// Parse a standalone JSON workload selector. The selector is the exact +/// `profiles[].workload` value copied from the benchmark artifact, making the +/// selected empirical input explicit in an offline planning run. +pub fn load_workload_selector(path: &Path) -> anyhow::Result { + let raw = std::fs::read_to_string(path).map_err(|e| { + anyhow::anyhow!( + "reading atomic-cost workload selector {}: {e}", + path.display() + ) + })?; + serde_json::from_str(&raw).map_err(|e| { + anyhow::anyhow!( + "parsing atomic-cost workload selector {}: {e}", + path.display() + ) + }) +} + +/// Read a versioned `sketch-bench atomic-costs` document and return entries +/// from exactly one requested workload profile. +pub fn load_atomic_cost_table( + path: &Path, + workload: &WorkloadDescription, +) -> anyhow::Result { let raw = std::fs::read_to_string(path) .map_err(|e| anyhow::anyhow!("reading atomic-cost table {}: {e}", path.display()))?; - serde_json::from_str(&raw) - .map_err(|e| anyhow::anyhow!("parsing atomic-cost table {}: {e}", path.display())) + let document: AtomicCostDocument = serde_json::from_str(&raw) + .map_err(|e| anyhow::anyhow!("parsing atomic-cost document {}: {e}", path.display()))?; + + if document.schema_version != ATOMIC_COST_SCHEMA_VERSION { + anyhow::bail!( + "unsupported atomic-cost schema_version {} in {} (this planner supports {})", + document.schema_version, + path.display(), + ATOMIC_COST_SCHEMA_VERSION + ); + } + + let matches: Vec<_> = document + .profiles + .iter() + .filter(|profile| profile.workload == *workload) + .collect(); + match matches.as_slice() { + [profile] => Ok(profile.entries.clone()), + [] => anyhow::bail!( + "no atomic-cost profile in {} matches workload selector {}", + path.display(), + serde_json::to_string(workload).unwrap_or_else(|_| "".into()) + ), + _ => anyhow::bail!( + "{} atomic-cost profiles in {} match workload selector {}; expected exactly one", + matches.len(), + path.display(), + serde_json::to_string(workload).unwrap_or_else(|_| "".into()) + ), + } +} + +/// Load the selector artifact and return the corresponding empirical table. +/// Offline callers use this single interface so selector validation cannot +/// drift between planner tools. +pub fn load_selected_atomic_cost_table( + document_path: &Path, + selector_path: &Path, +) -> anyhow::Result { + let workload = load_workload_selector(selector_path)?; + load_atomic_cost_table(document_path, &workload) +} + +/// Resolve the optional atomic-cost CLI inputs as one unit. A document without +/// its workload selector is invalid; callers can keep their no-document +/// fallback without duplicating that validation. +pub fn load_optional_selected_atomic_cost_table( + document_path: Option<&Path>, + selector_path: Option<&Path>, +) -> anyhow::Result> { + match (document_path, selector_path) { + (Some(document_path), Some(selector_path)) => { + load_selected_atomic_cost_table(document_path, selector_path).map(Some) + } + (Some(_), None) => anyhow::bail!( + "--atomic-cost-workload is required with --atomic-costs; \ + it must contain the selected profiles[].workload JSON value" + ), + (None, Some(_)) => anyhow::bail!("--atomic-cost-workload requires --atomic-costs"), + (None, None) => Ok(None), + } } /// sketch-bench's (algorithm, params) key for one of ASAPQuery's benchmarked @@ -312,6 +445,120 @@ fn valid_cost_entry(entry: &AtomicCostEntry) -> bool { mod tests { use super::*; + #[test] + fn loader_selects_only_the_requested_external_profile() { + let requested = WorkloadDescription::External(ExternalWorkload { + source: "google".into(), + dataset: "google/task_usage.csv.gz".into(), + mode: "grouped".into(), + key_columns: vec![], + group_columns: vec!["machine_id".into()], + variate: None, + value_column: "cpu_rate".into(), + window_start_ns: 10, + window_end_ns: 20, + records_loaded: 100, + source_timestamp_unit: "microseconds".into(), + timestamp_unit: "nanoseconds".into(), + }); + let other = WorkloadDescription::External(ExternalWorkload { + window_end_ns: 30, + ..match requested.clone() { + WorkloadDescription::External(workload) => workload, + WorkloadDescription::Synthetic { .. } => unreachable!(), + } + }); + let document = serde_json::json!({ + "schema_version": 1, + "profiles": [ + {"workload": other, "entries": []}, + {"workload": requested, "entries": [{ + "sketch": "kll-percall", + "sketch_config": {"algorithm": "kll-percall", "params": {"k": 200}}, + "mem_bytes_per_instance": 6400.0, + "insert_cpu_secs": 1e-8, + "merge_cpu_secs": 1e-3, + "query_cpu_secs": 1e-4, + "query_accuracy": {"mean_rank_err": 0.01} + }]} + ] + }); + let file = tempfile::NamedTempFile::new().unwrap(); + std::fs::write(file.path(), document.to_string()).unwrap(); + + let table = load_atomic_cost_table(file.path(), &requested).unwrap(); + + assert_eq!(table.len(), 1); + assert_eq!(table[0].sketch, "kll-percall"); + } + + #[test] + fn loader_rejects_an_ambiguous_or_incompatible_document() { + let workload = WorkloadDescription::Synthetic { + description: serde_json::json!({"name": "one"}), + }; + let file = tempfile::NamedTempFile::new().unwrap(); + + std::fs::write( + file.path(), + serde_json::json!({ + "schema_version": 2, + "profiles": [] + }) + .to_string(), + ) + .unwrap(); + let err = load_atomic_cost_table(file.path(), &workload).unwrap_err(); + assert!(err.to_string().contains("schema_version 2")); + assert!(err.to_string().contains("supports 1")); + + std::fs::write( + file.path(), + serde_json::json!({ + "schema_version": 1, + "profiles": [ + {"workload": workload, "entries": []}, + {"workload": workload, "entries": []} + ] + }) + .to_string(), + ) + .unwrap(); + let err = load_atomic_cost_table(file.path(), &workload).unwrap_err(); + assert!(err.to_string().contains("2 atomic-cost profiles")); + + std::fs::write( + file.path(), + serde_json::json!({ + "schema_version": 1, + "profiles": [{ + "workload": {"synthetic": {"description": {"name": "other"}}}, + "entries": [] + }] + }) + .to_string(), + ) + .unwrap(); + let err = load_atomic_cost_table(file.path(), &workload).unwrap_err(); + assert!(err.to_string().contains("no atomic-cost profile")); + } + + #[test] + fn atomic_cost_entry_requires_query_accuracy() { + let json = r#"{"sketch":"kll-percall","sketch_config":null,"mem_bytes_per_instance":1.0,"insert_cpu_secs":1.0,"merge_cpu_secs":1.0,"query_cpu_secs":1.0}"#; + assert!(serde_json::from_str::(json).is_err()); + } + + #[test] + fn optional_loader_rejects_an_unselected_document() { + let document = tempfile::NamedTempFile::new().unwrap(); + let err = + load_optional_selected_atomic_cost_table(Some(document.path()), None).unwrap_err(); + assert!(err + .to_string() + .contains("--atomic-cost-workload is required")); + } + fn cms_entry(depth: i64, width: i64) -> AtomicCostEntry { AtomicCostEntry { sketch: "cms-fastpath-vector2d".into(), @@ -323,6 +570,7 @@ mod tests { insert_cpu_secs: 8e-9, merge_cpu_secs: 4.5e-4, query_cpu_secs: 7.8e-8, + query_accuracy: BTreeMap::new(), } } @@ -344,6 +592,7 @@ mod tests { insert_cpu_secs: 2.0, merge_cpu_secs: 4.0, query_cpu_secs: 8.0, + query_accuracy: BTreeMap::new(), } } @@ -363,10 +612,8 @@ mod tests { #[test] fn atomic_cost_entry_deserializes_sketch_benchs_documented_shape() { - // Pinned against a real row sketch-bench's `atomic-costs` subcommand - // actually emitted (out/atomic_costs.json, cms-fastpath-vector2d - // rows=3 cols=1024) -- a canary for the two structs drifting apart. - let json = r#"{"sketch":"cms-fastpath-vector2d","sketch_config":{"algorithm":"cms-fastpath-vector2d","params":{"cols":1024,"rows":3}},"mem_bytes_per_instance":12288.0,"insert_cpu_secs":8.484689139741214e-9,"merge_cpu_secs":0.00045364040539336466,"query_cpu_secs":7.799774697708031e-8}"#; + // Pinned against the current sketch-bench atomic-cost entry shape. + let json = r#"{"sketch":"cms-fastpath-vector2d","sketch_config":{"algorithm":"cms-fastpath-vector2d","params":{"cols":1024,"rows":3}},"mem_bytes_per_instance":12288.0,"insert_cpu_secs":8.484689139741214e-9,"merge_cpu_secs":0.00045364040539336466,"query_cpu_secs":7.799774697708031e-8,"query_accuracy":{"relative_error":0.01}}"#; let entry: AtomicCostEntry = serde_json::from_str(json).expect("documented shape parses"); assert_eq!(entry.sketch, "cms-fastpath-vector2d"); assert_eq!(entry.mem_bytes_per_instance, 12288.0); @@ -512,6 +759,7 @@ mod tests { insert_cpu_secs: 1.68e-9, merge_cpu_secs: 2.76e-4, query_cpu_secs: 1.23e-4, + query_accuracy: BTreeMap::new(), }]; let hll_params = HashMap::from([("precision".to_string(), Value::from(14u64))]); assert!(resolve_atomic_costs(&hll_table, AggregationType::HLL, &hll_params).is_some()); @@ -523,6 +771,7 @@ mod tests { insert_cpu_secs: 1.6e-8, merge_cpu_secs: 1.0e-3, query_cpu_secs: 1.6e-4, + query_accuracy: BTreeMap::new(), }]; let kll_params = HashMap::from([("K".to_string(), Value::from(200u64))]); assert!( diff --git a/asap-planner-rs/src/optimizer/greedy.rs b/asap-planner-rs/src/optimizer/greedy.rs index 3a3719dd..fbcd61a3 100644 --- a/asap-planner-rs/src/optimizer/greedy.rs +++ b/asap-planner-rs/src/optimizer/greedy.rs @@ -223,6 +223,7 @@ mod tests { insert_cpu_secs: 0.0, merge_cpu_secs: 0.0, query_cpu_secs: 0.0, + query_accuracy: std::collections::BTreeMap::new(), }]; let aqe = make_aqe(Statistic::Topk, 60_000, 60_000, 1.0 / 60.0); let solution = greedy_assign( diff --git a/asap-planner-rs/src/optimizer/mod.rs b/asap-planner-rs/src/optimizer/mod.rs index e88fc759..521db371 100644 --- a/asap-planner-rs/src/optimizer/mod.rs +++ b/asap-planner-rs/src/optimizer/mod.rs @@ -12,7 +12,9 @@ pub mod translator; pub use aqe_extractor::{extract_aqes, RQE}; pub use atomic_costs::{ - load_atomic_cost_table, resolve_atomic_costs, AtomicCostEntry, AtomicCostTable, + load_atomic_cost_table, load_optional_selected_atomic_cost_table, + load_selected_atomic_cost_table, resolve_atomic_costs, AtomicCostEntry, AtomicCostTable, + ExternalWorkload, WorkloadDescription, }; pub use candidate_gen::{ enumerate_candidates, enumerate_candidates_with_label_group_count, CandidateConfig, From f272b5f95d8092e0afb880f4f08eea5837cfd32f Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 9 Sep 2026 08:35:26 -0400 Subject: [PATCH 02/18] feat(optimizer): enforce KLL rank-error limits --- .design_docs/cost-optimizer-decision-map.md | 93 +++++++++++++++++++ asap-planner-rs/src/bin/candidate_gen_dump.rs | 1 + asap-planner-rs/src/config/input.rs | 4 + .../src/optimizer/aqe_extractor.rs | 79 +++++++++++----- asap-planner-rs/src/optimizer/atomic_costs.rs | 26 ++++++ .../src/optimizer/candidate_gen.rs | 1 + asap-planner-rs/src/optimizer/cost_model.rs | 1 + asap-planner-rs/src/optimizer/dataset.rs | 1 + asap-planner-rs/src/optimizer/greedy.rs | 59 ++++++++++-- asap-planner-rs/src/optimizer/pipeline.rs | 1 + asap-planner-rs/src/optimizer/solution.rs | 4 + 11 files changed, 239 insertions(+), 31 deletions(-) diff --git a/.design_docs/cost-optimizer-decision-map.md b/.design_docs/cost-optimizer-decision-map.md index 58a0764e..d09893c3 100644 --- a/.design_docs/cost-optimizer-decision-map.md +++ b/.design_docs/cost-optimizer-decision-map.md @@ -113,6 +113,99 @@ key size) belong in the benchmark profile. For the first vertical slice, scope this decision to one KLL `quantile_over_time` workload. Do not define the full cross-dataset matrix yet. +2026-09-07 next prerequisite: the current Google benchmark workload groups +KLL input by `machine_id`, whereas ordinary PromQL `quantile_over_time` is +evaluated per input time series. The raw Google task-usage identity also +contains `job_id` and `task_index`; the planner's KLL model is marked +non-subpopulation-aware and currently scales by query grouping count. Before a +KLL sweep, choose and implement the computational unit consistently in the +benchmark, series inventory, and planner: (a) one sketch per complete source +series (the recommended standard-PromQL interpretation), (b) an explicitly +defined pooled/global ASAP operation, or (c) an explicitly defined per-machine +operation. Do not treat a per-machine benchmark as evidence for a per-series +query without this alignment. + +2026-09-08 scope decision: for the first optimizer experiment, treat every +spatial filter as having selectivity 1.0. Thus each candidate uses the metric's +full arrival rate for ingest costing; no selectivity estimator is needed yet. +Filters can still be retained syntactically for query/config identity, but do +not reduce estimated arrival rate or instance count in this slice. A later +extension can estimate selectivity from the series/sample inventory and use +`arrival_rate_hz * selectivity` for the affected configuration. + +2026-09-07 current-state evidence: the 2011 Google OTLP mapper exports +`google_cluster_2011_cpu_rate` and identifies an uncapped series by the full +attribute tuple `(zone, rack, host, service, task)`. `zone` and `rack` are +deterministic functions of `machine_id`; `host` is `machine_id`, `service` is +`job_id`, and `task` is `task_index`, so the independent source identity is +effectively `(machine_id, job_id, task_index)`. With a positive cardinality +cap, all three are instead projected to a common hashed cell. The current +sketch-bench Google workload is grouped only by `machine_id`, `cpu_rate`, and +a fixed three-minute window. The current planner loads an inventory of full +label tuples, but `SeriesDataset::profile` returns `1` for no query grouping +labels and otherwise returns the number of distinct requested groups. That +count is copied into every candidate; KLL is non-subpopulation-aware, so it +multiplies KLL memory and query/merge CPU by this count. Crucially, the +canonical `build_query_requirements_promql` helper assigns **all metric-schema +labels** to an OnlyTemporal query such as `quantile_over_time`; it therefore +does count one KLL per exported series when the inventory/schema are complete. +The `1` case applies to a query whose result has an empty grouping (for example +a spatial aggregate with no `by (...)`), not to a plain temporal quantile. + +2026-09-08 correction: prior notes incorrectly claimed that a plain +`quantile_over_time` creates empty `QueryRequirements.grouping_labels` and +therefore one KLL. In ASAPQuery-only scope this is false: +`asap_types::build_query_requirements_promql` detects the absence of a spatial +aggregation and preserves all labels from `PromQLSchema`. Collector behavior +is out of scope for this research loop. The remaining issue is only scenario +alignment: the current sketch-bench profile is per machine, whereas a planner +inventory/schema may describe per-series machine/job/task KLLs. Use matching +synthetic metric projections and benchmark `group_columns` for each study. + +2026-09-08 reduced TODO after correction: no ASAPQuery partition-key model is +needed for the initial standard temporal-KLL slice. Select one synthetic metric +scenario; make its ASAPQuery schema and unique-series inventory match it; +benchmark the identical raw grouping in sketch-bench; export/select that +profile; and run the optimizer. The remaining planner work is profile plumbing +already implemented on `feat/profiled-atomic-cost-loader` plus a small +reproducible experiment fixture. Filter selectivity remains fixed at 1.0. + +2026-09-08 implementation split: a first measured-cost KLL run requires no +additional ASAPQuery optimizer algorithm change after the profiled-cost-loader +branch lands. It needs experiment infrastructure only: a scenario-matching +series inventory and workload YAML in ASAPQuery, an external Google KLL sweep +and atomic-cost export in sketch-bench, plus a selector and reproducible runner. +One subsequent, meaningful ASAPQuery code slice remains for a constrained +optimizer: `ControllerOptions.accuracy_sla` is parsed but not propagated to an +AQE or compared with the selected entry's `query_accuracy["mean_rank_err"]`. +Its semantics must be fixed explicitly (recommended: maximum acceptable mean +rank error, e.g. 0.02) before adding that feasibility filter. Existing +untracked `asap-tools/experiments/datasets/quantile_demo` artifacts are user +work and are out of scope for this experiment. + +2026-09-08 implemented KLL feasibility slice: `controller_options` now accepts +optional `max_mean_rank_error` (a fraction: `0.02` is 2%). The AQE extractor +propagates it and uses the smallest limit when identical AQEs are deduplicated. +The greedy optimizer rejects a `DatasketchesKLL` candidate unless its selected +atomic-cost row contains finite `query_accuracy.mean_rank_err` at or below the +limit; EXACT remains feasible. Focused optimizer tests cover strictest-limit +deduplication and selection of a more expensive KLL configuration when the +cheaper one exceeds the 2% bound. + +2026-09-07 design refinement: a benchmark need not expose every raw trace +column. A paper experiment may define a **synthetic metric scenario** as a +chosen label projection of a trace (for example, a machine-only metric or a +full machine/job/task metric), provided the projection is recorded and used +consistently. The scenario, not the raw CSV alone, must bind (1) the mapper's +exported metric name and retained labels, (2) the unique-series inventory fed +to the planner, (3) the sketch-bench `group_columns` that identify physical +sketch instances, and (4) the selected atomic-cost workload profile. For the +first ungrouped standard-PromQL KLL temporal quantile, #3 is the number of +exported source series; for an intentionally pooled or per-machine synthetic +metric it is the corresponding scenario-defined instance count. This makes +controlled label-projection experiments credible instead of accidental schema +drift. + ## #2: What is the minimum credible empirical planning loop? Blocked by: #1 diff --git a/asap-planner-rs/src/bin/candidate_gen_dump.rs b/asap-planner-rs/src/bin/candidate_gen_dump.rs index b0af6864..a9c5d616 100644 --- a/asap-planner-rs/src/bin/candidate_gen_dump.rs +++ b/asap-planner-rs/src/bin/candidate_gen_dump.rs @@ -60,6 +60,7 @@ fn main() -> anyhow::Result<()> { qg.queries.iter().map(|q| RQE { query_string: q.clone(), t_repeat_ms: qg.repetition_delay_ms, + max_mean_rank_error: qg.controller_options.max_mean_rank_error, }) }) .collect(); diff --git a/asap-planner-rs/src/config/input.rs b/asap-planner-rs/src/config/input.rs index 66626c18..3c8979f0 100644 --- a/asap-planner-rs/src/config/input.rs +++ b/asap-planner-rs/src/config/input.rs @@ -81,6 +81,10 @@ pub struct QueryGroup { pub struct ControllerOptions { pub accuracy_sla: f64, pub latency_sla: f64, + /// KLL-specific feasibility constraint: the selected benchmark entry's + /// `mean_rank_err` must not exceed this fraction (0.02 = 2%). + #[serde(default)] + pub max_mean_rank_error: Option, } #[derive(Debug, Clone, Deserialize)] diff --git a/asap-planner-rs/src/optimizer/aqe_extractor.rs b/asap-planner-rs/src/optimizer/aqe_extractor.rs index 52869d1b..6c05d02f 100644 --- a/asap-planner-rs/src/optimizer/aqe_extractor.rs +++ b/asap-planner-rs/src/optimizer/aqe_extractor.rs @@ -17,6 +17,7 @@ use super::solution::AQE; pub struct RQE { pub query_string: String, pub t_repeat_ms: u64, + pub max_mean_rank_error: Option, } /// Stable deduplication key for an AQE. @@ -33,6 +34,15 @@ struct AQEKey { topk_count_events: Option, } +struct AQEAccumulator { + requirements: QueryRequirements, + query_strings: Vec, + query_frequency_hz: f64, + min_t_repeat_ms: u64, + t_repeat_gcd_ms: u64, + max_mean_rank_error: Option, +} + impl AQEKey { fn from_requirements(req: &QueryRequirements) -> Self { Self { @@ -66,7 +76,7 @@ pub fn extract_aqes( scrape_interval_ms: u64, ) -> Vec { // (key) -> (requirements, query_strings, sum_freq, min_t, gcd_t) - let mut acc: HashMap, f64, u64, u64)> = HashMap::new(); + let mut acc: HashMap = HashMap::new(); for rqe in rqes { if rqe.t_repeat_ms == 0 { @@ -84,21 +94,33 @@ pub fn extract_aqes( match extract_requirements(&leaf, metric_schema, scrape_interval_ms) { Some(req) => { let key = AQEKey::from_requirements(&req); - let entry = acc - .entry(key) - .or_insert_with(|| (req, Vec::new(), 0.0, u64::MAX, 0)); - if !entry.1.contains(&leaf) { - entry.1.push(leaf); + let entry = acc.entry(key).or_insert_with(|| AQEAccumulator { + requirements: req, + query_strings: Vec::new(), + query_frequency_hz: 0.0, + min_t_repeat_ms: u64::MAX, + t_repeat_gcd_ms: 0, + max_mean_rank_error: rqe.max_mean_rank_error, + }); + if !entry.query_strings.contains(&leaf) { + entry.query_strings.push(leaf); } // query_frequency_hz must stay in Hz (queries per real second) // regardless of t_repeat_ms's internal unit — 1000.0 / ms, not 1.0 / ms. - entry.2 += 1000.0 / rqe.t_repeat_ms as f64; - entry.3 = entry.3.min(rqe.t_repeat_ms); - entry.4 = if entry.4 == 0 { + entry.query_frequency_hz += 1000.0 / rqe.t_repeat_ms as f64; + entry.min_t_repeat_ms = entry.min_t_repeat_ms.min(rqe.t_repeat_ms); + entry.t_repeat_gcd_ms = if entry.t_repeat_gcd_ms == 0 { rqe.t_repeat_ms } else { - gcd(entry.4, rqe.t_repeat_ms) + gcd(entry.t_repeat_gcd_ms, rqe.t_repeat_ms) }; + entry.max_mean_rank_error = + match (entry.max_mean_rank_error, rqe.max_mean_rank_error) { + (Some(a), Some(b)) => Some(a.min(b)), + (Some(a), None) => Some(a), + (None, Some(b)) => Some(b), + (None, None) => None, + }; } None => { warn!( @@ -111,21 +133,14 @@ pub fn extract_aqes( } acc.into_values() - .map( - |( - requirements, - query_strings, - query_frequency_hz, - min_t_repeat_ms, - t_repeat_gcd_ms, - )| AQE { - requirements, - query_strings, - query_frequency_hz, - min_t_repeat_ms, - t_repeat_gcd_ms, - }, - ) + .map(|accumulator| AQE { + requirements: accumulator.requirements, + query_strings: accumulator.query_strings, + query_frequency_hz: accumulator.query_frequency_hz, + min_t_repeat_ms: accumulator.min_t_repeat_ms, + t_repeat_gcd_ms: accumulator.t_repeat_gcd_ms, + max_mean_rank_error: accumulator.max_mean_rank_error, + }) .collect() } @@ -199,6 +214,7 @@ mod tests { RQE { query_string: query.to_string(), t_repeat_ms: t_ms, + max_mean_rank_error: None, } } @@ -246,6 +262,19 @@ mod tests { assert_eq!(aqes[0].query_strings.len(), 1); // same string, deduplicated } + #[test] + fn deduplicated_aqe_keeps_the_strictest_rank_error_limit() { + let mut loose = rqe("quantile_over_time(0.99, metric[5m])", 60_000); + loose.max_mean_rank_error = Some(0.02); + let mut strict = rqe("quantile_over_time(0.99, metric[5m])", 30_000); + strict.max_mean_rank_error = Some(0.01); + + let aqes = extract_aqes(&[loose, strict], &empty_schema(), 15_000); + + assert_eq!(aqes.len(), 1); + assert_eq!(aqes[0].max_mean_rank_error, Some(0.01)); + } + #[test] fn unsupported_query_is_skipped() { let rqes = vec![rqe("not_a_real_function(metric[5m])", 60_000)]; diff --git a/asap-planner-rs/src/optimizer/atomic_costs.rs b/asap-planner-rs/src/optimizer/atomic_costs.rs index 43f15a76..e1101812 100644 --- a/asap-planner-rs/src/optimizer/atomic_costs.rs +++ b/asap-planner-rs/src/optimizer/atomic_costs.rs @@ -264,6 +264,32 @@ pub fn resolve_atomic_costs( }) } +/// Whether a benchmarked KLL candidate satisfies a requested maximum mean rank +/// error. Missing accuracy is infeasible: a constrained plan must not silently +/// substitute an unmeasured quality value. +pub fn satisfies_max_mean_rank_error( + table: &AtomicCostTable, + agg_type: AggregationType, + params: &HashMap, + max_mean_rank_error: Option, +) -> bool { + let Some(limit) = max_mean_rank_error else { + return true; + }; + if agg_type != AggregationType::DatasketchesKLL { + return true; + } + let Some((sketch, sketch_params)) = sketch_bench_key(agg_type, params) else { + return false; + }; + let expected_config = serde_json::json!({ "algorithm": sketch, "params": sketch_params }); + table + .iter() + .find(|entry| entry.sketch == sketch && entry.sketch_config == expected_config) + .and_then(|entry| entry.query_accuracy.get("mean_rank_err")) + .is_some_and(|error| error.is_finite() && *error <= limit) +} + /// Temporary cost model for the runtime CMS-with-heap implementation. /// /// sketch-bench currently measures a fixed top-k=32 wrapper, while the diff --git a/asap-planner-rs/src/optimizer/candidate_gen.rs b/asap-planner-rs/src/optimizer/candidate_gen.rs index 9ddf13f3..7f2eca9c 100644 --- a/asap-planner-rs/src/optimizer/candidate_gen.rs +++ b/asap-planner-rs/src/optimizer/candidate_gen.rs @@ -351,6 +351,7 @@ mod tests { query_frequency_hz: 1.0 / 60.0, min_t_repeat_ms: min_t, t_repeat_gcd_ms: min_t, + max_mean_rank_error: None, } } diff --git a/asap-planner-rs/src/optimizer/cost_model.rs b/asap-planner-rs/src/optimizer/cost_model.rs index 3f11ef19..53a3ff02 100644 --- a/asap-planner-rs/src/optimizer/cost_model.rs +++ b/asap-planner-rs/src/optimizer/cost_model.rs @@ -193,6 +193,7 @@ mod tests { query_frequency_hz: 1.0 / 60.0, min_t_repeat_ms: min_t, t_repeat_gcd_ms: min_t, + max_mean_rank_error: None, } } diff --git a/asap-planner-rs/src/optimizer/dataset.rs b/asap-planner-rs/src/optimizer/dataset.rs index 9f97219b..de1edc30 100644 --- a/asap-planner-rs/src/optimizer/dataset.rs +++ b/asap-planner-rs/src/optimizer/dataset.rs @@ -558,6 +558,7 @@ mod tests { query_frequency_hz: 1.0, min_t_repeat_ms: 1, t_repeat_gcd_ms: 1, + max_mean_rank_error: None, }; assert!(matches!( diff --git a/asap-planner-rs/src/optimizer/greedy.rs b/asap-planner-rs/src/optimizer/greedy.rs index fbcd61a3..0e8a0fb4 100644 --- a/asap-planner-rs/src/optimizer/greedy.rs +++ b/asap-planner-rs/src/optimizer/greedy.rs @@ -2,7 +2,7 @@ use std::collections::HashMap; use tracing::debug; -use super::atomic_costs::{resolve_atomic_costs, AtomicCostTable}; +use super::atomic_costs::{resolve_atomic_costs, satisfies_max_mean_rank_error, AtomicCostTable}; use super::candidate_gen::enumerate_candidates_with_label_group_count; use super::cost_model::{ingest_cost, query_cost, total_cost_rate, AtomicCosts, CostWeights}; use super::dataset::ProfileKey; @@ -53,11 +53,21 @@ pub fn greedy_assign( // no sketch_type/params for the table to key on. let costs = match &c.config { None => AtomicCosts::default(), - Some(cfg) => resolve_atomic_costs( - atomic_cost_table, - cfg.aggregation_type, - &cfg.parameters, - )?, + Some(cfg) => { + if !satisfies_max_mean_rank_error( + atomic_cost_table, + cfg.aggregation_type, + &cfg.parameters, + aqe.max_mean_rank_error, + ) { + return None; + } + resolve_atomic_costs( + atomic_cost_table, + cfg.aggregation_type, + &cfg.parameters, + )? + } }; let cost = total_cost_rate(&aqe, &c, arrival_rate_hz, &costs, weights); Some((c, costs, cost)) @@ -124,6 +134,7 @@ mod tests { query_frequency_hz: freq_hz, min_t_repeat_ms: min_t, t_repeat_gcd_ms: min_t, + max_mean_rank_error: None, } } @@ -180,6 +191,7 @@ mod tests { query_frequency_hz: 1.0 / 60.0, min_t_repeat_ms: 60_000, t_repeat_gcd_ms: 60_000, + max_mean_rank_error: None, }; let solution = greedy_assign( vec![aqe.clone()], @@ -247,4 +259,39 @@ mod tests { AggregationType::CountMinSketchWithHeap ); } + + #[test] + fn rank_error_limit_rejects_cheaper_kll_candidate() { + fn kll(k: u64, mean_rank_err: f64) -> AtomicCostEntry { + AtomicCostEntry { + sketch: "kll-percall".into(), + sketch_config: serde_json::json!({ + "algorithm": "kll-percall", "params": { "k": k } + }), + mem_bytes_per_instance: 1.0, + insert_cpu_secs: 0.0, + merge_cpu_secs: 0.0, + query_cpu_secs: k as f64 * 1e-9, + query_accuracy: std::collections::BTreeMap::from([( + "mean_rank_err".into(), + mean_rank_err, + )]), + } + } + + let mut aqe = make_aqe(Statistic::Quantile, 60_000, 60_000, 1.0); + aqe.max_mean_rank_error = Some(0.02); + let solution = greedy_assign( + vec![aqe.clone()], + 60_000, + 1.0, + &vec![kll(200, 0.03), kll(500, 0.01)], + &CostWeights::default(), + &HashMap::from([(ProfileKey::from_requirements(&aqe.requirements), 1)]), + ); + + let config = solution.deployed_configs().values().next().unwrap(); + assert_eq!(config.aggregation_type, AggregationType::DatasketchesKLL); + assert_eq!(config.parameters["K"], serde_json::Value::from(500)); + } } diff --git a/asap-planner-rs/src/optimizer/pipeline.rs b/asap-planner-rs/src/optimizer/pipeline.rs index 54387c70..5855ed0c 100644 --- a/asap-planner-rs/src/optimizer/pipeline.rs +++ b/asap-planner-rs/src/optimizer/pipeline.rs @@ -122,6 +122,7 @@ fn config_to_rqes(config: &ControllerConfig) -> Vec { qg.queries.iter().map(|q| RQE { query_string: q.clone(), t_repeat_ms: qg.repetition_delay_ms, + max_mean_rank_error: qg.controller_options.max_mean_rank_error, }) }) .collect() diff --git a/asap-planner-rs/src/optimizer/solution.rs b/asap-planner-rs/src/optimizer/solution.rs index 34516226..be1f8400 100644 --- a/asap-planner-rs/src/optimizer/solution.rs +++ b/asap-planner-rs/src/optimizer/solution.rs @@ -34,6 +34,10 @@ pub struct AQE { /// every GCD ms align harmonically with all dashboard refresh cycles, /// ensuring every dashboard can always be served a fresh result on-cycle. pub t_repeat_gcd_ms: u64, + + /// Optional KLL feasibility constraint propagated from all query groups + /// contributing to this AQE. Smaller means stricter. + pub max_mean_rank_error: Option, } /// How an AQE is answered from its assigned streaming config. From 53c0afaf2ff12bcae2122a1a0e54e2610e6825f8 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 9 Sep 2026 11:39:55 -0400 Subject: [PATCH 03/18] fix(optimizer): require measured costs for hydra kll --- .design_docs/cost-optimizer-decision-map.md | 58 +++++++++++++++++++ asap-planner-rs/src/optimizer/atomic_costs.rs | 27 ++++++++- 2 files changed, 83 insertions(+), 2 deletions(-) diff --git a/.design_docs/cost-optimizer-decision-map.md b/.design_docs/cost-optimizer-decision-map.md index d09893c3..7b24c699 100644 --- a/.design_docs/cost-optimizer-decision-map.md +++ b/.design_docs/cost-optimizer-decision-map.md @@ -192,6 +192,64 @@ limit; EXACT remains feasible. Focused optimizer tests cover strictest-limit deduplication and selection of a more expensive KLL configuration when the cheaper one exceeds the 2% bound. +2026-09-09 proposed first experiment (discussion, do not run yet): fixed +Google-2011 task-usage CPU trace slice, one synthetic per-machine metric +(`host <- machine_id`), one 3-minute `quantile_over_time(0.99, metric[3m])` +query, selectivity 1.0, and KLL `k in {200,500}`. sketch-bench must measure +insert, merge, per-call quantile, memory, and mean rank error for exactly that +raw grouping/window; ASAPQuery consumes that one selected profile, the matching +unique-host inventory, a measured arrival rate (`records_loaded / 180s`), and +the query repetition rate. Plumbing passes only if both K values resolve as +real costs and the plan is reproducible. The paper-facing decision test should +evaluate a predeclared sweep of rank-error limits: each KLL candidate is +eligible iff its measured error is at most the limit, and the selected plan +must equal the minimum predicted cost among eligible candidates. A separate +hold-out/replay measurement is required before claiming that the atomic model +predicts real end-to-end plan cost; do not call the first slice that validation. + +2026-09-09 agreed provisional control for Experiment 1: temporarily set +ASAPQuery's `EXACT_QUERY_CPU_SECS` to a documented high value on the experiment +branch to force the optimizer to compare feasible KLL candidates, then restore +its original `1e-3` value after the experiment. Do not claim an +exact-vs-approximate result while this forced baseline is active. The +runner/output must label it `forced_exact_baseline`; replacing it with a +measured raw-query baseline is required for a later end-to-end comparison. + +2026-09-09 Experiment 1 run (artifacts retained in +`sketch-bench/output/cost_optimizer_experiments/google_task_usage_cpu_per_machine/2026-09-09/`): +the fixed Google task-usage CPU slice loaded 20,051 rows from the declared +three-minute window and produced 6,481 distinct `machine_id` values. The +scenario exports one synthetic metric `google_task_cpu_rate` with `host <- +machine_id`; its headered unique-series inventory, workload YAML, selector, +raw JSONL passes, flattened records, cost document, planner YAMLs, and planner +logs are all retained there. The profile has two real KLL entries: + +| K | mean rank error | memory / instance | selected under forced exact=100 | +|---|---:|---:|---| +| 200 | 0.00128963 (0.129%) | 6,400 B | limits 0.005, 0.01, 0.02, 0.05 | +| 500 | 0.00063105 (0.063%) | 16,000 B | exploratory limit 0.001 | + +The four predeclared limits are all looser than K=200's measured error, so +they correctly select K=200. The 0.001 result is explicitly exploratory (it +was chosen after observing the measurements) and verifies the intended +feasibility switch: K=200 is rejected and K=500 is selected. Exact was forced +to 100 CPU-seconds/query only for those diagnostic runs and restored to +`1e-3`; a smaller forced value of 1.0 left EXACT cheaper because the KLL plan +holds 6,481 instances, and that run is also retained. None of these results is +an exact-vs-approximate claim. + +The run exposed two plumbing findings. First, the sketch-bench flattener keeps +the *first* shared timing field, while `scripts/export_atomic_costs.sh` says +accuracy should run first because it assumes the last field wins. The preserved +accuracy-first `atomic_costs.json` therefore has zero profiles (one-sample +accuracy wall-time versus five throughput/CPU samples); the cost-first rerun +`atomic_costs_cost_first.json` reduces successfully to one profile/two entries. +The export script should be corrected before this is made a reusable runner. +Second, an empirical profile previously allowed unmeasured HydraKLL candidates +to use a flat stub and win. ASAPQuery now drops HydraKLL when a nonempty +empirical table is present, and treats it as infeasible under a rank-error +limit; focused regression tests cover both cases. + 2026-09-07 design refinement: a benchmark need not expose every raw trace column. A paper experiment may define a **synthetic metric scenario** as a chosen label projection of a trace (for example, a machine-only metric or a diff --git a/asap-planner-rs/src/optimizer/atomic_costs.rs b/asap-planner-rs/src/optimizer/atomic_costs.rs index e1101812..629b16d1 100644 --- a/asap-planner-rs/src/optimizer/atomic_costs.rs +++ b/asap-planner-rs/src/optimizer/atomic_costs.rs @@ -219,8 +219,9 @@ fn sketch_bench_key( /// /// - `agg_type` outside the benchmarked families (see [`sketch_bench_key`]): /// `Some(AtomicCosts::default())` — the flat stub, unchanged from before -/// this table existed. Logged, since it's silently wrong for anything -/// sketch-bench could plausibly measure later. +/// this table existed. `HydraKLL` is the exception: once a nonempty +/// empirical table is supplied it is dropped, rather than being allowed to +/// beat measured KLL alternatives with an unmeasured stub. /// TODO(#524): remove this fallback once every family the optimizer can /// select has a real sketch-bench entry; costing should end up 100% /// empirical, with nothing left reading `AtomicCosts::default()`. @@ -242,6 +243,13 @@ pub fn resolve_atomic_costs( return resolve_cms_heap_costs(table, params, &CmsHeapCostAssumptions::default()); } + if agg_type == AggregationType::HydraKLL && !table.is_empty() { + tracing::warn!( + "no sketch-bench atomic-cost data for HydraKLL; dropping it while using empirical costs" + ); + return None; + } + let Some((sketch, sketch_params)) = sketch_bench_key(agg_type, params) else { tracing::warn!( ?agg_type, @@ -276,6 +284,9 @@ pub fn satisfies_max_mean_rank_error( let Some(limit) = max_mean_rank_error else { return true; }; + if agg_type == AggregationType::HydraKLL { + return false; + } if agg_type != AggregationType::DatasketchesKLL { return true; } @@ -684,6 +695,18 @@ mod tests { ); } + #[test] + fn hydra_kll_drops_when_an_empirical_profile_is_present() { + let table = vec![cms_entry(3, 1024)]; + assert!(resolve_atomic_costs(&table, AggregationType::HydraKLL, &HashMap::new()).is_none()); + assert!(!satisfies_max_mean_rank_error( + &table, + AggregationType::HydraKLL, + &HashMap::new(), + Some(0.02), + )); + } + #[test] fn cms_with_heap_without_reference_cost_drops_the_candidate() { // Until sketch-bench has a matching reference row, CMS-with-heap must From 4688117089efe25d8fa4ec307a5e270ef2bddf3e Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 9 Sep 2026 13:42:32 -0400 Subject: [PATCH 04/18] docs: record strict benchmark pass ownership --- .design_docs/cost-optimizer-decision-map.md | 21 +++++++++++++-------- 1 file changed, 13 insertions(+), 8 deletions(-) diff --git a/.design_docs/cost-optimizer-decision-map.md b/.design_docs/cost-optimizer-decision-map.md index 7b24c699..969060be 100644 --- a/.design_docs/cost-optimizer-decision-map.md +++ b/.design_docs/cost-optimizer-decision-map.md @@ -238,14 +238,19 @@ to 100 CPU-seconds/query only for those diagnostic runs and restored to holds 6,481 instances, and that run is also retained. None of these results is an exact-vs-approximate claim. -The run exposed two plumbing findings. First, the sketch-bench flattener keeps -the *first* shared timing field, while `scripts/export_atomic_costs.sh` says -accuracy should run first because it assumes the last field wins. The preserved -accuracy-first `atomic_costs.json` therefore has zero profiles (one-sample -accuracy wall-time versus five throughput/CPU samples); the cost-first rerun -`atomic_costs_cost_first.json` reduces successfully to one profile/two entries. -The export script should be corrected before this is made a reusable runner. -Second, an empirical profile previously allowed unmeasured HydraKLL candidates +The run exposed two plumbing findings. First, accuracy and throughput records +both carried incidental timing metadata, and the old generic flattener kept +the first one it encountered. The preserved accuracy-first `atomic_costs.json` +therefore initially had zero profiles (one-sample accuracy wall-time versus +five throughput/CPU samples). This is now fixed in sketch-bench: flattened +fields have strict primary-pass ownership—accuracy contributes only accuracy, +throughput contributes cost timing/resources, and latency contributes latency. +Unknown/contradictory primary fields fail loudly; merge latency also fails +loudly because `MergedRecord` has no field to represent it. Reflattening the +same preserved accuracy-first raw report now yields +`atomic_costs_strict_accuracy_first.json` with one profile/two entries, so the +export order is no longer a correctness condition. Second, an empirical profile +previously allowed unmeasured HydraKLL candidates to use a flat stub and win. ASAPQuery now drops HydraKLL when a nonempty empirical table is present, and treats it as infeasible under a rank-error limit; focused regression tests cover both cases. From 913d631788cf612d462493ed8078ac792f982ddc Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 9 Sep 2026 13:45:29 -0400 Subject: [PATCH 05/18] docs: clarify strict flatten pass contract --- .design_docs/cost-optimizer-decision-map.md | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/.design_docs/cost-optimizer-decision-map.md b/.design_docs/cost-optimizer-decision-map.md index 969060be..565aaf2d 100644 --- a/.design_docs/cost-optimizer-decision-map.md +++ b/.design_docs/cost-optimizer-decision-map.md @@ -243,10 +243,12 @@ both carried incidental timing metadata, and the old generic flattener kept the first one it encountered. The preserved accuracy-first `atomic_costs.json` therefore initially had zero profiles (one-sample accuracy wall-time versus five throughput/CPU samples). This is now fixed in sketch-bench: flattened -fields have strict primary-pass ownership—accuracy contributes only accuracy, -throughput contributes cost timing/resources, and latency contributes latency. -Unknown/contradictory primary fields fail loudly; merge latency also fails -loudly because `MergedRecord` has no field to represent it. Reflattening the +fields have strict primary-pass ownership—accuracy contributes only query +accuracy, throughput contributes cost timing/resources, and latency contributes +only insert latency. Only representable `(operation, pass)` pairs are accepted; +unknown/contradictory primary fields, query/merge/prepare latency, and other +unrepresentable pairs fail loudly rather than being silently dropped. +Reflattening the same preserved accuracy-first raw report now yields `atomic_costs_strict_accuracy_first.json` with one profile/two entries, so the export order is no longer a correctness condition. Second, an empirical profile From 9b7e139728266830a53c9b58d3d0f323adad7df7 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 9 Sep 2026 13:46:38 -0400 Subject: [PATCH 06/18] docs: record required flatten fields --- .design_docs/cost-optimizer-decision-map.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/.design_docs/cost-optimizer-decision-map.md b/.design_docs/cost-optimizer-decision-map.md index 565aaf2d..9e38971d 100644 --- a/.design_docs/cost-optimizer-decision-map.md +++ b/.design_docs/cost-optimizer-decision-map.md @@ -247,7 +247,8 @@ fields have strict primary-pass ownership—accuracy contributes only query accuracy, throughput contributes cost timing/resources, and latency contributes only insert latency. Only representable `(operation, pass)` pairs are accepted; unknown/contradictory primary fields, query/merge/prepare latency, and other -unrepresentable pairs fail loudly rather than being silently dropped. +unrepresentable pairs fail loudly rather than being silently dropped. A +representable pass must also contain its required primary result field. Reflattening the same preserved accuracy-first raw report now yields `atomic_costs_strict_accuracy_first.json` with one profile/two entries, so the From c2258f9c31c604e7d6ca28145e94f5a28872e9d5 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 9 Sep 2026 18:20:47 -0400 Subject: [PATCH 07/18] docs: define e2e profile validation protocol --- .design_docs/cost-optimizer-decision-map.md | 42 ++++++++++++++++++--- 1 file changed, 37 insertions(+), 5 deletions(-) diff --git a/.design_docs/cost-optimizer-decision-map.md b/.design_docs/cost-optimizer-decision-map.md index 9e38971d..a736974f 100644 --- a/.design_docs/cost-optimizer-decision-map.md +++ b/.design_docs/cost-optimizer-decision-map.md @@ -285,8 +285,35 @@ PromQL workload and series inventory? ### Answer -Open. The existing offline `asap-optimizer-cli` is the intended harness after -the document-loader/profile-selection gap is closed. +The existing offline `asap-optimizer-cli` is the selection harness; the +Hydra-based `asap-tools/experiments/experiment_run_e2e.py` is sufficient as +the execution harness. It already materializes a controller input from +`experiment_params`, passes `windowing` and `sketch_parameters` overrides, +and preserves its resolved Hydra config and controller/client output. It does +*not* read an atomic-cost profile or invoke the offline optimizer, so the +validation protocol must run the optimizer offline to choose/cost candidates, +then run the selected K values in E2E via `sketch_parameters.DatasketchesKLL.K`. + +Provisional validation experiment (not yet run): test whether the profile +predicts the relative end-to-end cost of `K=200` versus `K=500`, not yet an +exact-versus-approximate win. Use one Google `task_usage` part and one +`quantile_over_time(0.99, google_mean_cpu_usage_rate_0[3m])` workload, with +one-minute tumbling sketches so the query merges three windows. Train the +profile on a declared source-time training interval; use a disjoint source-time +holdout interval for E2E replay. Execute both forced K values in randomized +or alternating repeated trials. Compare the optimizer's predicted ordering +with observed query-engine CPU rate and steady-state memory; also report query +latency and accuracy against the Prometheus baseline as secondary outcomes. + +Before running, close the scenario-alignment gaps: the runtime Google exporter +exposes `job_id`, `task_index`, and `machine_id`, and metric suffix `_0` +filters `aggregation_type=0`. The sketch-bench workload/profile must use the +same complete series key and filter, not the current machine-only unfiltered +profile. The E2E exporter currently selects a part but exposes no source-time +window control in its Hydra configuration; add that filter (preferred), or +explicitly pre-slice the input, before calling the replay a temporal holdout. +Record the replay speed/arrival rate as well. Without those alignments, E2E +would be a useful smoke test but not validation of the measured profile. ## #3: What feasibility evidence constrains optimization? @@ -301,6 +328,11 @@ measured rather than assumed? ### Answer -Open. `sketch-bench` already retains capability-specific `query_accuracy`, but -the ASAPQuery greedy optimizer does not use it; arrival rate (`rho`) and exact -query cost are currently placeholders. +`sketch-bench` retains capability-specific `query_accuracy`, and the greedy +optimizer now enforces `query_accuracy.mean_rank_err` for KLL when +`max_mean_rank_error` is supplied. Arrival rate (`rho`) and exact-query cost +remain placeholders for an end-to-end comparison. After the K-versus-K +holdout, measure the actual raw/exact query path for the same scenario and +replace the temporary diagnostic exact baseline; then test whether the +optimizer selects exact or approximate according to observed cost under the +same accuracy limit. From cfc823ad7a4f988a3c536013d9137c1f3cb9c551 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 9 Sep 2026 18:25:18 -0400 Subject: [PATCH 08/18] docs: record google replay interval requirement --- .design_docs/cost-optimizer-decision-map.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/.design_docs/cost-optimizer-decision-map.md b/.design_docs/cost-optimizer-decision-map.md index a736974f..c1d07ff4 100644 --- a/.design_docs/cost-optimizer-decision-map.md +++ b/.design_docs/cost-optimizer-decision-map.md @@ -315,6 +315,13 @@ explicitly pre-slice the input, before calling the replay a temporal holdout. Record the replay speed/arrival rate as well. Without those alignments, E2E would be a useful smoke test but not validation of the measured profile. +2026-09-09 source-file check: Google `task_usage/part-00262-of-00500.csv.gz` +spans 5,265 source seconds (about 87.75 minutes), while the first profile used +only 180 seconds. Selecting `part_index: 262` in `experiment_run_e2e.py` +therefore does not by itself replay the same calibration data. Even the +same-data consistency experiment needs either exporter source-time bounds or +a materialized pre-sliced copy of that interval. + ## #3: What feasibility evidence constrains optimization? Blocked by: #1, #2 From 9ae8ddaa084d6b1e28a6ab6e437991a0db8793d3 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 9 Sep 2026 18:28:56 -0400 Subject: [PATCH 09/18] docs: record temporary derived replay input --- .design_docs/cost-optimizer-decision-map.md | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/.design_docs/cost-optimizer-decision-map.md b/.design_docs/cost-optimizer-decision-map.md index c1d07ff4..f73b62d2 100644 --- a/.design_docs/cost-optimizer-decision-map.md +++ b/.design_docs/cost-optimizer-decision-map.md @@ -322,6 +322,15 @@ therefore does not by itself replay the same calibration data. Even the same-data consistency experiment needs either exporter source-time bounds or a materialized pre-sliced copy of that interval. +2026-09-09 temporary same-data shortcut: materialized +`asap-tools/experiments/datasets/cost_optimizer_validation/google_task_usage_262_3m_agg0/part-00262-of-00500.csv.gz` +from that exact interval, retaining rows with missing/zero `aggregation_type` +to match the exporter’s `_0` convention. It has 20,051 rows, 6,481 machines, +and 10,379 distinct `(job_id, task_index, machine_id)` source series. Its +adjacent README records the predicate and SHA-256. This enables the first E2E +consistency run without changing the exporter; native source-time/filter +configuration remains the follow-up architectural work. + ## #3: What feasibility evidence constrains optimization? Blocked by: #1, #2 From d8593780d4d1f25cd0f0e3352cf1ed84b14a1e24 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 9 Sep 2026 21:58:14 -0400 Subject: [PATCH 10/18] experiment: add 1x KLL optimizer validation --- .design_docs/cost-optimizer-decision-map.md | 61 +++++++++++++++++++ .../src/google_metrics.rs | 5 +- .../cost_optimizer_validation_google.yaml | 57 +++++++++++++++++ .../google_task_usage_262_3m_agg0/README.md | 37 +++++++++++ .../README.md | 27 ++++++++ 5 files changed, 186 insertions(+), 1 deletion(-) create mode 100644 asap-tools/experiments/config/experiment_type/cost_optimizer_validation_google.yaml create mode 100644 asap-tools/experiments/datasets/cost_optimizer_validation/google_task_usage_262_3m_agg0/README.md create mode 100644 asap-tools/experiments/datasets/cost_optimizer_validation/google_task_usage_262_3m_agg0_rebased/README.md diff --git a/.design_docs/cost-optimizer-decision-map.md b/.design_docs/cost-optimizer-decision-map.md index f73b62d2..087a1748 100644 --- a/.design_docs/cost-optimizer-decision-map.md +++ b/.design_docs/cost-optimizer-decision-map.md @@ -331,6 +331,67 @@ adjacent README records the predicate and SHA-256. This enables the first E2E consistency run without changing the exporter; native source-time/filter configuration remains the follow-up architectural work. +2026-09-09 CloudLab execution preparation: copied that exact derived file to +`node1` (the cluster-data-exporter host) at +`/scratch/sketch_db_for_prometheus/experiment_inputs/cost_optimizer_validation/google_task_usage_262_3m_agg0/`. +The remote SHA-256 is `13dd00844558627365848f5252d9f86fe9c302afe875ae2fd702016b4acdf70f`, +matching the local README. The runner's `num_nodes=1` denotes one worker plus +the node-0 coordinator; its cluster-data exporter requires exactly this one +worker and runs on `node_offset + 1` (node1). Do not set `num_nodes=2`: the +service fails fast. The dedicated local Hydra scenario is +`cost_optimizer_validation_google`; it runs the 3-minute temporal quantile +with one-minute tumbling sketches and preserves K as a CLI override. + +2026-09-09 aligned atomic-cost profile: reran the KLL `k={200,500}` accuracy +and cost passes serially over the derived file with +`group_columns=[job_id, task_index, machine_id]`. The valid raw, flattened, +and document artifacts are in +`sketch-bench/output/cost_optimizer_experiments/google_task_usage_full_series/2026-09-09/final_serial/`. +The document has one profile/two entries and no skips. The earlier +`strict_rebuild/` artifacts are intentionally retained: they demonstrate that +the strict reducer rejected a duplicate entry caused by an overlapping local +benchmark invocation, rather than accepting ambiguous measurements. + +2026-09-09 E2E implementation and outcome: the temporary replay input was +also rebased (both Google timestamp columns) to the exporter's fixed 600-second +epoch. The rebased copy has the same 20,051 values/labels and SHA-256 +`744ce369e5dd638b800cc5eb202e8140a34166b3a8e301d583041ee640268d5e`; without +rebasing, its original 2011 timestamps would make the fixed-offset exporter +wait about fifteen days. The dedicated local `experiment_run_e2e.py` scenario +uses the full source-series labels `[instance, job, job_id, task_index, +machine_id]`, an 18-second sliding window and 1-second slide (the offline +optimizer's chosen K=200 geometry), and the runtime exporter at 1x rather than +the former 10x dilation. After CloudLab Docker access was enabled on both +nodes, the local runner completed both K=200 and K=500 runs in approximately +200 seconds each (190-second replay/warmup plus ten query repetitions), in +both SketchDB and Prometheus-baseline modes. The controller-resolved streaming +configs explicitly record K=200 and K=500 respectively. Raw client results, +latencies, resolved configs, controller logs, and monitor output are retained +under `experiment_outputs/cost_optimizer_validation_k{200,500}_1x_20260909/`. +Earlier retry directories are retained as failed preflight evidence (first +node1 Docker access, then node0 controller Docker access); the interrupted +10x run was superseded by the 1x runs. + +2026-09-09 first E2E reduction (same-data consistency, not holdout): SketchDB +completed ten queries in each run. Mean client latency was 169.25 ms for K=200 +and 170.61 ms for K=500; their independent Prometheus baseline runs averaged +114.54 ms and 112.12 ms. Query-engine monitor peak RSS was 2,754,297,856 B +(2.57 GiB) for K=200 versus 6,360,596,480 B (5.92 GiB) for K=500; mean sampled +query-engine CPU was 18.50% versus 23.43%. Thus this short run is consistent +with the predicted direction that K=200 is the cheaper feasible option in +memory/CPU, while latency is too close and too sparsely sampled to support a +latency-ordering claim. Each SketchDB result had a baseline result with the +same repetition and label set; baseline produced 251 (K=200) and 288 (K=500) +additional rows because the independently timed replays advanced through the +source stream at slightly different rates. On the matched rows the median and +95th-percentile absolute value differences were zero for both K values +(maximum 0.0091102). This is only a replay/result-path smoke check: values +from independently timed query executions do not establish KLL rank error. +The 2% eligibility claim remains supported by the paired sketch-bench profiles +(mean rank errors: K=200 0.158583%, K=500 0.07025%). A reproducible reduction +must retain the matched-key rule and report unmatched baseline rows rather than +silently treating them as zero error. + ## #3: What feasibility evidence constrains optimization? Blocked by: #1, #2 diff --git a/asap-tools/data-sources/prometheus-exporters/cluster_data_exporter/src/google_metrics.rs b/asap-tools/data-sources/prometheus-exporters/cluster_data_exporter/src/google_metrics.rs index 2fefed6a..ff421cd2 100644 --- a/asap-tools/data-sources/prometheus-exporters/cluster_data_exporter/src/google_metrics.rs +++ b/asap-tools/data-sources/prometheus-exporters/cluster_data_exporter/src/google_metrics.rs @@ -22,7 +22,10 @@ const CSV_MAX_PART_NO: u16 = 500; const MICRO_SECONDS_PER_SECOND: u64 = 1_000_000; const T_OFFSET_SECS: u64 = 600; -const DILATION_FACTOR: u64 = 10; // Factor for scaling time stamps relative to when they are exported +// Replay the materialized validation trace at its original rate. The source +// interval is only three minutes; dilating it would turn the E2E smoke test +// into a 30-minute run without changing its data or query semantics. +const DILATION_FACTOR: u64 = 1; /// Each line of the csv file is serialized into the following struct. /// The ordering of the struct fields MUST match the order that fields diff --git a/asap-tools/experiments/config/experiment_type/cost_optimizer_validation_google.yaml b/asap-tools/experiments/config/experiment_type/cost_optimizer_validation_google.yaml new file mode 100644 index 00000000..fe42d488 --- /dev/null +++ b/asap-tools/experiments/config/experiment_type/cost_optimizer_validation_google.yaml @@ -0,0 +1,57 @@ +# @package experiment_params +# +# Same-data integration validation for the cost-based KLL planner. +# +# The source is the materialized 3-minute, aggregation_type=0 subset described +# in datasets/cost_optimizer_validation/.../README.md. The runner's cluster +# exporter runs on node_offset + 1, so place that directory on node1 when the +# standard CloudLab invocation uses node_offset=0. + +experiment: + - mode: sketchdb + server: sketchdb + - mode: baseline + server: prometheus + +monitoring: + tool: prometheus + deployment_mode: bare_metal + +servers: + - name: prometheus + url: http://localhost:9090 + - name: sketchdb + url: http://localhost:8088 + +exporters: + only_start_if_queries_exist: true + exporter_list: + cluster_data_exporter: + provider: google + port: 40000 + metrics: mean-cpu-usage-rate + parts_mode: part-index + part_index: 262 + scrape_timeout: 1s + +query_groups: + - id: 1 + queries: + - quantile_over_time(0.99, google_mean_cpu_usage_rate_0[3m]) + repetition_delay_ms: 1000 + client_options: + repetitions: 10 + query_time_offset: 10 + # Wait for the complete 3m PromQL lookback before issuing measurement + # queries. The validation replay uses the trace's original 1x rate. + starting_delay: 190 + controller_options: + accuracy_sla: 0.98 + latency_sla: 1 + +metrics: + - metric: google_mean_cpu_usage_rate_0 + # instance/job are Prometheus scrape labels; the remaining labels are + # emitted by the Google exporter and identify a physical source series. + labels: [instance, job, job_id, task_index, machine_id] + exporter: cluster_data_exporter diff --git a/asap-tools/experiments/datasets/cost_optimizer_validation/google_task_usage_262_3m_agg0/README.md b/asap-tools/experiments/datasets/cost_optimizer_validation/google_task_usage_262_3m_agg0/README.md new file mode 100644 index 00000000..33772ae2 --- /dev/null +++ b/asap-tools/experiments/datasets/cost_optimizer_validation/google_task_usage_262_3m_agg0/README.md @@ -0,0 +1,37 @@ +# Derived Google task-usage input for cost-profile consistency validation + +This is a temporary, materialized input for the first same-data E2E +consistency experiment. It preserves the headerless Google `task_usage` CSV +shape expected by `cluster_data_exporter` and contains one correctly named +part file: + +`part-00262-of-00500.csv.gz` + +## Derivation + +Source: + +`../../../../benchmarks/metrics_observability/data/google-cluster-data/ClusterData2011/clusterdata-2011-2/task_usage/part-00262-of-00500.csv.gz` + +Selection, using raw trace microseconds and inclusive interval containment: + +```text +start_time >= 1313535000000 +end_time <= 1313715000000 +aggregation_type is absent or 0 +``` + +The absent-value rule matches the exporter, which treats a missing +`aggregation_type` as zero and exports it in the `_0` metric family. + +## Provenance + +- Rows: 20,051 +- Source-time interval: `[1313535000000, 1313715000000]` microseconds (180 s) +- Distinct machine IDs: 6,481 +- Distinct `(job_id, task_index, machine_id)` series: 10,379 +- Derived file SHA-256: `13dd00844558627365848f5252d9f86fe9c302afe875ae2fd702016b4acdf70f` + +This avoids adding source-time/filter options to the E2E exporter during the +integration check. Replace it with native filtering before treating the setup +as a reusable benchmark interface. diff --git a/asap-tools/experiments/datasets/cost_optimizer_validation/google_task_usage_262_3m_agg0_rebased/README.md b/asap-tools/experiments/datasets/cost_optimizer_validation/google_task_usage_262_3m_agg0_rebased/README.md new file mode 100644 index 00000000..2584769b --- /dev/null +++ b/asap-tools/experiments/datasets/cost_optimizer_validation/google_task_usage_262_3m_agg0_rebased/README.md @@ -0,0 +1,27 @@ +# Rebased Google task-usage replay input + +This is the executable replay counterpart of the adjacent +`google_task_usage_262_3m_agg0` calibration input. It retains exactly the +same 20,051 rows, values, and label columns. Only CSV columns 1 and 2 +(`start_time`, `end_time`, in microseconds) are transformed: + +``` +rebased_time_us = original_time_us - 1313535000000 + 600000000 +``` + +The Google exporter computes replay time as +`(start_time_us / 1_000_000 - 600) * 1` for this validation scenario. +Rebasing therefore emits the first sample immediately and retains the original +180-second relative schedule. Without this temporary input transformation, the +source interval's original time origin would delay first output by roughly 15 +days. + +The atomic-cost profile remains tied to the unmodified calibration input and +its original source-time window. This file is solely a replay-clock adapter; +it is not a new dataset or a substitute for native exporter time filtering. + +Validation: + +- rows: 20,051 +- rebased start-time range: 600000000–779000000 microseconds +- SHA-256: `744ce369e5dd638b800cc5eb202e8140a34166b3a8e301d583041ee640268d5e` From 90677c404af206959dfdbf17ac1caf7d0d2fc312 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 9 Sep 2026 21:59:09 -0400 Subject: [PATCH 11/18] experiment: add KLL validation result reducer --- .design_docs/cost-optimizer-decision-map.md | 5 + .../analyze_cost_optimizer_validation.py | 127 ++++++++++++++++++ 2 files changed, 132 insertions(+) create mode 100644 asap-tools/experiments/analyze_cost_optimizer_validation.py diff --git a/.design_docs/cost-optimizer-decision-map.md b/.design_docs/cost-optimizer-decision-map.md index 087a1748..bda8e26c 100644 --- a/.design_docs/cost-optimizer-decision-map.md +++ b/.design_docs/cost-optimizer-decision-map.md @@ -392,6 +392,11 @@ The 2% eligibility claim remains supported by the paired sketch-bench profiles must retain the matched-key rule and report unmatched baseline rows rather than silently treating them as zero error. +The reducer is `asap-tools/experiments/analyze_cost_optimizer_validation.py`. +It has been run over both retained directories and wrote +`experiment_outputs/cost_optimizer_validation_1x_summary.json`; it is the +reproducible source for the figures above. + ## #3: What feasibility evidence constrains optimization? Blocked by: #1, #2 diff --git a/asap-tools/experiments/analyze_cost_optimizer_validation.py b/asap-tools/experiments/analyze_cost_optimizer_validation.py new file mode 100644 index 00000000..28397a97 --- /dev/null +++ b/asap-tools/experiments/analyze_cost_optimizer_validation.py @@ -0,0 +1,127 @@ +#!/usr/bin/env python3 +"""Reduce the retained KLL cost-optimizer E2E experiment outputs. + +This is intentionally a result reducer, not an accuracy oracle. It compares +SketchDB and baseline values only where their independently timed replays have +the same `(repetition_idx, result_labels)` key. KLL rank-error feasibility is +reported from sketch-bench's atomic-cost profile, not inferred here. +""" + +import argparse +import gzip +import json +import math +import statistics +from pathlib import Path + + +def read_jsonl_gz(path): + with gzip.open(path, "rt") as source: + return [json.loads(line) for line in source] + + +def percentile(values, fraction): + values = sorted(values) + return values[math.ceil(fraction * len(values)) - 1] + + +def client_summary(output_dir): + client_dir = output_dir / "prometheus_client_output" + latencies = [row["latency"] for row in read_jsonl_gz(client_dir / "query_latencies.jsonl.gz")] + results = read_jsonl_gz(client_dir / "query_results.jsonl.gz") + return { + "latency_seconds": { + "count": len(latencies), + "mean": statistics.mean(latencies), + "median": statistics.median(latencies), + "min": min(latencies), + "max": max(latencies), + }, + "result_rows_by_repetition": { + str(repetition): sum( + row["repetition_idx"] == repetition for row in results + ) + for repetition in sorted({row["repetition_idx"] for row in results}) + }, + "results": results, + } + + +def result_map(rows): + return { + (row["repetition_idx"], row["result_labels"]): float(row["result_value"]) + for row in rows + } + + +def matched_value_summary(sketch_rows, baseline_rows): + sketch = result_map(sketch_rows) + baseline = result_map(baseline_rows) + common = sketch.keys() & baseline.keys() + absolute = [abs(sketch[key] - baseline[key]) for key in common] + relative = [ + abs(sketch[key] - baseline[key]) / abs(baseline[key]) + for key in common + if baseline[key] != 0 + ] + return { + "matched_rows": len(common), + "sketchdb_only_rows": len(sketch.keys() - baseline.keys()), + "baseline_only_rows": len(baseline.keys() - sketch.keys()), + "absolute_value_difference": { + "p50": percentile(absolute, 0.50), + "p95": percentile(absolute, 0.95), + "max": max(absolute), + }, + "note": "Value differences are a same-key replay smoke check, not KLL rank error.", + } + + +def query_engine_summary(output_dir): + monitor = json.loads( + (output_dir / "remote_monitor_output" / "monitor_output.json").read_text() + ) + engines = [entry for entry in monitor.values() if entry.get("keyword") == "sketchdb-queryengine-rust"] + if len(engines) != 1: + raise RuntimeError(f"expected exactly one SketchDB query engine, found {len(engines)}") + engine = engines[0] + memory = engine["memory_info"] + cpu = engine["cpu_percent"] + return { + "samples": len(memory), + "peak_rss_bytes": max(memory), + "mean_cpu_percent": statistics.mean(cpu), + "max_cpu_percent": max(cpu), + } + + +def summarize(run_dir): + sketchdb = client_summary(run_dir / "sketchdb") + baseline = client_summary(run_dir / "baseline") + return { + "run": run_dir.name, + "sketchdb": { + "client": {key: value for key, value in sketchdb.items() if key != "results"}, + "query_engine": query_engine_summary(run_dir / "sketchdb"), + }, + "baseline": { + "client": {key: value for key, value in baseline.items() if key != "results"}, + }, + "matched_value_check": matched_value_summary(sketchdb["results"], baseline["results"]), + } + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("run", type=Path, nargs="+", help="one or more E2E output directories") + parser.add_argument("--output", type=Path, help="write JSON summary to this path") + args = parser.parse_args() + document = {"runs": [summarize(run) for run in args.run]} + rendered = json.dumps(document, indent=2, sort_keys=True) + if args.output: + args.output.write_text(rendered + "\n") + print(rendered) + + +if __name__ == "__main__": + main() From 9a5d7ef965ea21653f98d9fd9923cd00b1f01a2e Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 9 Sep 2026 21:59:39 -0400 Subject: [PATCH 12/18] docs: record optimizer to E2E boundary --- .design_docs/cost-optimizer-decision-map.md | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/.design_docs/cost-optimizer-decision-map.md b/.design_docs/cost-optimizer-decision-map.md index bda8e26c..124a69f9 100644 --- a/.design_docs/cost-optimizer-decision-map.md +++ b/.design_docs/cost-optimizer-decision-map.md @@ -397,6 +397,20 @@ It has been run over both retained directories and wrote `experiment_outputs/cost_optimizer_validation_1x_summary.json`; it is the reproducible source for the figures above. +2026-09-09 optimizer-to-E2E wiring assessment: no new bridge is required for +this single-AQE KLL validation. `experiment_run_e2e.py` already forwards its +global `sketch_parameters` and `windowing` Hydra objects into the controller +input; overriding `DatasketchesKLL.K`, the window size, and the slide exactly +recreates this selected KLL deployment. The offline CLI is nevertheless not +general E2E wiring: it prints a human-readable `OptimizerSolution`, while E2E +then invokes the ordinary controller generator. A multi-AQE optimizer result +may contain distinct K values/window geometries and AQE-to-aggregation +assignments, none representable by those global overrides. The next genuine +product/plumbing milestone is a machine-readable deployment artifact from the +optimizer plus a controller/E2E input mode that consumes it. Do this only after +the paper's single-AQE evidence is stable; it is not a prerequisite for the +current controlled experiment. + ## #3: What feasibility evidence constrains optimization? Blocked by: #1, #2 From 19047747d3cc9bac2f11ffbbfb9d43134feec722 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 10 Sep 2026 09:45:35 -0400 Subject: [PATCH 13/18] feat(planner): enforce CPU and accuracy SLAs --- .design_docs/cost-optimizer-decision-map.md | 83 +++++++++++++++++++ asap-planner-rs/src/bin/candidate_gen_dump.rs | 5 +- asap-planner-rs/src/config/input.rs | 5 +- .../src/optimizer/aqe_extractor.rs | 14 ++++ asap-planner-rs/src/optimizer/atomic_costs.rs | 53 ++++++++++-- .../src/optimizer/candidate_gen.rs | 1 + asap-planner-rs/src/optimizer/constants.rs | 4 +- asap-planner-rs/src/optimizer/cost_model.rs | 64 +++++++------- asap-planner-rs/src/optimizer/dataset.rs | 7 ++ asap-planner-rs/src/optimizer/greedy.rs | 59 +++++++++---- asap-planner-rs/src/optimizer/pipeline.rs | 31 ++++++- asap-planner-rs/src/optimizer/solution.rs | 2 + 12 files changed, 267 insertions(+), 61 deletions(-) diff --git a/.design_docs/cost-optimizer-decision-map.md b/.design_docs/cost-optimizer-decision-map.md index 124a69f9..41591799 100644 --- a/.design_docs/cost-optimizer-decision-map.md +++ b/.design_docs/cost-optimizer-decision-map.md @@ -411,6 +411,89 @@ optimizer plus a controller/E2E input mode that consumes it. Do this only after the paper's single-AQE evidence is stable; it is not a prerequisite for the current controlled experiment. +2026-09-10 generalization decisions: retain `aggregation_type=0` and assume +no spatial filters/selectivity for the present study. Add synthetic metric +projections and additional query workloads; different cardinality projections +are desirable but not a required first expansion. Evaluate disjoint trace +intervals, but do not frame the immediate objective as prediction on unseen +data. Replace the exporter’s temporary replay-clock treatment with native time +range logic. Use the optimizer for broad scenario/query/SLA sweeps, reserving +E2E for a small number of selected-plan evidence runs rather than repeating +every optimizer simulation. + +2026-09-10 proposed architecture: introduce a dataset-wrangler module whose +single scenario specification materializes an exporter-ready dataset, complete +series inventory, and immutable manifest. The same manifest is passed to +sketch-bench and E2E, preventing label/time/filter drift. The benchmark export +is an `AtomicCostDocument` containing one `AtomicCostProfile` per exact +wrangled workload and measurement environment; each `AtomicCostEntry` is one +sketch/configuration’s insert, merge, query, memory, and accuracy observation. +Time range belongs in the profile/workload identity, not in a free-standing +entry. The optimizer should consume selected profiles plus an explicit backend +exact-cost profile and emit a machine-readable deployment plan, which E2E can +deploy without manually translating K/window overrides. + +2026-09-10 cost-model correction to make before claiming a calibrated +objective: expose a resource vector rather than immediately collapsing values +into one scalar: resident state memory (bytes), ingest CPU rate (CPU-s/s), +query CPU rate (CPU-s/s), and optional query working-memory constraint. Only +combine these with declared unit-bearing weights (for example, $/(byte-s) and +$/CPU-s), or state a multi-objective/constraint policy. Do not charge temporary +query memory as a per-query scalar without a measured lifetime. Keep KLL mean +rank error as the sole eligibility constraint for now. Measure the actual raw +Prometheus/exact query path rather than use a forced exact constant. + +2026-09-10 policy decision: minimize actual deployment cost subject to the +mean-rank-error SLA and a latency SLA. For the stated 16-vCPU, 21-GB instance +at $0.638/hour, one provisioned instance costs $459.36 per 30-day month. Do +not manufacture a per-workload dollar saving from lower CPU/RAM use when two +plans both fit on one fixed-price node: their standalone provisioned price is +identical. Instead estimate aggregate workload CPU demand and resident memory, +compute required instances as the maximum of the CPU- and memory-capacity +ceilings, and price that integer capacity. Savings arise when a lower-resource +plan permits more workload packing or one fewer instance. The latency SLA +requires a separately defined end-to-end latency estimate/measurement; atomic +CPU time alone is not a latency prediction. + +2026-09-10 first cost-model scope refinement: use CPU seconds only; defer +deployment-price/memory policy. For an unfiltered query, approximate query +read fanout by `N_G`, the number of distinct grouping-label tuples (not the +cardinality of individual labels unless there is exactly one grouping label). +For KLL, one grouping tuple has one KLL and `F=N_G`. For a CMS grouped by +`label_0` whose key is `(label_1,label_2)`, one CMS exists per `label_0` +value and `F=N_G=cardinality(label_0)`; key labels affect the measured CMS +profile/accuracy but do not multiply the number of CMS instances. Define +request CPU/latency conservatively as `F * (query_cpu + (windows_read-1) * +merge_cpu)` and optimize total CPU rate `ingest_cpu_rate + Σ frequency * +request_cpu`, subject to error and this atomic-CPU latency SLA. The current +planner violates this intended CMS rule: its `subpopulation_aware` branch +uses the hardcoded `SUBPOPULATION_COUNT=1`. Replace that placeholder with +explicit grouping-state count and query-fanout fields before adding CMS +experiments. + +2026-09-10 terminology decision: `N_G` is the cardinality of the **distinct +tuples** formed by the labels that partition/deploy sketches, calculated from +the wrangled series inventory. It is not the product of individual label +cardinalities and does not include key labels. A true global partition has +`N_G=1`; an ordinary temporal query whose planner preserves all source labels +uses the cardinality of the full exported-series tuple. Retire +`subpopulation_aware` from the public cost-model vocabulary unless the runtime +actually uses a distinct physical multi-group container model. Deployment +price is deferred. Do not use a burstable CPU-credit instance such as t2.nano +as the reference for a CPU-seconds/latency model: its sustained CPU capacity +is not represented by its one nominal vCPU. + +2026-09-10 implementation: `accuracy_sla` and `latency_sla` are now the sole +public optimizer feasibility inputs. The optimizer derives KLL's internal +mean-rank-error limit as `1 - accuracy_sla`, and derives an optional maximum +whole-request atomic CPU limit from positive `latency_sla`. Candidate request +CPU is rejected when it exceeds that limit. CPU-only selection is enabled by +zeroing the legacy memory weights. CMS now scales query work by its explicit +grouping-state count rather than the old hardcoded-one `subpopulation_aware` +branch; ingest CPU remains based on total input arrival rate and therefore +does not multiply by grouping count. Focused tests cover public-SLA conversion, +KLL accuracy conversion, and CMS query scaling. + ## #3: What feasibility evidence constrains optimization? Blocked by: #1, #2 diff --git a/asap-planner-rs/src/bin/candidate_gen_dump.rs b/asap-planner-rs/src/bin/candidate_gen_dump.rs index a9c5d616..d204b7c5 100644 --- a/asap-planner-rs/src/bin/candidate_gen_dump.rs +++ b/asap-planner-rs/src/bin/candidate_gen_dump.rs @@ -60,7 +60,10 @@ fn main() -> anyhow::Result<()> { qg.queries.iter().map(|q| RQE { query_string: q.clone(), t_repeat_ms: qg.repetition_delay_ms, - max_mean_rank_error: qg.controller_options.max_mean_rank_error, + max_mean_rank_error: (qg.controller_options.accuracy_sla > 0.0) + .then(|| 1.0 - qg.controller_options.accuracy_sla), + max_atomic_query_cpu_secs: (qg.controller_options.latency_sla > 0.0) + .then_some(qg.controller_options.latency_sla), }) }) .collect(); diff --git a/asap-planner-rs/src/config/input.rs b/asap-planner-rs/src/config/input.rs index 3c8979f0..0f67f8a1 100644 --- a/asap-planner-rs/src/config/input.rs +++ b/asap-planner-rs/src/config/input.rs @@ -80,11 +80,8 @@ pub struct QueryGroup { #[derive(Debug, Clone, Deserialize, Default)] pub struct ControllerOptions { pub accuracy_sla: f64, + /// Maximum atomic CPU seconds allowed to answer one request. pub latency_sla: f64, - /// KLL-specific feasibility constraint: the selected benchmark entry's - /// `mean_rank_err` must not exceed this fraction (0.02 = 2%). - #[serde(default)] - pub max_mean_rank_error: Option, } #[derive(Debug, Clone, Deserialize)] diff --git a/asap-planner-rs/src/optimizer/aqe_extractor.rs b/asap-planner-rs/src/optimizer/aqe_extractor.rs index 6c05d02f..d0f78bd2 100644 --- a/asap-planner-rs/src/optimizer/aqe_extractor.rs +++ b/asap-planner-rs/src/optimizer/aqe_extractor.rs @@ -18,6 +18,7 @@ pub struct RQE { pub query_string: String, pub t_repeat_ms: u64, pub max_mean_rank_error: Option, + pub max_atomic_query_cpu_secs: Option, } /// Stable deduplication key for an AQE. @@ -41,6 +42,7 @@ struct AQEAccumulator { min_t_repeat_ms: u64, t_repeat_gcd_ms: u64, max_mean_rank_error: Option, + max_atomic_query_cpu_secs: Option, } impl AQEKey { @@ -101,6 +103,7 @@ pub fn extract_aqes( min_t_repeat_ms: u64::MAX, t_repeat_gcd_ms: 0, max_mean_rank_error: rqe.max_mean_rank_error, + max_atomic_query_cpu_secs: rqe.max_atomic_query_cpu_secs, }); if !entry.query_strings.contains(&leaf) { entry.query_strings.push(leaf); @@ -121,6 +124,15 @@ pub fn extract_aqes( (None, Some(b)) => Some(b), (None, None) => None, }; + entry.max_atomic_query_cpu_secs = match ( + entry.max_atomic_query_cpu_secs, + rqe.max_atomic_query_cpu_secs, + ) { + (Some(a), Some(b)) => Some(a.min(b)), + (Some(a), None) => Some(a), + (None, Some(b)) => Some(b), + (None, None) => None, + }; } None => { warn!( @@ -140,6 +152,7 @@ pub fn extract_aqes( min_t_repeat_ms: accumulator.min_t_repeat_ms, t_repeat_gcd_ms: accumulator.t_repeat_gcd_ms, max_mean_rank_error: accumulator.max_mean_rank_error, + max_atomic_query_cpu_secs: accumulator.max_atomic_query_cpu_secs, }) .collect() } @@ -215,6 +228,7 @@ mod tests { query_string: query.to_string(), t_repeat_ms: t_ms, max_mean_rank_error: None, + max_atomic_query_cpu_secs: None, } } diff --git a/asap-planner-rs/src/optimizer/atomic_costs.rs b/asap-planner-rs/src/optimizer/atomic_costs.rs index 629b16d1..e99417d8 100644 --- a/asap-planner-rs/src/optimizer/atomic_costs.rs +++ b/asap-planner-rs/src/optimizer/atomic_costs.rs @@ -272,18 +272,21 @@ pub fn resolve_atomic_costs( }) } -/// Whether a benchmarked KLL candidate satisfies a requested maximum mean rank -/// error. Missing accuracy is infeasible: a constrained plan must not silently -/// substitute an unmeasured quality value. -pub fn satisfies_max_mean_rank_error( +/// Whether a benchmarked KLL candidate satisfies a requested minimum accuracy. +/// Accuracy is defined as `1 - mean_rank_err`. Missing or invalid measurements +/// are infeasible when the query declares an accuracy SLA. +pub fn satisfies_accuracy_sla( table: &AtomicCostTable, agg_type: AggregationType, params: &HashMap, - max_mean_rank_error: Option, + min_accuracy_sla: Option, ) -> bool { - let Some(limit) = max_mean_rank_error else { + let Some(min_accuracy) = min_accuracy_sla else { return true; }; + if !min_accuracy.is_finite() || !(0.0..=1.0).contains(&min_accuracy) { + return false; + } if agg_type == AggregationType::HydraKLL { return false; } @@ -298,7 +301,9 @@ pub fn satisfies_max_mean_rank_error( .iter() .find(|entry| entry.sketch == sketch && entry.sketch_config == expected_config) .and_then(|entry| entry.query_accuracy.get("mean_rank_err")) - .is_some_and(|error| error.is_finite() && *error <= limit) + .is_some_and(|error| { + error.is_finite() && (0.0..=1.0).contains(error) && 1.0 - error >= min_accuracy + }) } /// Temporary cost model for the runtime CMS-with-heap implementation. @@ -699,11 +704,41 @@ mod tests { fn hydra_kll_drops_when_an_empirical_profile_is_present() { let table = vec![cms_entry(3, 1024)]; assert!(resolve_atomic_costs(&table, AggregationType::HydraKLL, &HashMap::new()).is_none()); - assert!(!satisfies_max_mean_rank_error( + assert!(!satisfies_accuracy_sla( &table, AggregationType::HydraKLL, &HashMap::new(), - Some(0.02), + Some(0.98), + )); + } + + #[test] + fn kll_accuracy_sla_is_one_minus_mean_rank_error() { + let table = vec![AtomicCostEntry { + sketch: "kll-percall".into(), + sketch_config: serde_json::json!({ + "algorithm": "kll-percall", + "params": { "k": 200 } + }), + mem_bytes_per_instance: 6400.0, + insert_cpu_secs: 1.0e-8, + merge_cpu_secs: 1.0e-6, + query_cpu_secs: 1.0e-6, + query_accuracy: BTreeMap::from([("mean_rank_err".into(), 0.015)]), + }]; + let params = HashMap::from([("K".to_string(), Value::from(200u64))]); + + assert!(satisfies_accuracy_sla( + &table, + AggregationType::DatasketchesKLL, + ¶ms, + Some(0.98), + )); + assert!(!satisfies_accuracy_sla( + &table, + AggregationType::DatasketchesKLL, + ¶ms, + Some(0.99), )); } diff --git a/asap-planner-rs/src/optimizer/candidate_gen.rs b/asap-planner-rs/src/optimizer/candidate_gen.rs index 7f2eca9c..25971a5f 100644 --- a/asap-planner-rs/src/optimizer/candidate_gen.rs +++ b/asap-planner-rs/src/optimizer/candidate_gen.rs @@ -352,6 +352,7 @@ mod tests { min_t_repeat_ms: min_t, t_repeat_gcd_ms: min_t, max_mean_rank_error: None, + max_atomic_query_cpu_secs: None, } } diff --git a/asap-planner-rs/src/optimizer/constants.rs b/asap-planner-rs/src/optimizer/constants.rs index 5ca899d0..feeea39f 100644 --- a/asap-planner-rs/src/optimizer/constants.rs +++ b/asap-planner-rs/src/optimizer/constants.rs @@ -36,9 +36,9 @@ pub const EXACT_QUERY_CPU_SECS: f64 = 1e-3; // magnitude cheaper per unit than CPU-time (e.g. ~$5/GB-month vs // ~$0.04/vCPU-hour is roughly a 1e6 ratio), so memory weights are scaled // down accordingly rather than left equal to CPU weights. -pub const INGEST_MEM_WEIGHT: f64 = 1e-9; +pub const INGEST_MEM_WEIGHT: f64 = 0.0; pub const INGEST_CPU_WEIGHT: f64 = 1.0; -pub const QUERY_MEM_WEIGHT: f64 = 1e-9; +pub const QUERY_MEM_WEIGHT: f64 = 0.0; pub const QUERY_CPU_WEIGHT: f64 = 1.0; /// Subpopulation count: 1 if subpopulation_aware else the distinct label-group diff --git a/asap-planner-rs/src/optimizer/cost_model.rs b/asap-planner-rs/src/optimizer/cost_model.rs index 53a3ff02..5e0c0213 100644 --- a/asap-planner-rs/src/optimizer/cost_model.rs +++ b/asap-planner-rs/src/optimizer/cost_model.rs @@ -1,11 +1,10 @@ use asap_types::enums::WindowType; -use promql_utilities::query_logics::enums::AggregationType; use super::candidate_gen::CandidateConfig; use super::constants::{ EXACT_QUERY_CPU_SECS, INGEST_CPU_WEIGHT, INGEST_MEM_WEIGHT, INSERT_CPU_SECS, MEM_BYTES_PER_INSTANCE, MERGE_CPU_SECS, QUERY_CPU_SECS, QUERY_CPU_WEIGHT, QUERY_MEM_WEIGHT, - SUBPOPULATION_COUNT, SUBTRACT_CPU_SECS, + SUBTRACT_CPU_SECS, }; use super::sketch_properties::sketch_properties; use super::solution::{QueryMethod, AQE}; @@ -76,7 +75,7 @@ pub fn ingest_cost( return 0.0; // EXACT: no streaming config deployed. }; - let subpopulation_count = effective_subpopulation_count(candidate, agg_config.aggregation_type); + let subpopulation_count = effective_subpopulation_count(candidate); // Defensive floor: slide_interval_ms is a plain u64 on a widely-shared struct; // guard against div-by-zero producing `inf` and poisoning cost comparisons. @@ -109,7 +108,7 @@ pub fn query_cost( }; // Subpopulation count; see ingest_cost comment. - let subpopulation_count = effective_subpopulation_count(candidate, agg_config.aggregation_type); + let subpopulation_count = effective_subpopulation_count(candidate); let props = sketch_properties(agg_config.aggregation_type); let (cpu, mem) = match &candidate.query_method { @@ -142,21 +141,34 @@ pub fn query_cost( weights.query_cpu * cpu + weights.query_mem * mem } -fn effective_subpopulation_count( - candidate: &CandidateConfig, - aggregation_type: AggregationType, -) -> f64 { - if sketch_properties(aggregation_type).subpopulation_aware { - SUBPOPULATION_COUNT - } else { - assert!( - candidate.label_group_count > 0, - "non-subpopulation-aware candidates require a positive label_group_count" - ); - candidate.label_group_count as f64 +/// Unweighted CPU seconds required for one complete request. +pub fn atomic_query_cpu_secs(candidate: &CandidateConfig, costs: &AtomicCosts) -> f64 { + let Some(_) = &candidate.config else { + return costs.exact_query_cpu_secs; + }; + let groups = effective_subpopulation_count(candidate); + match &candidate.query_method { + QueryMethod::Direct => groups * costs.query_cpu_secs, + QueryMethod::Merge { num_windows } => { + groups + * (((*num_windows).saturating_sub(1) as f64) * costs.merge_cpu_secs + + costs.query_cpu_secs) + } + QueryMethod::Subtract => { + groups * (costs.merge_cpu_secs + costs.subtract_cpu_secs + costs.query_cpu_secs) + } + QueryMethod::Exact => unreachable!("Exact query_method must not have a config"), } } +fn effective_subpopulation_count(candidate: &CandidateConfig) -> f64 { + assert!( + candidate.label_group_count > 0, + "candidates require a positive label_group_count" + ); + candidate.label_group_count as f64 +} + /// Total cost rate contributed by assigning AQE `aqe` (with frequency /// `aqe.query_frequency_hz`) to `candidate`: IngestCost(g) + frequency * QueryCost(a,g). /// This is the per-(a,g) term the greedy/MIP solver minimizes. @@ -194,6 +206,7 @@ mod tests { min_t_repeat_ms: min_t, t_repeat_gcd_ms: min_t, max_mean_rank_error: None, + max_atomic_query_cpu_secs: None, } } @@ -296,16 +309,11 @@ mod tests { ..candidate }; let costs = AtomicCosts::default(); - let weights = CostWeights { - ingest_mem: 1.0, - ingest_cpu: 0.0, - query_mem: 1.0, - query_cpu: 0.0, - }; + let weights = CostWeights::default(); assert_eq!( ingest_cost(&five_groups, 1.0, &costs, &weights), - 5.0 * ingest_cost(&one_group, 1.0, &costs, &weights) + ingest_cost(&one_group, 1.0, &costs, &weights) ); assert_eq!( query_cost(&a, &five_groups, &costs, &weights), @@ -314,7 +322,7 @@ mod tests { } #[test] - fn subpopulation_aware_cost_ignores_label_group_count() { + fn cms_cost_scales_with_grouping_state_count() { let a = make_aqe(Statistic::Sum, 300_000, 300_000); let candidate = enumerate_candidates(&a, 60_000) .into_iter() @@ -336,12 +344,12 @@ mod tests { let weights = CostWeights::default(); assert_eq!( - ingest_cost(&one_group, 1.0, &costs, &weights), - ingest_cost(&five_groups, 1.0, &costs, &weights) + ingest_cost(&five_groups, 1.0, &costs, &weights), + ingest_cost(&one_group, 1.0, &costs, &weights) ); assert_eq!( - query_cost(&a, &one_group, &costs, &weights), - query_cost(&a, &five_groups, &costs, &weights) + query_cost(&a, &five_groups, &costs, &weights), + 5.0 * query_cost(&a, &one_group, &costs, &weights) ); } } diff --git a/asap-planner-rs/src/optimizer/dataset.rs b/asap-planner-rs/src/optimizer/dataset.rs index de1edc30..f2e288f1 100644 --- a/asap-planner-rs/src/optimizer/dataset.rs +++ b/asap-planner-rs/src/optimizer/dataset.rs @@ -19,6 +19,12 @@ const METRIC_COLUMN: &str = "metric"; #[derive(Debug, Error)] pub enum DatasetError { + #[error("controller_options.accuracy_sla must be finite and within [0, 1], got {0}")] + InvalidAccuracySla(f64), + #[error("controller_options.latency_sla must be finite and non-negative, got {0}")] + InvalidLatencySla(f64), + #[error("no feasible optimizer candidate for metric '{0}' under the declared SLAs")] + NoFeasibleCandidate(String), #[error("failed to open dataset '{path}': {source}")] Open { path: std::path::PathBuf, @@ -559,6 +565,7 @@ mod tests { min_t_repeat_ms: 1, t_repeat_gcd_ms: 1, max_mean_rank_error: None, + max_atomic_query_cpu_secs: None, }; assert!(matches!( diff --git a/asap-planner-rs/src/optimizer/greedy.rs b/asap-planner-rs/src/optimizer/greedy.rs index 0e8a0fb4..3696c1e0 100644 --- a/asap-planner-rs/src/optimizer/greedy.rs +++ b/asap-planner-rs/src/optimizer/greedy.rs @@ -2,9 +2,12 @@ use std::collections::HashMap; use tracing::debug; -use super::atomic_costs::{resolve_atomic_costs, satisfies_max_mean_rank_error, AtomicCostTable}; +use super::atomic_costs::{resolve_atomic_costs, satisfies_accuracy_sla, AtomicCostTable}; use super::candidate_gen::enumerate_candidates_with_label_group_count; -use super::cost_model::{ingest_cost, query_cost, total_cost_rate, AtomicCosts, CostWeights}; +use super::cost_model::{ + atomic_query_cpu_secs, ingest_cost, query_cost, total_cost_rate, AtomicCosts, CostWeights, +}; +use super::dataset::DatasetError; use super::dataset::ProfileKey; use super::solution::{AQEAssignment, OptimizerSolution, AQE}; @@ -29,7 +32,7 @@ pub fn greedy_assign( atomic_cost_table: &AtomicCostTable, weights: &CostWeights, label_group_counts: &HashMap, -) -> OptimizerSolution { +) -> Result { let mut solution = OptimizerSolution::empty(); for aqe in aqes { @@ -54,11 +57,11 @@ pub fn greedy_assign( let costs = match &c.config { None => AtomicCosts::default(), Some(cfg) => { - if !satisfies_max_mean_rank_error( + if !satisfies_accuracy_sla( atomic_cost_table, cfg.aggregation_type, &cfg.parameters, - aqe.max_mean_rank_error, + aqe.max_mean_rank_error.map(|error| 1.0 - error), ) { return None; } @@ -70,15 +73,19 @@ pub fn greedy_assign( } }; let cost = total_cost_rate(&aqe, &c, arrival_rate_hz, &costs, weights); + let request_cpu_secs = atomic_query_cpu_secs(&c, &costs); + if aqe + .max_atomic_query_cpu_secs + .is_some_and(|limit| !limit.is_finite() || request_cpu_secs > limit) + { + return None; + } Some((c, costs, cost)) }) // total_cmp (not partial_cmp().unwrap()) so a stray NaN cost can't panic. .min_by(|(_, _, a), (_, _, b)| a.total_cmp(b)) .map(|(c, costs, _)| (c, costs)) - .expect( - "enumerate_candidates always returns at least the EXACT fallback, \ - which always resolves (flat stub, no table lookup)", - ); + .ok_or_else(|| DatasetError::NoFeasibleCandidate(aqe.requirements.metric.clone()))?; let ingest = ingest_cost(&best, arrival_rate_hz, &costs, weights); let query_rate = aqe.query_frequency_hz * query_cost(&aqe, &best, &costs, weights); @@ -106,7 +113,7 @@ pub fn greedy_assign( }); } - solution + Ok(solution) } #[cfg(test)] @@ -135,6 +142,7 @@ mod tests { min_t_repeat_ms: min_t, t_repeat_gcd_ms: min_t, max_mean_rank_error: None, + max_atomic_query_cpu_secs: None, } } @@ -164,7 +172,8 @@ mod tests { 1, ), ]), - ); + ) + .unwrap(); let mut seen_ids: StdHashMap = StdHashMap::new(); for id in solution.deployed_configs().keys() { @@ -192,6 +201,7 @@ mod tests { min_t_repeat_ms: 60_000, t_repeat_gcd_ms: 60_000, max_mean_rank_error: None, + max_atomic_query_cpu_secs: None, }; let solution = greedy_assign( vec![aqe.clone()], @@ -200,7 +210,8 @@ mod tests { &AtomicCostTable::default(), &CostWeights::default(), &HashMap::from([(ProfileKey::from_requirements(&aqe.requirements), 1)]), - ); + ) + .unwrap(); assert_eq!(solution.num_exact_fallback(), 1); assert!(solution.deployed_configs().is_empty()); } @@ -217,7 +228,8 @@ mod tests { &AtomicCostTable::default(), &CostWeights::default(), &HashMap::from([(ProfileKey::from_requirements(&aqe.requirements), 1)]), - ); + ) + .unwrap(); assert_eq!(solution.num_exact_fallback(), 1); assert!(solution.deployed_configs().is_empty()); @@ -245,7 +257,8 @@ mod tests { &table, &CostWeights::default(), &HashMap::from([(ProfileKey::from_requirements(&aqe.requirements), 1)]), - ); + ) + .unwrap(); assert_eq!(solution.num_exact_fallback(), 0); assert_eq!(solution.deployed_configs().len(), 1); @@ -288,10 +301,26 @@ mod tests { &vec![kll(200, 0.03), kll(500, 0.01)], &CostWeights::default(), &HashMap::from([(ProfileKey::from_requirements(&aqe.requirements), 1)]), - ); + ) + .unwrap(); let config = solution.deployed_configs().values().next().unwrap(); assert_eq!(config.aggregation_type, AggregationType::DatasketchesKLL); assert_eq!(config.parameters["K"], serde_json::Value::from(500)); } + + #[test] + fn infeasible_atomic_cpu_sla_returns_an_error() { + let mut aqe = make_aqe(Statistic::Min, 60_000, 60_000, 1.0); + aqe.max_atomic_query_cpu_secs = Some(0.0); + let result = greedy_assign( + vec![aqe.clone()], + 60_000, + 1.0, + &AtomicCostTable::default(), + &CostWeights::default(), + &HashMap::from([(ProfileKey::from_requirements(&aqe.requirements), 1)]), + ); + assert!(matches!(result, Err(DatasetError::NoFeasibleCandidate(_)))); + } } diff --git a/asap-planner-rs/src/optimizer/pipeline.rs b/asap-planner-rs/src/optimizer/pipeline.rs index 5855ed0c..4b0c9d71 100644 --- a/asap-planner-rs/src/optimizer/pipeline.rs +++ b/asap-planner-rs/src/optimizer/pipeline.rs @@ -85,6 +85,19 @@ pub fn run_greedy_pipeline( atomic_cost_table: &AtomicCostTable, ) -> Result<(StreamingConfig, InferenceConfig), super::dataset::DatasetError> { dataset.validate_metric_hints(config.metrics.as_deref())?; + for group in &config.query_groups { + let options = &group.controller_options; + if !options.accuracy_sla.is_finite() || !(0.0..=1.0).contains(&options.accuracy_sla) { + return Err(super::dataset::DatasetError::InvalidAccuracySla( + options.accuracy_sla, + )); + } + if !options.latency_sla.is_finite() || options.latency_sla < 0.0 { + return Err(super::dataset::DatasetError::InvalidLatencySla( + options.latency_sla, + )); + } + } let schema = dataset.schema(); let rqes = config_to_rqes(config); let aqes = extract_aqes(&rqes, &schema, scrape_interval_ms); @@ -107,7 +120,7 @@ pub fn run_greedy_pipeline( atomic_cost_table, &CostWeights::default(), &label_group_counts, - ); + )?; Ok(finish_pipeline(solution, "greedy")) } @@ -122,7 +135,10 @@ fn config_to_rqes(config: &ControllerConfig) -> Vec { qg.queries.iter().map(|q| RQE { query_string: q.clone(), t_repeat_ms: qg.repetition_delay_ms, - max_mean_rank_error: qg.controller_options.max_mean_rank_error, + max_mean_rank_error: (qg.controller_options.accuracy_sla > 0.0) + .then(|| 1.0 - qg.controller_options.accuracy_sla), + max_atomic_query_cpu_secs: (qg.controller_options.latency_sla > 0.0) + .then_some(qg.controller_options.latency_sla), }) }) .collect() @@ -212,4 +228,15 @@ mod tests { assert_eq!(rqes[0].t_repeat_ms, 60_000); assert_eq!(rqes[1].t_repeat_ms, 30_000); } + + #[test] + fn config_to_rqes_maps_public_slas_to_atomic_constraints() { + let mut config = make_config(&[("quantile_over_time(0.99, metric[5m])", 60_000)]); + config.query_groups[0].controller_options.accuracy_sla = 0.98; + config.query_groups[0].controller_options.latency_sla = 0.001; + + let rqes = config_to_rqes(&config); + assert!((rqes[0].max_mean_rank_error.unwrap() - 0.02).abs() < f64::EPSILON); + assert_eq!(rqes[0].max_atomic_query_cpu_secs, Some(0.001)); + } } diff --git a/asap-planner-rs/src/optimizer/solution.rs b/asap-planner-rs/src/optimizer/solution.rs index be1f8400..08668066 100644 --- a/asap-planner-rs/src/optimizer/solution.rs +++ b/asap-planner-rs/src/optimizer/solution.rs @@ -38,6 +38,8 @@ pub struct AQE { /// Optional KLL feasibility constraint propagated from all query groups /// contributing to this AQE. Smaller means stricter. pub max_mean_rank_error: Option, + /// Maximum atomic CPU seconds allowed to answer one request. + pub max_atomic_query_cpu_secs: Option, } /// How an AQE is answered from its assigned streaming config. From d96bef37f032704ffe3d08c7d7ec99e8983324bb Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 10 Sep 2026 10:15:02 -0400 Subject: [PATCH 14/18] feat(experiments): add Google dataset wrangler --- .design_docs/cost-optimizer-decision-map.md | 11 ++ asap-tools/experiments/dataset_wrangler.py | 111 ++++++++++++++++++ .../google_task_usage_full_series.json | 7 ++ .../tests/test_dataset_wrangler.py | 69 +++++++++++ 4 files changed, 198 insertions(+) create mode 100644 asap-tools/experiments/dataset_wrangler.py create mode 100644 asap-tools/experiments/datasets/cost_optimizer_validation/google_task_usage_full_series.json create mode 100644 asap-tools/experiments/tests/test_dataset_wrangler.py diff --git a/.design_docs/cost-optimizer-decision-map.md b/.design_docs/cost-optimizer-decision-map.md index 41591799..acdb6030 100644 --- a/.design_docs/cost-optimizer-decision-map.md +++ b/.design_docs/cost-optimizer-decision-map.md @@ -433,6 +433,17 @@ entry. The optimizer should consume selected profiles plus an explicit backend exact-cost profile and emit a machine-readable deployment plan, which E2E can deploy without manually translating K/window overrides. +2026-09-10 implementation update: the first executable wrangler materializes +the Google task-usage validation scenario, applies the source interval and +`aggregation_type = 0` rule, validates duplicate exported samples, rebases the +replay timestamps, and writes the derived CSV, series inventory, and scenario +manifest. The manifest records both the compressed artifact hash and the +canonical decompressed-payload hash. The payload hash, not gzip metadata, is +the dataset identity that a future `AtomicCostProfile` must cite. Thus an +`AtomicCostDocument` remains a collection of measurements, not a dataset +container: `scenario spec -> wrangled CSV + ScenarioManifest -> sketch-bench +-> AtomicCostDocument`. + 2026-09-10 cost-model correction to make before claiming a calibrated objective: expose a resource vector rather than immediately collapsing values into one scalar: resident state memory (bytes), ingest CPU rate (CPU-s/s), diff --git a/asap-tools/experiments/dataset_wrangler.py b/asap-tools/experiments/dataset_wrangler.py new file mode 100644 index 00000000..f8f4690f --- /dev/null +++ b/asap-tools/experiments/dataset_wrangler.py @@ -0,0 +1,111 @@ +#!/usr/bin/env python3 +"""Materialize a reproducible Google task-usage experiment scenario.""" + +import argparse +import csv +import gzip +import hashlib +import io +import json +from pathlib import Path + + +START_TIME = 0 +END_TIME = 1 +JOB_ID = 2 +TASK_INDEX = 3 +MACHINE_ID = 4 +AGGREGATION_TYPE = 18 +REPLAY_OFFSET_US = 600_000_000 + + +def sha256(path): + digest = hashlib.sha256() + with path.open("rb") as source: + for block in iter(lambda: source.read(1024 * 1024), b""): + digest.update(block) + return digest.hexdigest() + + +def gzip_payload_sha256(path): + """Hash the decompressed CSV, so dataset identity ignores gzip metadata.""" + digest = hashlib.sha256() + with gzip.open(path, "rb") as source: + for block in iter(lambda: source.read(1024 * 1024), b""): + digest.update(block) + return digest.hexdigest() + + +def materialize(spec, spec_directory, output_directory): + source = (spec_directory / spec["source_file"]).resolve() + start_us, end_us = spec["source_time_range_us"] + labels = spec["grouping_labels"] + if labels != ["job_id", "task_index", "machine_id"]: + raise ValueError("Google v1 supports exactly job_id, task_index, machine_id grouping") + + output_directory.mkdir(parents=True, exist_ok=True) + output_data = output_directory / source.name + series = set() + seen_samples = set() + rows = 0 + with gzip.open(source, "rt", newline="") as input_file, gzip.GzipFile( + output_data, "wb", mtime=0 + ) as compressed_output, io.TextIOWrapper( + compressed_output, newline="" + ) as output_file: + reader = csv.reader(input_file) + writer = csv.writer(output_file, lineterminator="\n") + for row in reader: + if len(row) <= AGGREGATION_TYPE: + raise ValueError(f"malformed source row with {len(row)} columns") + row_start, row_end = int(row[START_TIME]), int(row[END_TIME]) + aggregation_type = row[AGGREGATION_TYPE] or "0" + if not (row_start >= start_us and row_end <= end_us and aggregation_type == "0"): + continue + group = (row[JOB_ID], row[TASK_INDEX], row[MACHINE_ID]) + sample_key = (row_start, group) + if sample_key in seen_samples: + raise ValueError(f"duplicate exported sample at {sample_key}") + seen_samples.add(sample_key) + if spec.get("rebase_to_offset", False): + row[START_TIME] = str(row_start - start_us + REPLAY_OFFSET_US) + row[END_TIME] = str(row_end - start_us + REPLAY_OFFSET_US) + writer.writerow(row) + series.add(group) + rows += 1 + + inventory = output_directory / "series_inventory.csv" + with inventory.open("w", newline="") as output_file: + writer = csv.writer(output_file) + writer.writerow(["metric", *labels]) + for group in sorted(series): + writer.writerow([spec["exported_metric"], *group]) + + manifest = { + "schema_version": 1, + "scenario_spec": spec, + "source_sha256": sha256(source), + "output_sha256": sha256(output_data), + "output_payload_sha256": gzip_payload_sha256(output_data), + "records_loaded": rows, + "grouping_state_count": len(series), + "arrival_rate_hz": rows / ((end_us - start_us) / 1_000_000), + "duplicate_policy": "fail", + } + (output_directory / "scenario_manifest.json").write_text( + json.dumps(manifest, indent=2) + "\n" + ) + return manifest + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--spec", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args() + spec = json.loads(args.spec.read_text()) + materialize(spec, args.spec.parent, args.output) + + +if __name__ == "__main__": + main() diff --git a/asap-tools/experiments/datasets/cost_optimizer_validation/google_task_usage_full_series.json b/asap-tools/experiments/datasets/cost_optimizer_validation/google_task_usage_full_series.json new file mode 100644 index 00000000..229cbb62 --- /dev/null +++ b/asap-tools/experiments/datasets/cost_optimizer_validation/google_task_usage_full_series.json @@ -0,0 +1,7 @@ +{ + "source_file": "../../../../../../benchmarks/metrics_observability/data/google-cluster-data/ClusterData2011/clusterdata-2011-2/task_usage/part-00262-of-00500.csv.gz", + "source_time_range_us": [1313535000000, 1313715000000], + "exported_metric": "google_mean_cpu_usage_rate_0", + "grouping_labels": ["job_id", "task_index", "machine_id"], + "rebase_to_offset": true +} diff --git a/asap-tools/experiments/tests/test_dataset_wrangler.py b/asap-tools/experiments/tests/test_dataset_wrangler.py new file mode 100644 index 00000000..5404347c --- /dev/null +++ b/asap-tools/experiments/tests/test_dataset_wrangler.py @@ -0,0 +1,69 @@ +import csv +import gzip +import importlib.util +import json +import tempfile +import unittest +from pathlib import Path + + +WRANGLER_PATH = Path(__file__).parents[1] / "dataset_wrangler.py" +SPEC = importlib.util.spec_from_file_location("dataset_wrangler", WRANGLER_PATH) +WRANGLER = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(WRANGLER) + + +def task_usage_row(start, end, job="job", task="task", machine="machine"): + row = [""] * 20 + row[0], row[1] = str(start), str(end) + row[2], row[3], row[4] = job, task, machine + row[18] = "0" + return row + + +class DatasetWranglerTest(unittest.TestCase): + def write_source(self, directory, rows): + source = directory / "source.csv.gz" + with gzip.open(source, "wt", newline="") as output: + writer = csv.writer(output, lineterminator="\n") + writer.writerows(rows) + return source + + def spec(self): + return { + "source_file": "source.csv.gz", + "source_time_range_us": [100, 200], + "exported_metric": "test_metric", + "grouping_labels": ["job_id", "task_index", "machine_id"], + "rebase_to_offset": True, + } + + def test_materializes_canonical_rebased_csv_and_manifest(self): + with tempfile.TemporaryDirectory() as temporary: + directory = Path(temporary) + self.write_source(directory, [task_usage_row(100, 110)]) + output = directory / "output" + + manifest = WRANGLER.materialize(self.spec(), directory, output) + + with gzip.open(output / "source.csv.gz", "rt", newline="") as result: + line = result.read() + self.assertTrue(line.startswith("600000000,600000010,job,task,machine")) + self.assertTrue(line.endswith("\n")) + self.assertNotIn("\r\n", line) + self.assertEqual(manifest["records_loaded"], 1) + self.assertEqual(manifest["grouping_state_count"], 1) + persisted = json.loads((output / "scenario_manifest.json").read_text()) + self.assertEqual(persisted["output_payload_sha256"], manifest["output_payload_sha256"]) + + def test_rejects_duplicate_exported_samples(self): + with tempfile.TemporaryDirectory() as temporary: + directory = Path(temporary) + self.write_source(directory, [task_usage_row(100, 110), task_usage_row(100, 111)]) + + with self.assertRaisesRegex(ValueError, "duplicate exported sample"): + WRANGLER.materialize(self.spec(), directory, directory / "output") + + +if __name__ == "__main__": + unittest.main() From 0e2cce35908f84be649803adf50e17c4bfd10ceb Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 10 Sep 2026 10:19:57 -0400 Subject: [PATCH 15/18] fix(planner): satisfy clippy SLA conversion --- asap-planner-rs/src/bin/candidate_gen_dump.rs | 2 +- asap-planner-rs/src/optimizer/pipeline.rs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/asap-planner-rs/src/bin/candidate_gen_dump.rs b/asap-planner-rs/src/bin/candidate_gen_dump.rs index d204b7c5..f140b2f4 100644 --- a/asap-planner-rs/src/bin/candidate_gen_dump.rs +++ b/asap-planner-rs/src/bin/candidate_gen_dump.rs @@ -61,7 +61,7 @@ fn main() -> anyhow::Result<()> { query_string: q.clone(), t_repeat_ms: qg.repetition_delay_ms, max_mean_rank_error: (qg.controller_options.accuracy_sla > 0.0) - .then(|| 1.0 - qg.controller_options.accuracy_sla), + .then_some(1.0 - qg.controller_options.accuracy_sla), max_atomic_query_cpu_secs: (qg.controller_options.latency_sla > 0.0) .then_some(qg.controller_options.latency_sla), }) diff --git a/asap-planner-rs/src/optimizer/pipeline.rs b/asap-planner-rs/src/optimizer/pipeline.rs index 4b0c9d71..acef62b1 100644 --- a/asap-planner-rs/src/optimizer/pipeline.rs +++ b/asap-planner-rs/src/optimizer/pipeline.rs @@ -136,7 +136,7 @@ fn config_to_rqes(config: &ControllerConfig) -> Vec { query_string: q.clone(), t_repeat_ms: qg.repetition_delay_ms, max_mean_rank_error: (qg.controller_options.accuracy_sla > 0.0) - .then(|| 1.0 - qg.controller_options.accuracy_sla), + .then_some(1.0 - qg.controller_options.accuracy_sla), max_atomic_query_cpu_secs: (qg.controller_options.latency_sla > 0.0) .then_some(qg.controller_options.latency_sla), }) From 7c94c2362f386742404eac9e41578ac4725b180a Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 10 Sep 2026 10:20:04 -0400 Subject: [PATCH 16/18] style(experiments): format dataset wrangler --- asap-tools/experiments/dataset_wrangler.py | 8 ++++++-- asap-tools/experiments/tests/test_dataset_wrangler.py | 8 ++++++-- 2 files changed, 12 insertions(+), 4 deletions(-) diff --git a/asap-tools/experiments/dataset_wrangler.py b/asap-tools/experiments/dataset_wrangler.py index f8f4690f..bf2f8144 100644 --- a/asap-tools/experiments/dataset_wrangler.py +++ b/asap-tools/experiments/dataset_wrangler.py @@ -41,7 +41,9 @@ def materialize(spec, spec_directory, output_directory): start_us, end_us = spec["source_time_range_us"] labels = spec["grouping_labels"] if labels != ["job_id", "task_index", "machine_id"]: - raise ValueError("Google v1 supports exactly job_id, task_index, machine_id grouping") + raise ValueError( + "Google v1 supports exactly job_id, task_index, machine_id grouping" + ) output_directory.mkdir(parents=True, exist_ok=True) output_data = output_directory / source.name @@ -60,7 +62,9 @@ def materialize(spec, spec_directory, output_directory): raise ValueError(f"malformed source row with {len(row)} columns") row_start, row_end = int(row[START_TIME]), int(row[END_TIME]) aggregation_type = row[AGGREGATION_TYPE] or "0" - if not (row_start >= start_us and row_end <= end_us and aggregation_type == "0"): + if not ( + row_start >= start_us and row_end <= end_us and aggregation_type == "0" + ): continue group = (row[JOB_ID], row[TASK_INDEX], row[MACHINE_ID]) sample_key = (row_start, group) diff --git a/asap-tools/experiments/tests/test_dataset_wrangler.py b/asap-tools/experiments/tests/test_dataset_wrangler.py index 5404347c..de521312 100644 --- a/asap-tools/experiments/tests/test_dataset_wrangler.py +++ b/asap-tools/experiments/tests/test_dataset_wrangler.py @@ -54,12 +54,16 @@ def test_materializes_canonical_rebased_csv_and_manifest(self): self.assertEqual(manifest["records_loaded"], 1) self.assertEqual(manifest["grouping_state_count"], 1) persisted = json.loads((output / "scenario_manifest.json").read_text()) - self.assertEqual(persisted["output_payload_sha256"], manifest["output_payload_sha256"]) + self.assertEqual( + persisted["output_payload_sha256"], manifest["output_payload_sha256"] + ) def test_rejects_duplicate_exported_samples(self): with tempfile.TemporaryDirectory() as temporary: directory = Path(temporary) - self.write_source(directory, [task_usage_row(100, 110), task_usage_row(100, 111)]) + self.write_source( + directory, [task_usage_row(100, 110), task_usage_row(100, 111)] + ) with self.assertRaisesRegex(ValueError, "duplicate exported sample"): WRANGLER.materialize(self.spec(), directory, directory / "output") From 8dabcd3b490bc162a749413885953151a3a953b4 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 10 Sep 2026 10:35:01 -0400 Subject: [PATCH 17/18] feat(planner): require scenario-matched cost profiles --- .design_docs/cost-optimizer-decision-map.md | 9 ++ asap-planner-rs/src/bin/candidate_gen_dump.rs | 9 +- asap-planner-rs/src/bin/optimizer_cli.rs | 13 +- asap-planner-rs/src/optimizer/atomic_costs.rs | 148 +++++++++++------- 4 files changed, 110 insertions(+), 69 deletions(-) diff --git a/.design_docs/cost-optimizer-decision-map.md b/.design_docs/cost-optimizer-decision-map.md index acdb6030..fbff92ad 100644 --- a/.design_docs/cost-optimizer-decision-map.md +++ b/.design_docs/cost-optimizer-decision-map.md @@ -444,6 +444,15 @@ the dataset identity that a future `AtomicCostProfile` must cite. Thus an container: `scenario spec -> wrangled CSV + ScenarioManifest -> sketch-bench -> AtomicCostDocument`. +2026-09-10 profile-provenance implementation: `AtomicCostDocument` schema v2 +adds a required profile-level `scenario` identity: the canonical wrangled CSV +payload hash, exported metric, grouping labels, and original source-time +range. `approxbench atomic-costs` now requires the wrangler's scenario +manifest and rejects malformed provenance. ASAPQuery's `--atomic-cost-profile` +selector must contain both `workload` and `scenario`; a workload-only match is +not eligible. This is the first enforced end-to-end guard against using a +benchmark profile measured on a different derived dataset. + 2026-09-10 cost-model correction to make before claiming a calibrated objective: expose a resource vector rather than immediately collapsing values into one scalar: resident state memory (bytes), ingest CPU rate (CPU-s/s), diff --git a/asap-planner-rs/src/bin/candidate_gen_dump.rs b/asap-planner-rs/src/bin/candidate_gen_dump.rs index f140b2f4..036b91a0 100644 --- a/asap-planner-rs/src/bin/candidate_gen_dump.rs +++ b/asap-planner-rs/src/bin/candidate_gen_dump.rs @@ -35,9 +35,10 @@ struct Args { #[arg(long = "atomic-costs")] atomic_costs: Option, - /// JSON `profiles[].workload` value selecting exactly one measured profile. - #[arg(long = "atomic-cost-workload", requires = "atomic_costs")] - atomic_cost_workload: Option, + /// JSON object containing the selected profile's `workload` and `scenario` + /// values, selecting exactly one measured profile. + #[arg(long = "atomic-cost-profile", requires = "atomic_costs")] + atomic_cost_profile: Option, } fn main() -> anyhow::Result<()> { @@ -45,7 +46,7 @@ fn main() -> anyhow::Result<()> { let atomic_cost_table = load_optional_selected_atomic_cost_table( args.atomic_costs.as_deref(), - args.atomic_cost_workload.as_deref(), + args.atomic_cost_profile.as_deref(), )? .unwrap_or_default(); diff --git a/asap-planner-rs/src/bin/optimizer_cli.rs b/asap-planner-rs/src/bin/optimizer_cli.rs index b6521b70..cbe16372 100644 --- a/asap-planner-rs/src/bin/optimizer_cli.rs +++ b/asap-planner-rs/src/bin/optimizer_cli.rs @@ -36,7 +36,7 @@ struct Args { rho: f64, /// Path to the versioned atomic-cost document sketch-bench's `atomic-costs` - /// subcommand exports. Requires --atomic-cost-workload to select exactly + /// subcommand exports. Requires --atomic-cost-profile to select exactly /// one measured workload profile. Omitted: every /// benchmarked-family candidate (CMS/HLL/KLL) is dropped, since there is /// no data to cost it at — only trivial accumulators and EXACT remain @@ -44,11 +44,10 @@ struct Args { #[arg(long = "atomic-costs")] atomic_costs: Option, - /// JSON `profiles[].workload` value copied from the sketch-bench atomic-cost - /// document. This makes the empirical workload profile explicit and avoids - /// mixing costs from different traces or time windows. - #[arg(long = "atomic-cost-workload", requires = "atomic_costs")] - atomic_cost_workload: Option, + /// JSON object containing the selected profile's `workload` and `scenario` + /// values. This prevents use of costs measured on a different wrangled CSV. + #[arg(long = "atomic-cost-profile", requires = "atomic_costs")] + atomic_cost_profile: Option, #[arg(short, long, action = clap::ArgAction::Count)] verbose: u8, @@ -79,7 +78,7 @@ fn main() -> anyhow::Result<()> { let atomic_cost_table = match load_optional_selected_atomic_cost_table( args.atomic_costs.as_deref(), - args.atomic_cost_workload.as_deref(), + args.atomic_cost_profile.as_deref(), )? { Some(table) => table, None => { diff --git a/asap-planner-rs/src/optimizer/atomic_costs.rs b/asap-planner-rs/src/optimizer/atomic_costs.rs index e99417d8..3aece6e0 100644 --- a/asap-planner-rs/src/optimizer/atomic_costs.rs +++ b/asap-planner-rs/src/optimizer/atomic_costs.rs @@ -22,26 +22,46 @@ use super::constants::{ use super::cost_model::AtomicCosts; const CMS_HEAP_BENCHMARK: &str = "cms-heap-topk-regularpath-vector2d"; -pub const ATOMIC_COST_SCHEMA_VERSION: u32 = 1; +pub const ATOMIC_COST_SCHEMA_VERSION: u32 = 2; /// Versioned atomic-cost document emitted by `approxbench atomic-costs`. /// /// A profile is deliberately selected before candidate resolution: costs from /// different input workloads must never be mixed by a flat lookup. -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct AtomicCostDocument { pub schema_version: u32, pub profiles: Vec, } -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct AtomicCostProfile { pub workload: WorkloadDescription, + pub scenario: ScenarioIdentity, pub entries: Vec, } +/// Immutable identity of the wrangled CSV that sketch-bench measured. +/// This must match exactly before the optimizer is allowed to use a profile. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ScenarioIdentity { + pub payload_sha256: String, + pub exported_metric: String, + pub grouping_labels: Vec, + pub source_time_range_us: [i64; 2], +} + +/// Exact profile identity selected for an optimizer run. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct AtomicCostProfileSelector { + pub workload: WorkloadDescription, + pub scenario: ScenarioIdentity, +} + /// The provenance of the input data on which atomic costs were measured. /// Synthetic descriptions remain opaque because their generator schema evolves /// independently; external profiles are represented explicitly for selection. @@ -72,7 +92,7 @@ pub struct ExternalWorkload { pub timestamp_unit: String, } -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct AtomicCostEntry { pub sketch: String, @@ -86,19 +106,19 @@ pub struct AtomicCostEntry { pub type AtomicCostTable = Vec; -/// Parse a standalone JSON workload selector. The selector is the exact -/// `profiles[].workload` value copied from the benchmark artifact, making the -/// selected empirical input explicit in an offline planning run. -pub fn load_workload_selector(path: &Path) -> anyhow::Result { +/// Parse a standalone JSON profile selector. The selector is the exact +/// `profiles[]` identity copied from the benchmark artifact, making both the +/// workload and the wrangled input explicit in an offline planning run. +pub fn load_profile_selector(path: &Path) -> anyhow::Result { let raw = std::fs::read_to_string(path).map_err(|e| { anyhow::anyhow!( - "reading atomic-cost workload selector {}: {e}", + "reading atomic-cost profile selector {}: {e}", path.display() ) })?; serde_json::from_str(&raw).map_err(|e| { anyhow::anyhow!( - "parsing atomic-cost workload selector {}: {e}", + "parsing atomic-cost profile selector {}: {e}", path.display() ) }) @@ -108,7 +128,7 @@ pub fn load_workload_selector(path: &Path) -> anyhow::Result anyhow::Result { let raw = std::fs::read_to_string(path) .map_err(|e| anyhow::anyhow!("reading atomic-cost table {}: {e}", path.display()))?; @@ -127,20 +147,22 @@ pub fn load_atomic_cost_table( let matches: Vec<_> = document .profiles .iter() - .filter(|profile| profile.workload == *workload) + .filter(|profile| { + profile.workload == selector.workload && profile.scenario == selector.scenario + }) .collect(); match matches.as_slice() { [profile] => Ok(profile.entries.clone()), [] => anyhow::bail!( - "no atomic-cost profile in {} matches workload selector {}", + "no atomic-cost profile in {} matches profile selector {}", path.display(), - serde_json::to_string(workload).unwrap_or_else(|_| "".into()) + serde_json::to_string(selector).unwrap_or_else(|_| "".into()) ), _ => anyhow::bail!( - "{} atomic-cost profiles in {} match workload selector {}; expected exactly one", + "{} atomic-cost profiles in {} match profile selector {}; expected exactly one", matches.len(), path.display(), - serde_json::to_string(workload).unwrap_or_else(|_| "".into()) + serde_json::to_string(selector).unwrap_or_else(|_| "".into()) ), } } @@ -152,8 +174,8 @@ pub fn load_selected_atomic_cost_table( document_path: &Path, selector_path: &Path, ) -> anyhow::Result { - let workload = load_workload_selector(selector_path)?; - load_atomic_cost_table(document_path, &workload) + let selector = load_profile_selector(selector_path)?; + load_atomic_cost_table(document_path, &selector) } /// Resolve the optional atomic-cost CLI inputs as one unit. A document without @@ -168,10 +190,10 @@ pub fn load_optional_selected_atomic_cost_table( load_selected_atomic_cost_table(document_path, selector_path).map(Some) } (Some(_), None) => anyhow::bail!( - "--atomic-cost-workload is required with --atomic-costs; \ - it must contain the selected profiles[].workload JSON value" + "--atomic-cost-profile is required with --atomic-costs; \ + it must contain the selected profiles[] workload and scenario JSON value" ), - (None, Some(_)) => anyhow::bail!("--atomic-cost-workload requires --atomic-costs"), + (None, Some(_)) => anyhow::bail!("--atomic-cost-profile requires --atomic-costs"), (None, None) => Ok(None), } } @@ -487,34 +509,40 @@ fn valid_cost_entry(entry: &AtomicCostEntry) -> bool { mod tests { use super::*; + fn scenario(payload_sha256: &str) -> ScenarioIdentity { + ScenarioIdentity { + payload_sha256: payload_sha256.into(), + exported_metric: "google_mean_cpu_usage_rate_0".into(), + grouping_labels: vec!["job_id".into(), "task_index".into(), "machine_id".into()], + source_time_range_us: [1_313_535_000_000, 1_313_715_000_000], + } + } + #[test] fn loader_selects_only_the_requested_external_profile() { - let requested = WorkloadDescription::External(ExternalWorkload { - source: "google".into(), - dataset: "google/task_usage.csv.gz".into(), - mode: "grouped".into(), - key_columns: vec![], - group_columns: vec!["machine_id".into()], - variate: None, - value_column: "cpu_rate".into(), - window_start_ns: 10, - window_end_ns: 20, - records_loaded: 100, - source_timestamp_unit: "microseconds".into(), - timestamp_unit: "nanoseconds".into(), - }); - let other = WorkloadDescription::External(ExternalWorkload { - window_end_ns: 30, - ..match requested.clone() { - WorkloadDescription::External(workload) => workload, - WorkloadDescription::Synthetic { .. } => unreachable!(), - } - }); + let requested = AtomicCostProfileSelector { + workload: WorkloadDescription::External(ExternalWorkload { + source: "google".into(), + dataset: "google/task_usage.csv.gz".into(), + mode: "grouped".into(), + key_columns: vec![], + group_columns: vec!["machine_id".into()], + variate: None, + value_column: "cpu_rate".into(), + window_start_ns: 10, + window_end_ns: 20, + records_loaded: 100, + source_timestamp_unit: "microseconds".into(), + timestamp_unit: "nanoseconds".into(), + }), + scenario: scenario("a"), + }; + let other_scenario = scenario("b"); let document = serde_json::json!({ - "schema_version": 1, + "schema_version": 2, "profiles": [ - {"workload": other, "entries": []}, - {"workload": requested, "entries": [{ + {"workload": requested.workload, "scenario": other_scenario, "entries": []}, + {"workload": requested.workload, "scenario": requested.scenario, "entries": [{ "sketch": "kll-percall", "sketch_config": {"algorithm": "kll-percall", "params": {"k": 200}}, "mem_bytes_per_instance": 6400.0, @@ -536,52 +564,56 @@ mod tests { #[test] fn loader_rejects_an_ambiguous_or_incompatible_document() { - let workload = WorkloadDescription::Synthetic { - description: serde_json::json!({"name": "one"}), + let selector = AtomicCostProfileSelector { + workload: WorkloadDescription::Synthetic { + description: serde_json::json!({"name": "one"}), + }, + scenario: scenario("a"), }; let file = tempfile::NamedTempFile::new().unwrap(); std::fs::write( file.path(), serde_json::json!({ - "schema_version": 2, + "schema_version": 3, "profiles": [] }) .to_string(), ) .unwrap(); - let err = load_atomic_cost_table(file.path(), &workload).unwrap_err(); - assert!(err.to_string().contains("schema_version 2")); - assert!(err.to_string().contains("supports 1")); + let err = load_atomic_cost_table(file.path(), &selector).unwrap_err(); + assert!(err.to_string().contains("schema_version 3")); + assert!(err.to_string().contains("supports 2")); std::fs::write( file.path(), serde_json::json!({ - "schema_version": 1, + "schema_version": 2, "profiles": [ - {"workload": workload, "entries": []}, - {"workload": workload, "entries": []} + {"workload": selector.workload, "scenario": selector.scenario, "entries": []}, + {"workload": selector.workload, "scenario": selector.scenario, "entries": []} ] }) .to_string(), ) .unwrap(); - let err = load_atomic_cost_table(file.path(), &workload).unwrap_err(); + let err = load_atomic_cost_table(file.path(), &selector).unwrap_err(); assert!(err.to_string().contains("2 atomic-cost profiles")); std::fs::write( file.path(), serde_json::json!({ - "schema_version": 1, + "schema_version": 2, "profiles": [{ "workload": {"synthetic": {"description": {"name": "other"}}}, + "scenario": selector.scenario, "entries": [] }] }) .to_string(), ) .unwrap(); - let err = load_atomic_cost_table(file.path(), &workload).unwrap_err(); + let err = load_atomic_cost_table(file.path(), &selector).unwrap_err(); assert!(err.to_string().contains("no atomic-cost profile")); } @@ -598,7 +630,7 @@ mod tests { load_optional_selected_atomic_cost_table(Some(document.path()), None).unwrap_err(); assert!(err .to_string() - .contains("--atomic-cost-workload is required")); + .contains("--atomic-cost-profile is required")); } fn cms_entry(depth: i64, width: i64) -> AtomicCostEntry { From 0471bf8a8617b2dd24d2ff25416b315f8e6b70dc Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 10 Sep 2026 10:42:53 -0400 Subject: [PATCH 18/18] docs: record provenance-gated optimizer run --- .design_docs/cost-optimizer-decision-map.md | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/.design_docs/cost-optimizer-decision-map.md b/.design_docs/cost-optimizer-decision-map.md index fbff92ad..2981bac6 100644 --- a/.design_docs/cost-optimizer-decision-map.md +++ b/.design_docs/cost-optimizer-decision-map.md @@ -453,6 +453,18 @@ selector must contain both `workload` and `scenario`; a workload-only match is not eligible. This is the first enforced end-to-end guard against using a benchmark profile measured on a different derived dataset. +2026-09-10 first provenance-gated run: regenerated the preserved Google KLL +grid as a schema-v2 document using the non-rebased calibration scenario +manifest. Its canonical payload hash is +`e6a533c06cab344c775b9b49b3672a2b976acbb4242fb05533ff0414b992a50e`; +the manifest payload exactly matches the CSV that was benchmarked. The +optimizer accepted the paired document and profile selector, then selected +EXACT for the 3-minute `quantile_over_time` workload under `accuracy_sla=0.98` +and `latency_sla=1`: with `N_G=10,379`, the CPU-only fanout estimate makes the +measured KLL query cost higher than the current exact-query constant. This is +an observed model result, not an error or a reason to restore the temporary +forced-exact-cost override. + 2026-09-10 cost-model correction to make before claiming a calibrated objective: expose a resource vector rather than immediately collapsing values into one scalar: resident state memory (bytes), ingest CPU rate (CPU-s/s),