From ed3ae273eddce9f59038d340fa940e6e42873cb2 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 8 Sep 2026 11:22:08 -0600 Subject: [PATCH 1/3] feat(planner): bind offline frequency evidence and preserve o11y exact fallbacks --- Cargo.lock | 6 +- control_plane/Cargo.toml | 6 +- control_plane/docs/offline-sketch-evidence.md | 86 ++++ .../examples/offline_frequency_plan.rs | 81 ++++ .../examples/offline_planner_replay.rs | 154 +++++++ control_plane/src/asap_tier_implement.rs | 2 +- control_plane/src/emit/mod.rs | 1 + control_plane/src/physical/allocator.rs | 7 +- .../src/physical/colored_dag/allocator.rs | 7 + .../src/physical/colored_dag/emitter.rs | 5 +- .../src/physical/colored_dag/tests.rs | 2 +- control_plane/src/physical/compiler.rs | 8 +- control_plane/src/physical/planner.rs | 3 +- .../src/physical/post_asap/cost_model.rs | 158 +++++++- control_plane/src/physical/post_asap/lower.rs | 12 + control_plane/src/physical/post_asap/tests.rs | 13 +- control_plane/src/query_parser/mod.rs | 4 +- control_plane/src/query_plan.rs | 3 + control_plane/src/workload.rs | 2 +- control_plane/tests/o11y_exact_fallback.rs | 42 ++ control_plane/tests/offline_evidence.rs | 376 ++++++++++++++++++ crates/asap_types/Cargo.toml | 2 +- data_plane/Cargo.toml | 2 +- .../asap_query_engine/post_asap_planner.rs | 8 +- .../asap_query_engine/summary_exec.rs | 71 +++- .../asap_query_engine/summary_executor.rs | 16 +- ...asapquery-compatibility-demo-snapshot.json | 2 +- .../examples/asapquery-planning-snapshot.json | 2 +- tools/run-offline-planner-replay.py | 51 +++ 29 files changed, 1092 insertions(+), 40 deletions(-) create mode 100644 control_plane/docs/offline-sketch-evidence.md create mode 100644 control_plane/examples/offline_frequency_plan.rs create mode 100644 control_plane/examples/offline_planner_replay.rs create mode 100644 control_plane/tests/o11y_exact_fallback.rs create mode 100644 control_plane/tests/offline_evidence.rs create mode 100644 tools/run-offline-planner-replay.py diff --git a/Cargo.lock b/Cargo.lock index 943595fa..76ab70fa 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -343,7 +343,7 @@ dependencies = [ [[package]] name = "asap-aware-mapping" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=cb50219c582d43f53ab77d3a595bd1ea4a9aa119#cb50219c582d43f53ab77d3a595bd1ea4a9aa119" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=dfdf6b5c1f7d667394a4ea1f56fb3786af04228d#dfdf6b5c1f7d667394a4ea1f56fb3786af04228d" dependencies = [ "asap-types", "serde", @@ -354,7 +354,7 @@ dependencies = [ [[package]] name = "asap-frontend-promql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=cb50219c582d43f53ab77d3a595bd1ea4a9aa119#cb50219c582d43f53ab77d3a595bd1ea4a9aa119" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=dfdf6b5c1f7d667394a4ea1f56fb3786af04228d#dfdf6b5c1f7d667394a4ea1f56fb3786af04228d" dependencies = [ "asap-types", "promql-parser 0.10.0", @@ -374,7 +374,7 @@ dependencies = [ [[package]] name = "asap-types" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=cb50219c582d43f53ab77d3a595bd1ea4a9aa119#cb50219c582d43f53ab77d3a595bd1ea4a9aa119" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=dfdf6b5c1f7d667394a4ea1f56fb3786af04228d#dfdf6b5c1f7d667394a4ea1f56fb3786af04228d" dependencies = [ "serde", "serde_json", diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml index 033c0d2a..041dde13 100644 --- a/control_plane/Cargo.toml +++ b/control_plane/Cargo.toml @@ -93,8 +93,8 @@ asap_types.workspace = true # scaffolding, unaware that `data_plane`'s `summary_executor.rs` in *this* # repo is a real one. Vendored locally instead of chased upstream -- see # `data_plane/src/query_engines/asap_query_engine/summary_exec.rs`. -planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "cb50219c582d43f53ab77d3a595bd1ea4a9aa119" } -asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "cb50219c582d43f53ab77d3a595bd1ea4a9aa119" } +planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "dfdf6b5c1f7d667394a4ea1f56fb3786af04228d" } +asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "dfdf6b5c1f7d667394a4ea1f56fb3786af04228d" } # L1 adoption (design-target-architecture.md Part B): the PromQL front # end itself, replacing control_plane's own query_parser/promql.rs. @@ -102,7 +102,7 @@ asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = # `planner-types`/`asap-aware-mapping` above -- these three MUST move # together (two revs of the same upstream repo's types in one workspace # resolve to distinct Rust types that won't unify). -asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "cb50219c582d43f53ab77d3a595bd1ea4a9aa119" } +asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "dfdf6b5c1f7d667394a4ea1f56fb3786af04228d" } [dev-dependencies] tokio = { version = "1", features = ["full", "test-util"] } diff --git a/control_plane/docs/offline-sketch-evidence.md b/control_plane/docs/offline-sketch-evidence.md new file mode 100644 index 00000000..a67df43b --- /dev/null +++ b/control_plane/docs/offline-sketch-evidence.md @@ -0,0 +1,86 @@ +# Offline sketch evidence replay + +Audience: developers evaluating planner integration without a deployed data plane. + +`ControlPlaneCostModel::with_offline_evidence` accepts the planner's validated +offline provider. Candidate ordering compares update CPU nanoseconds only when +every candidate has compatible evidence for the exact parameters returned by +this deployment's sizing policy. Missing, stale, incompatible, or ambiguous +measurements preserve the existing order. Offline errors never change formal +accuracy guarantees. Existing physical and lifecycle costs retain their units; +CPU nanoseconds are not added to legacy dimensionless costs. + +For explicit integer-key point-frequency queries, +`with_offline_frequency_comparison(evidence, request)` additionally compares +query-matched sketch measurements against an exact snapshot baseline. The +request supplies the observed mean-error budget, number of reads, retained +state count, horizon and CPU/memory weights. This is a fixed-snapshot offline +comparison: the caller asserts the recorded integer-key distribution and probe +population apply. The mean-error threshold applies to that recorded population, +not to each queried key or future live data. + +The backend restricts candidates to measured power-of-two CMS layouts before +comparison and supplies its own formal minimum parameters from the tighter +workload and query accuracy. Observed error may select a larger measured sketch, +but never weakens formal sizing. Missing evidence, an unacceptable error budget, +or an exact winner yields `PassThrough`, preserving exact execution. Unfiltered +legacy frequency totals and `count_over_time` never receive point-frequency +error acceptance. `offline_frequency_recommendation(payload)` exposes the same +decision and rejection reasons used by the binder. + +The typed frequency example reads real comparison artifacts without starting a +data plane, using the planner revision pinned in this backend: + +```bash +cargo run -p control_plane --example offline_frequency_plan -- \ + comparison-evidence.json comparison-request.json 7 0.01 +``` + +It binds a named integer-key source and reports the chosen parameters, exact +alternative, cost estimates and preserved point readout. It does not certify +that an existing Prometheus metric or deployed materialization uses that source. + +Run the PromQL replay using the pinned planner dependency: + +```bash +cargo run -p control_plane --example offline_planner_replay -- \ + o11y_bench_promql.txt planner-evidence.json context.json > control-plane-o11y.json +``` + +For development against unpublished planner changes, the optional local-checkout +wrapper supplies source patches and records the checkout revisions: + +```bash +python3 tools/run-offline-planner-replay.py \ + --planner /path/to/ASAPPlanner \ + --queries /path/to/ASAPPlanner/crates/frontend-promql/tests/observability/data/o11y_bench_promql.txt \ + --evidence /path/to/planner-evidence.json \ + --context /path/to/context.json \ + --output /tmp/control-plane-o11y.json +``` + +The wrapper supplies local Cargo source patches for all three planner crates, +preserving a single set of IR types. It also changes Cargo.lock; normal use of +the pinned dependency requires no source patches. The normal backend sibling +dependencies (ASAPCollector and asap_sketchlib) +must remain available at the paths in its workspace manifests. + +The replay calls the real control-plane parser and typed summary binder for +exact, default, and empirical modes. It records per-query rejection/fallback, +selected summary states, matching update/state evidence, provenance, and elapsed +planning time. The selected offline context is an explicit simulation assumption; +it is not an assertion that an o11y metric has that measured distribution. +Source scans beneath summaries count as raw subtrees, so they are not themselves +evidence that the whole query fell back. Root fallback is a separate field. +Bare selectors, sort roots and comparison/filter roots retain the complete +original query as `KeepPreAsap`. They bind successfully while preserving label +predicates, ordering and filtering. Executable query compilation marks these +roots `ExactFallback` and requests no summary materializations; successful +binding therefore does not mean they are served by the warm tier. + +This is binding coverage, not successful deployment compilation or execution. +No collectors or query servers start. Point-frequency benchmark errors are +exported as observations with `error_applies_to_current_query: false`. +Whole-plan resource savings remain null without matched raw/residual physical +operator evidence. Unsupported summary binary operators lower to explicit warm +tier fallback; they are not silently executed with different semantics. diff --git a/control_plane/examples/offline_frequency_plan.rs b/control_plane/examples/offline_frequency_plan.rs new file mode 100644 index 00000000..b30d5de1 --- /dev/null +++ b/control_plane/examples/offline_frequency_plan.rs @@ -0,0 +1,81 @@ +//! Bind a typed fixed-snapshot point-frequency query using offline evidence. +use std::rc::Rc; + +use anyhow::{bail, Result}; +use asap_aware_mapping::empirical_comparison::{ + OfflineComparisonEvidence, OfflineComparisonRequest, +}; +use control_plane::{ + physical::post_asap::{bind_query_expr_with_cost_model, cost_model::ControlPlaneCostModel}, + planner_selection::frequency, + types_v2::AccuracyTarget, +}; +use planner_types::pre_asap::{AggIntent, Column, DataType, QueryExpr, Reduction, Schema, Source}; +use serde_json::json; + +fn main() -> Result<()> { + let args: Vec<_> = std::env::args().skip(1).collect(); + if args.len() != 4 { + bail!("usage: offline_frequency_plan COMPARISON.json REQUEST.json INTEGER_KEY EPSILON"); + } + let evidence: OfflineComparisonEvidence = + serde_json::from_str(&std::fs::read_to_string(&args[0])?)?; + let request: OfflineComparisonRequest = + serde_json::from_str(&std::fs::read_to_string(&args[1])?)?; + let item: i64 = args[2].parse()?; + let epsilon: f64 = args[3].parse()?; + if !epsilon.is_finite() || epsilon <= 0.0 || epsilon >= 1.0 { + bail!("epsilon must be finite and strictly between zero and one"); + } + let accuracy = AccuracyTarget::Epsilon(epsilon); + let model = ControlPlaneCostModel::new(accuracy.clone()) + .with_offline_frequency_comparison(evidence, request.clone()); + let intent = frequency(accuracy, Some(("key".into(), item.to_string()))); + let AggIntent::Extension { payload, .. } = &intent else { + unreachable!() + }; + let recommendation = model.offline_frequency_recommendation(payload); + let query = QueryExpr::Aggregate { + reduction: Reduction::PerEntity, + measures: vec![intent], + output_names: vec![], + having: None, + child: Rc::new(QueryExpr::Scan { + source: Source::TimeSeries { + metric: "offline_integer_snapshot".into(), + }, + predicates: vec![], + schema: Schema::with_time_index( + vec![ + Column::new("ts", DataType::Timestamp, false), + Column::new("key", DataType::Int64, false), + Column::new("value", DataType::Float64, false), + ], + 0, + vec![], + ), + }), + }; + let bound = bind_query_expr_with_cost_model(&query, &model)?; + let root_raw_fallback = matches!(&bound, + control_plane::physical::post_asap::PhysicalExpr::Committed( + control_plane::physical::post_asap::PostAsapPlan::Summary(node) + ) if matches!(node.expr, planner_types::post_asap::SummaryExpr::KeepPreAsap(_))); + let (recommendation, unavailable_reason) = match recommendation { + Ok(value) => (Some(value), None), + Err(reason) => (None, Some(reason)), + }; + serde_json::to_writer_pretty( + std::io::stdout(), + &json!({ + "scope":"typed offline point-frequency recommendation and control-plane binding", + "request":request, "item":item, "formal_epsilon":epsilon, + "recommendation":recommendation,"unavailable_reason":unavailable_reason, + "bound_plan":format!("{bound:#?}"), "root_raw_fallback":root_raw_fallback, + "limitations":["The caller asserts the fixed-snapshot integer-key benchmark context", + "Observed mean error applies to the recorded offline probe population, not an individual key guarantee", + "No materializations deployed or data-plane execution performed"] + }), + )?; + Ok(()) +} diff --git a/control_plane/examples/offline_planner_replay.rs b/control_plane/examples/offline_planner_replay.rs new file mode 100644 index 00000000..a28a598a --- /dev/null +++ b/control_plane/examples/offline_planner_replay.rs @@ -0,0 +1,154 @@ +//! Offline control-plane binding, without starting collectors or query servers. +use std::{collections::HashSet, rc::Rc, time::Instant}; + +use anyhow::{bail, Context, Result}; +use asap_aware_mapping::empirical_cost::{ + EmpiricalEvidenceProvider, EvidenceArtifact, EvidenceContext, +}; +use control_plane::{ + physical::post_asap::{ + bind_query_expr_with_cost_model, cost_model::ControlPlaneCostModel, PhysicalExpr, + PostAsapPlan, + }, + query_parser::parse_query_expr_canonical, + types_v2::AccuracyTarget, +}; +use planner_types::post_asap::{SummaryExpr, SummaryFamilyType, SummaryNode}; +use serde_json::{json, Value}; + +fn inspect( + node: &Rc, + model: &ControlPlaneCostModel, + seen: &mut HashSet, + states: &mut Vec, + raw: &mut usize, +) { + if !seen.insert(Rc::as_ptr(node) as usize) { + return; + } + match &node.expr { + SummaryExpr::KeepPreAsap(_) => *raw += 1, + SummaryExpr::SummaryAgg { family, child, .. } => { + if let SummaryFamilyType::Sketch(kind, _) = family { + let lookup = model + .offline_evidence() + .map(|provider| provider.lookup(kind.algorithm(), kind.params())); + let (measurement, reason) = match lookup { + Some(Ok(row)) => ( + Some(json!({ + "record_id": row.id, + "provenance": row.provenance, + "update_cpu_ns": row.metrics.update_cpu_ns, + "retained_bytes": row.metrics.retained_bytes, + "offline_error_observation": row.error, + "error_applies_to_current_query": false, + })), + None, + ), + Some(Err(error)) => (None, Some(error.to_string())), + None => (None, Some("offline evidence not supplied".into())), + }; + states.push(json!({"algorithm":kind.algorithm(), "params":kind.params(), "measurement":measurement, "unavailable_reason":reason})); + } else { + states.push(json!({"exact_family":format!("{family:?}")})); + } + inspect(child, model, seen, states, raw); + } + SummaryExpr::SummaryEstimate { summary_input, .. } + | SummaryExpr::SummaryDelete { summary_input, .. } => { + inspect(summary_input, model, seen, states, raw) + } + SummaryExpr::SummaryMerge { children } => { + for child in children { + inspect(child, model, seen, states, raw); + } + } + SummaryExpr::SummaryJoin { + outer: lhs, + inner: rhs, + .. + } + | SummaryExpr::SummarySubtract { + left: lhs, + right: rhs, + } + | SummaryExpr::BinaryOp { lhs, rhs, .. } => { + inspect(lhs, model, seen, states, raw); + inspect(rhs, model, seen, states, raw); + } + } +} + +fn main() -> Result<()> { + let args: Vec<_> = std::env::args().skip(1).collect(); + if args.len() != 1 && args.len() != 3 { + bail!("usage: offline_planner_replay QUERIES.txt [EVIDENCE.json CONTEXT.json]"); + } + let corpus = std::fs::read_to_string(&args[0])?; + let evidence = if args.len() == 3 { + let artifact: EvidenceArtifact = serde_json::from_str(&std::fs::read_to_string(&args[1])?)?; + let context: EvidenceContext = serde_json::from_str(&std::fs::read_to_string(&args[2])?)?; + Some((artifact, context)) + } else { + None + }; + let mut rows = Vec::new(); + let mut modes = vec!["exact", "default"]; + if evidence.is_some() { + modes.push("empirical"); + } + for mode in modes { + let accuracy = if mode == "exact" { + AccuracyTarget::Exact + } else { + AccuracyTarget::Epsilon(0.01) + }; + let mut model = ControlPlaneCostModel::new(accuracy.clone()); + if mode == "empirical" { + let (artifact, context) = evidence.as_ref().context("missing evidence")?; + model = model.with_offline_evidence(EmpiricalEvidenceProvider::new( + artifact.clone(), + context.clone(), + )?); + } + for query in corpus + .lines() + .map(str::trim) + .filter(|q| !q.is_empty() && !q.starts_with('#')) + { + let start = Instant::now(); + let result = parse_query_expr_canonical(query, accuracy.clone()).and_then(|expr| { + bind_query_expr_with_cost_model(&expr, &model).map_err(Into::into) + }); + let elapsed_ns = start.elapsed().as_nanos(); + let result = match result { + Ok(PhysicalExpr::Committed(PostAsapPlan::Summary(node))) => { + let mut states = Vec::new(); + let mut raw = 0; + inspect(&node, &model, &mut HashSet::new(), &mut states, &mut raw); + json!({"status":"bound", "states":states, "raw_subtrees":raw, + "root_raw_fallback":matches!(node.expr, SummaryExpr::KeepPreAsap(_)), + "summary_plan":format!("{node:#?}")}) + } + Ok(plan) => json!({"status":"other_physical_plan", "plan":format!("{plan:?}")}), + Err(error) => json!({"status":"rejected", "reason":error.to_string()}), + }; + rows.push( + json!({"query":query,"mode":mode,"planning_elapsed_ns":elapsed_ns,"result":result}), + ); + } + } + serde_json::to_writer_pretty( + std::io::stdout(), + &json!({ + "schema_version":1, + "evaluation":"offline control-plane parser and typed summary binder", + "corpus_path":args[0], + "offline_context":evidence.as_ref().map(|(_, context)|context), + "limitations":["No deployed execution or measured end-to-end speedup", "Binding does not establish executable placement or complete physical cost", "Offline point-frequency errors do not establish current query guarantees", "raw_subtrees includes necessary source scans beneath summaries"], + "estimated_end_to_end_savings":null, + "rows":rows + }), + )?; + Ok(()) +} diff --git a/control_plane/src/asap_tier_implement.rs b/control_plane/src/asap_tier_implement.rs index 627fc91d..02809866 100644 --- a/control_plane/src/asap_tier_implement.rs +++ b/control_plane/src/asap_tier_implement.rs @@ -134,7 +134,7 @@ fn collect_aggregate_roots<'a>(expr: &'a QueryExpr, out: &mut Vec<&'a QueryExpr> | QueryExpr::Sort { child, .. } | QueryExpr::Limit { child, .. } | QueryExpr::PromqlSubquery { child, .. } => collect_aggregate_roots(child, out), - QueryExpr::Concat { children } => { + QueryExpr::Concat { children, .. } => { for c in children { collect_aggregate_roots(c, out); } diff --git a/control_plane/src/emit/mod.rs b/control_plane/src/emit/mod.rs index b4731ae1..6d9553b6 100644 --- a/control_plane/src/emit/mod.rs +++ b/control_plane/src/emit/mod.rs @@ -322,6 +322,7 @@ fn extract_from_node(node: &Rc) -> Option { // Not surfaced by any `Bind*` path yet (gated on rules that // haven't landed — see `deployment_expr.rs`'s module docs). SummaryExpr::SummaryJoin { .. } + | SummaryExpr::BinaryOp { .. } | SummaryExpr::SummarySubtract { .. } | SummaryExpr::SummaryDelete { .. } | SummaryExpr::KeepPreAsap(_) => None, diff --git a/control_plane/src/physical/allocator.rs b/control_plane/src/physical/allocator.rs index 31e3fa80..a5055da8 100644 --- a/control_plane/src/physical/allocator.rs +++ b/control_plane/src/physical/allocator.rs @@ -263,7 +263,10 @@ impl SketchAllocator { } // ── Merge — Backend ─────────────────────────────────────────── - QueryExpr::Concat { children: inputs } => { + QueryExpr::Concat { + children: inputs, + discriminator_unique_key, + } => { let children: Vec = inputs .into_iter() .map(|inp| self.alloc_node(inp, budget)) @@ -272,6 +275,7 @@ impl SketchAllocator { PlanNode { expr: QueryExpr::Concat { children: children.iter().map(|c| c.expr.clone()).collect(), + discriminator_unique_key, }, stage: PipelineStage::Backend, mode: ExecutionMode::Passthrough, @@ -952,6 +956,7 @@ mod tests { fn merge_goes_to_backend() { let expr = QueryExpr::Concat { children: vec![scan("a"), scan("b")], + discriminator_unique_key: None, }; let node = alloc(unlimited(), expr); assert_eq!(node.stage, PipelineStage::Backend); diff --git a/control_plane/src/physical/colored_dag/allocator.rs b/control_plane/src/physical/colored_dag/allocator.rs index ae260412..fe5487e0 100644 --- a/control_plane/src/physical/colored_dag/allocator.rs +++ b/control_plane/src/physical/colored_dag/allocator.rs @@ -177,6 +177,13 @@ impl ThreeStageWalker { // (scrape locality); `Ref` resolves through the lexical // scope map. SummaryExpr::KeepPreAsap(qe) => self.colour_logical(qe)?, + SummaryExpr::BinaryOp { lhs, rhs, .. } => { + let (left, _) = self.visit_l4node(lhs)?; + let (right, _) = self.visit_l4node(rhs)?; + self.dag.edges.push((id, left)); + self.dag.edges.push((id, right)); + StageId::Backend + } // ── SummaryAgg: always edge per design.md §6 batched-queries // table — true for both approximate sketches (the old diff --git a/control_plane/src/physical/colored_dag/emitter.rs b/control_plane/src/physical/colored_dag/emitter.rs index 026e1c7a..bf29ef56 100644 --- a/control_plane/src/physical/colored_dag/emitter.rs +++ b/control_plane/src/physical/colored_dag/emitter.rs @@ -115,7 +115,8 @@ fn classify(expr: &PhysicalExpr) -> NodeKind<'_> { SummaryExpr::SummaryMerge { .. } => NodeKind::SketchMerge, SummaryExpr::SummaryJoin { .. } | SummaryExpr::SummarySubtract { .. } - | SummaryExpr::SummaryDelete { .. } => NodeKind::Other, + | SummaryExpr::SummaryDelete { .. } + | SummaryExpr::BinaryOp { .. } => NodeKind::Other, }, PhysicalExpr::Committed(PostAsapPlan::LetBinding { name, .. }) => { NodeKind::LetBinding { name } @@ -1125,7 +1126,7 @@ fn extract_edge_facts(qe: &planner_types::pre_asap::QueryExpr, edge: &mut EdgeSt | QE::Sort { child, .. } | QE::Limit { child, .. } | QE::PromqlSubquery { child, .. } => extract_edge_facts(child, edge), - QE::Concat { children } => { + QE::Concat { children, .. } => { for c in children { extract_edge_facts(c, edge); } diff --git a/control_plane/src/physical/colored_dag/tests.rs b/control_plane/src/physical/colored_dag/tests.rs index 8257711d..49948f15 100644 --- a/control_plane/src/physical/colored_dag/tests.rs +++ b/control_plane/src/physical/colored_dag/tests.rs @@ -109,7 +109,7 @@ fn sketch_agg_l4( SketchKind::new(kind, params), GroupingStrategy::default(), ), - col: ColumnRef::SampleValue, + input: planner_types::post_asap::SummaryUpdate::column(ColumnRef::SampleValue), reduction: Reduction::by(vec![]), grouping: GroupingStrategy::default(), }, diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index dce55179..9061b2af 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -1975,7 +1975,12 @@ fn summary_agg_metric(node: &SummaryNode) -> Option { inner: right, .. } - | SummaryExpr::SummarySubtract { left, right } => { + | SummaryExpr::SummarySubtract { left, right } + | SummaryExpr::BinaryOp { + lhs: left, + rhs: right, + .. + } => { walk(left, metrics); walk(right, metrics); } @@ -2338,6 +2343,7 @@ fn collect_selected_materializations( }); } SummaryExpr::KeepPreAsap(_) + | SummaryExpr::BinaryOp { .. } | SummaryExpr::SummaryAgg { .. } | SummaryExpr::SummaryJoin { .. } | SummaryExpr::SummarySubtract { .. } diff --git a/control_plane/src/physical/planner.rs b/control_plane/src/physical/planner.rs index ff0532ca..9cd263bd 100644 --- a/control_plane/src/physical/planner.rs +++ b/control_plane/src/physical/planner.rs @@ -351,7 +351,7 @@ fn plan_node(expr: &QueryExpr, config: &PhysicalPlannerConfig) -> PhysicalNode { // (`intent_algebra::lower`), so the `HashAggregate { keys }` this // arm used to build now comes straight out of the `Aggregate` // arm above. - QueryExpr::Concat { children } => { + QueryExpr::Concat { children, .. } => { let children: Vec = children.iter().map(|c| plan_node(c, config)).collect(); let sketch_type = children @@ -774,6 +774,7 @@ mod tests { having: None, child: QueryExpr::Concat { children: vec![windowed_agg(default_frequency(), 60, "requests")], + discriminator_unique_key: None, } .into(), }; diff --git a/control_plane/src/physical/post_asap/cost_model.rs b/control_plane/src/physical/post_asap/cost_model.rs index 244322bf..0f753a1c 100644 --- a/control_plane/src/physical/post_asap/cost_model.rs +++ b/control_plane/src/physical/post_asap/cost_model.rs @@ -36,6 +36,11 @@ #![allow(dead_code)] use asap_aware_mapping::cost_model::{Cost, CostedSummaryDeployment}; +use asap_aware_mapping::empirical_comparison::{ + recommend_offline, OfflineComparisonEvidence, OfflineComparisonRequest, OfflineRecommendation, + SketchConfiguration, +}; +use asap_aware_mapping::empirical_cost::EmpiricalEvidenceProvider; use asap_aware_mapping::{ CompleteSummaryCandidateEstimate, CostModel, Horizon, Implementation, SummaryMaintenanceCapabilities, SummaryMaintenanceLifecycleCostInputs, @@ -59,6 +64,8 @@ pub struct ControlPlaneCostModel { lifecycle_costs: SummaryMaintenanceLifecycleCostInputs, summary_maintenance: SummaryMaintenanceCapabilities, window_framework_costs: Vec<(SummaryWindowFramework, Cost)>, + offline_evidence: Option, + offline_frequency_comparison: Option<(OfflineComparisonEvidence, OfflineComparisonRequest)>, } impl ControlPlaneCostModel { @@ -68,9 +75,141 @@ impl ControlPlaneCostModel { lifecycle_costs: SummaryMaintenanceLifecycleCostInputs::default(), summary_maintenance: SummaryMaintenanceCapabilities::default(), window_framework_costs: Vec::new(), + offline_evidence: None, + offline_frequency_comparison: None, } } + /// Use offline update CPU evidence for algorithm ordering. Physical costs + /// and accuracy guarantees retain their deployment-specific contracts. + pub fn with_offline_evidence(mut self, evidence: EmpiricalEvidenceProvider) -> Self { + self.offline_evidence = Some(evidence); + self + } + + pub fn offline_evidence(&self) -> Option<&EmpiricalEvidenceProvider> { + self.offline_evidence.as_ref() + } + + /// Opt into a fixed-snapshot frequency comparison. The caller asserts that + /// the explicit point queries use the supplied integer-key distribution and + /// offline probe population. The observed mean is an acceptance criterion + /// over that population, not a per-key or real-time error guarantee. + pub fn with_offline_frequency_comparison( + mut self, + evidence: OfflineComparisonEvidence, + request: OfflineComparisonRequest, + ) -> Self { + self.offline_frequency_comparison = Some((evidence, request)); + self + } + + /// Explain the same recommendation used by the frequency extension binder. + /// Exact selection and any unavailable comparison preserve exact execution. + pub fn offline_frequency_recommendation( + &self, + payload: &serde_json::Value, + ) -> Result { + let (evidence, request) = self + .offline_frequency_comparison + .as_ref() + .ok_or("offline frequency comparison is not configured")?; + if payload + .get("item_label") + .and_then(|v| v.as_str()) + .is_none_or(str::is_empty) + || payload + .get("item_value") + .and_then(|v| v.as_str()) + .and_then(|v| v.parse::().ok()) + .is_none() + { + return Err( + "offline point-frequency comparison requires an explicit integer item readout" + .into(), + ); + } + let accuracy: AccuracyTarget = serde_json::from_value( + payload + .get("accuracy") + .cloned() + .ok_or("missing frequency accuracy")?, + ) + .map_err(|error| error.to_string())?; + let (eps, delta) = self + .combined_eps_delta(&accuracy) + .ok_or("exact accuracy requires exact execution")?; + if !eps.is_finite() + || eps <= 0.0 + || eps >= 1.0 + || !delta.is_finite() + || delta <= 0.0 + || delta >= 1.0 + { + return Err("invalid frequency accuracy budget".into()); + } + let (width, depth) = Self::cms_width_depth(eps, delta); + let width = width + .checked_next_power_of_two() + .ok_or("frequency width overflows")?; + let mut request = request.clone(); + // Only CMS is supported by the backend's frequency capability table. + // Caller-supplied minima cannot weaken workload/intent requirements. + request.formal_minimums = Some(vec![SketchConfiguration { + algorithm: SketchAlgorithm::Cms, + params: SketchParams::Cms { width, depth }, + }]); + evidence + .sketch_evidence + .validate() + .map_err(|error| error.to_string())?; + // Exclude layouts the backend cannot instantiate before selecting the + // winner, so a cheap unsupported width cannot hide a legal measured one. + let mut evidence = evidence.clone(); + evidence.sketch_evidence.records.retain(|row| { + row.algorithm == SketchAlgorithm::Cms + && matches!(&row.params, SketchParams::Cms { width, .. } if width.is_power_of_two()) + }); + evidence.query_bindings.retain(|binding| { + evidence + .sketch_evidence + .records + .iter() + .any(|row| row.id == binding.record_id) + }); + recommend_offline(&evidence, &request) + } + + fn rank_with_offline_evidence( + &self, + intent: &AggIntent, + defaults: Vec, + ) -> Vec { + let Some(provider) = &self.offline_evidence else { + return defaults; + }; + let accuracy = intent_accuracy(intent); + if matches!(accuracy, AccuracyTarget::Exact) { + return defaults; + } + let (eps, delta) = asap_aware_mapping::replacement::accuracy_budget(&accuracy); + // Compare the parameters this deployment will actually bind, not the + // planner's default sizing or another benchmark configuration. + let costs: Option> = defaults + .iter() + .map(|algorithm| { + let params = self.size_params(algorithm.clone(), intent, eps, delta); + let row = provider.lookup(algorithm, ¶ms).ok()?; + Some((algorithm.clone(), row.metrics.update_cpu_ns.as_ref()?.value)) + }) + .collect(); + let Some(mut costs) = costs else { + return defaults; + }; + costs.sort_by(|left, right| left.1.total_cmp(&right.1)); + costs.into_iter().map(|(algorithm, _)| algorithm).collect() + } + /// Bind the cheapest complete, workload-scoped physical realization for /// each Planner-owned abstract window framework. Concrete implementation /// identities stay in the physical compiler; only framework and cost @@ -252,6 +391,7 @@ impl CostModel for ControlPlaneCostModel { .min_by(|left, right| left.1 .0.total_cmp(&right.1 .0)) .map( |(framework, physical_cost)| CompleteSummaryCandidateEstimate { + physical_plan_id: None, cost: Cost(lifecycle_cost + physical_cost.0), window_frameworks: vec![Some(framework.clone()); deployments.len()], window_accuracy_guarantee: Some( @@ -268,7 +408,7 @@ impl CostModel for ControlPlaneCostModel { intent: &AggIntent, candidates: &[SketchAlgorithm], ) -> Vec { - match intent { + let defaults = match intent { // bind_ddsketch_quantile (priority 6) always wins the old // dispatcher's tie-break over bind_kll_quantile (priority 5) // whenever both can bind (see bind_kll_quantile.rs's @@ -302,7 +442,8 @@ impl CostModel for ControlPlaneCostModel { // binds one family for each; asap-plan's static order already // puts it first (`summary_candidates`), nothing to reorder. _ => candidates.to_vec(), - } + }; + self.rank_with_offline_evidence(intent, defaults) } fn size_params( @@ -387,6 +528,19 @@ impl CostModel for ControlPlaneCostModel { if ext_kind != FREQUENCY_EXT_KIND { return Implementation::PassThrough; } + if self.offline_frequency_comparison.is_some() { + return self + .offline_frequency_recommendation(payload) + .ok() + .and_then(|recommendation| recommendation.selected_sketch().cloned()) + .map(|configuration| { + Implementation::Sketch(planner_types::post_asap::SketchKind::new( + configuration.algorithm, + configuration.params, + )) + }) + .unwrap_or(Implementation::PassThrough); + } let Some(accuracy) = payload .get("accuracy") .and_then(|v| serde_json::from_value::(v.clone()).ok()) diff --git a/control_plane/src/physical/post_asap/lower.rs b/control_plane/src/physical/post_asap/lower.rs index 4a3dda6f..db1a74f2 100644 --- a/control_plane/src/physical/post_asap/lower.rs +++ b/control_plane/src/physical/post_asap/lower.rs @@ -87,6 +87,18 @@ fn bind_recursive( cost_model: &dyn CostModel, ) -> Result { match expr { + // These roots describe exact query semantics, not summary candidate + // sites. Preserve the complete expression so archive execution retains + // selector labels, ordering, and comparison filtering. + QueryExpr::Scan { .. } + | QueryExpr::Sort { .. } + | QueryExpr::Filter { .. } + | QueryExpr::BinaryOp { + op: planner_types::pre_asap::BinaryOpKind::Compare(_), + .. + } => Ok(PostAsapPlan::Summary( + crate::planner_selection::keep_pre_asap(expr)?, + )), // `QueryExpr::LetBinding`/`::Ref` don't exist in the canonical IR // anymore (ASAPPlanner#181/#192 -- see // control_plane/docs/design-asapplanner-pin-migration.md), so diff --git a/control_plane/src/physical/post_asap/tests.rs b/control_plane/src/physical/post_asap/tests.rs index 324f710c..4a1aaae4 100644 --- a/control_plane/src/physical/post_asap/tests.rs +++ b/control_plane/src/physical/post_asap/tests.rs @@ -115,9 +115,12 @@ fn node_is_archive(node: &Rc) -> bool { SummaryExpr::SummaryJoin { outer, inner, .. } => { node_is_archive(outer) || node_is_archive(inner) } - SummaryExpr::SummarySubtract { left, right } => { - node_is_archive(left) || node_is_archive(right) - } + SummaryExpr::SummarySubtract { left, right } + | SummaryExpr::BinaryOp { + lhs: left, + rhs: right, + .. + } => node_is_archive(left) || node_is_archive(right), SummaryExpr::SummaryDelete { summary_input, .. } => node_is_archive(summary_input), } } @@ -774,7 +777,9 @@ fn phase_b_e2e_topk_well_formed() { let accuracy = AccuracyTarget::Epsilon(0.05); let expr = crate::query_parser::parse_query_expr_canonical(query, accuracy.clone()) .expect("TopK parses"); - assert!(bind_query_expr(&expr, accuracy).is_err()); + // Current Planner supports the temporal TopK shape; the old pin rejected it. + let bound = bind_query_expr(&expr, accuracy).expect("temporal TopK binds"); + assert!(matches!(bound, PhysicalExpr::Committed(_))); } /// Archive-only routing through the full L1→L3→L4 pipeline. Asserts the diff --git a/control_plane/src/query_parser/mod.rs b/control_plane/src/query_parser/mod.rs index 5b452424..f5721c8f 100644 --- a/control_plane/src/query_parser/mod.rs +++ b/control_plane/src/query_parser/mod.rs @@ -147,7 +147,7 @@ fn root_scan_schema(qe: &QueryExpr) -> Option<&planner_types::pre_asap::Schema> | QueryExpr::Sort { child, .. } | QueryExpr::Limit { child, .. } | QueryExpr::PromqlSubquery { child, .. } => root_scan_schema(child), - QueryExpr::Concat { children } => children.iter().find_map(root_scan_schema), + QueryExpr::Concat { children, .. } => children.iter().find_map(root_scan_schema), QueryExpr::Join { left, right, .. } | QueryExpr::SetOp { left, right, .. } | QueryExpr::BinaryOp { @@ -257,7 +257,7 @@ impl QeCollector { self.visit(child, schema); } QueryExpr::Dedup { child, .. } => self.visit(child, schema), - QueryExpr::Concat { children } => { + QueryExpr::Concat { children, .. } => { for c in children { self.visit(c, schema); } diff --git a/control_plane/src/query_plan.rs b/control_plane/src/query_plan.rs index 61e5796e..44cbb9eb 100644 --- a/control_plane/src/query_plan.rs +++ b/control_plane/src/query_plan.rs @@ -334,6 +334,9 @@ where SummaryExpr::KeepPreAsap(_) => QueryPlanNode::ExactFallback { reason: "post-ASAP node requires exact execution".into(), }, + SummaryExpr::BinaryOp { .. } => QueryPlanNode::ExactFallback { + reason: "summary binary operation is not executable by the warm tier".into(), + }, SummaryExpr::SummaryAgg { family, reduction, diff --git a/control_plane/src/workload.rs b/control_plane/src/workload.rs index ae4f053a..29af1072 100644 --- a/control_plane/src/workload.rs +++ b/control_plane/src/workload.rs @@ -194,7 +194,7 @@ fn collect_agg_intents(expr: &planner_types::pre_asap::QueryExpr, out: &mut Vec< | QueryExpr::TimeRange { child, .. } | QueryExpr::TimeShift { child, .. } | QueryExpr::SQLWindowFunc { child, .. } => collect_agg_intents(child, out), - QueryExpr::Concat { children } => { + QueryExpr::Concat { children, .. } => { for child in children { collect_agg_intents(child, out); } diff --git a/control_plane/tests/o11y_exact_fallback.rs b/control_plane/tests/o11y_exact_fallback.rs new file mode 100644 index 00000000..6f0b12d8 --- /dev/null +++ b/control_plane/tests/o11y_exact_fallback.rs @@ -0,0 +1,42 @@ +use control_plane::{ + physical::post_asap::{bind_query_expr, PhysicalExpr, PostAsapPlan}, + query_parser::parse_query_expr_canonical, + types_v2::AccuracyTarget, +}; +use planner_types::post_asap::SummaryExpr; + +/// Non-summary o11y roots preserve labels, ordering and comparison filtering +/// as the original exact IR, instead of failing summary candidate selection. +#[test] +fn o11y_non_summary_roots_preserve_exact_query_semantics() { + for query in [ + "service_cache_refresh_lag_seconds{job=\"user-service\"}", + "sort_desc(sum by (job) (rate(process_cpu_seconds_total{job=~\".+\"}[6h])))", + "sum by (job) (increase(http_requests_total{status=~\"5..\",job=~\".+-service\"}[24h])) > 0", + ] { + for accuracy in [AccuracyTarget::Exact, AccuracyTarget::Epsilon(0.01)] { + let expr = parse_query_expr_canonical(query, accuracy.clone()).unwrap(); + let result = bind_query_expr(&expr, accuracy) + .unwrap_or_else(|error| panic!("{query}: {error}")); + let PhysicalExpr::Committed(PostAsapPlan::Summary(node)) = result else { + panic!("expected an explicit exact fallback for {query}"); + }; + let SummaryExpr::KeepPreAsap(original) = &node.expr else { + panic!("expected original exact query for {query}"); + }; + assert_eq!(original.as_ref(), &expr, "query semantics changed: {query}"); + assert_eq!(node.schema.fields.len(), expr.output_schema().unwrap().columns.len()); + let executable = control_plane::query_plan::QueryPlanEntry::compile_bound( + "fixture".into(), query.into(), &node, + control_plane::query_plan::InstantExecution { + lookback_ms: 300_000, full_history: false, cumulative_readout: false, + }, + control_plane::query_plan::FallbackPolicy::Reject, + |_, _| panic!("exact fallback must not request summary materializations"), + ).unwrap(); + assert!(matches!(executable.nodes[&executable.root], + control_plane::query_plan::QueryPlanNode::ExactFallback { .. })); + assert!(executable.materialization_bindings().is_empty()); + } + } +} diff --git a/control_plane/tests/offline_evidence.rs b/control_plane/tests/offline_evidence.rs new file mode 100644 index 00000000..96b5a82d --- /dev/null +++ b/control_plane/tests/offline_evidence.rs @@ -0,0 +1,376 @@ +//! Synthetic fixtures test integration decisions, not measured sketch benefits. +use asap_aware_mapping::{ + empirical_cost::{EmpiricalEvidenceProvider, EvidenceArtifact, EvidenceContext}, + CostModel, +}; +use control_plane::{ + physical::post_asap::{ + bind_query_expr_with_cost_model, cost_model::ControlPlaneCostModel, PhysicalExpr, + PostAsapPlan, + }, + query_parser::parse_query_expr_canonical, + types_v2::AccuracyTarget, +}; +use planner_types::{ + post_asap::{SketchAlgorithm, SketchParams, SummaryExpr, SummaryFamilyType, SummaryNode}, + pre_asap::AggIntent, +}; +use serde_json::json; + +fn frequency_comparison() -> ( + asap_aware_mapping::empirical_comparison::OfflineComparisonEvidence, + asap_aware_mapping::empirical_comparison::OfflineComparisonRequest, +) { + let (mut artifact, context) = fixture(&model(), &intent()); + let query = + json!({"kind":"point_frequency", "value_type":"i64", "probe_set":"all_distinct_keys"}); + artifact.records[0].algorithm = SketchAlgorithm::Cms; + artifact.records[0].params = SketchParams::Cms { + width: 512, + depth: 5, + }; + artifact.records[0].error.as_mut().unwrap().mean = Some(0.2); + artifact.records[1].algorithm = SketchAlgorithm::Cms; + artifact.records[1].params = SketchParams::Cms { + width: 1024, + depth: 5, + }; + artifact.records[1].error.as_mut().unwrap().mean = Some(0.001); + for row in &mut artifact.records { + row.error.as_mut().unwrap().query = query.clone(); + row.metrics.build_cpu_ns = + serde_json::from_value(json!({"value":1.0,"samples":3,"stddev":0.0})).unwrap(); + row.metrics.read_cpu_ns = row.metrics.build_cpu_ns.clone(); + } + let bindings: Vec<_> = artifact + .records + .iter() + .map(|r| json!({"record_id":r.id,"query":query})) + .collect(); + let evidence = serde_json::from_value(json!({"schema_version":1,"timing_contract":"disjoint_live_state_v1", + "sketch_evidence":artifact, "query_bindings":bindings, + "exact_records":[{"id":"exact-test", "distribution":context.distribution, + "environment":context.environment, "query":query, + "measured_at_unix_seconds":100,"valid_until_unix_seconds":200, + "provenance":{"command":"synthetic test","dataset":"test","source_revision":"test","repetitions":3}, + "metrics":{"empty_build_cpu_ns":{"value":1.0,"samples":3}, + "update_cpu_ns":{"value":100.0,"samples":3}, + "prepare_cpu_ns":{"value":1.0,"samples":3}, + "read_cpu_ns":{"value":100.0,"samples":3}}}]})).unwrap(); + let request = serde_json::from_value(json!({"context":context, + "exact_environment":context.environment,"query":query, + "accuracy":{"metric":"absolute relative frequency error","max_observed_mean":0.01,"minimum_trials":3}, + "workload":{"input_items_per_state":1000,"reads_per_state":100,"merges_per_state":0,"state_instances":1,"horizon_seconds":3600.0}, + "weights":{"cpu_ns_weight":1.0,"retained_byte_seconds_weight":0.0},"formal_minimums":null})).unwrap(); + (evidence, request) +} + +fn point_frequency(accuracy: AccuracyTarget) -> AggIntent { + control_plane::planner_selection::frequency(accuracy, Some(("key".into(), "7".into()))) +} + +/// A query-matched mean-error budget selects a larger measured CMS, while +/// preserving the formal workload bound and the requested point readout. +#[test] +fn offline_frequency_error_budget_changes_configuration() { + let (evidence, request) = frequency_comparison(); + let empirical = model().with_offline_frequency_comparison(evidence, request); + let intent = point_frequency(AccuracyTarget::Epsilon(0.01)); + let AggIntent::Extension { ext_kind, payload } = &intent else { + unreachable!() + }; + let asap_aware_mapping::Implementation::Sketch(selected) = + empirical.realize_extension(ext_kind, payload) + else { + panic!( + "expected a measured frequency sketch: {:?}", + empirical.offline_frequency_recommendation(payload) + ); + }; + assert_eq!( + selected.params(), + &SketchParams::Cms { + width: 1024, + depth: 5 + } + ); + let recommendation = empirical.offline_frequency_recommendation(payload).unwrap(); + assert!(recommendation.checked_formal_minimums); + assert!(recommendation.candidates[0] + .rejection + .as_ref() + .unwrap() + .contains("observed error")); + + let scan = + parse_query_expr_canonical("offline_metric{key=~\".+\"}", AccuracyTarget::Exact).unwrap(); + let expr = planner_types::pre_asap::QueryExpr::Aggregate { + reduction: planner_types::pre_asap::Reduction::PerEntity, + measures: vec![intent], + output_names: vec![], + having: None, + child: std::rc::Rc::new(scan), + }; + let PhysicalExpr::Committed(PostAsapPlan::Summary(node)) = + bind_query_expr_with_cost_model(&expr, &empirical).unwrap() + else { + panic!("expected summary") + }; + assert_eq!( + sketch(&node).1, + &SketchParams::Cms { + width: 1024, + depth: 5 + } + ); + assert!( + matches!(&node.expr, SummaryExpr::SummaryEstimate { query: planner_types::post_asap::SketchQuery::PointCount { + key: planner_types::pre_asap::ColumnRef::Named(key), value: Some(value) + }, .. } if key == "key" && value == "7") + ); +} + +/// An infeasible empirical budget, stale context, exact request, or missing +/// item filter must retain exact execution, never invent a default sketch. +#[test] +fn offline_frequency_comparison_falls_back_when_inapplicable() { + for scenario in ["error", "stale", "exact", "missing_item", "formal_minimum"] { + let (evidence, mut request) = frequency_comparison(); + if scenario == "error" { + request.accuracy.max_observed_mean = 0.0; + } + if scenario == "stale" { + request.context.now_unix_seconds = 201; + } + let empirical = model().with_offline_frequency_comparison(evidence, request); + let intent = match scenario { + "exact" => point_frequency(AccuracyTarget::Exact), + "missing_item" => { + control_plane::planner_selection::frequency(AccuracyTarget::Epsilon(0.01), None) + } + "formal_minimum" => point_frequency(AccuracyTarget::Epsilon(0.001)), + _ => point_frequency(AccuracyTarget::Epsilon(0.01)), + }; + let AggIntent::Extension { ext_kind, payload } = intent else { + unreachable!() + }; + assert!( + matches!( + empirical.realize_extension(&ext_kind, &payload), + asap_aware_mapping::Implementation::PassThrough + ), + "{scenario}" + ); + } +} + +/// An unsupported cheap CMS width must not hide a valid measured alternative. +#[test] +fn offline_frequency_filters_backend_layout_before_comparison() { + let (mut evidence, request) = frequency_comparison(); + let mut unsupported = evidence.sketch_evidence.records[1].clone(); + unsupported.id = "unsupported-cheap-width".into(); + unsupported.params = SketchParams::Cms { + width: 1500, + depth: 5, + }; + unsupported.metrics.update_cpu_ns.as_mut().unwrap().value = 0.001; + let mut query_binding = evidence.query_bindings[1].clone(); + query_binding.record_id = unsupported.id.clone(); + evidence.query_bindings.push(query_binding); + evidence.sketch_evidence.records.push(unsupported); + let empirical = model().with_offline_frequency_comparison(evidence, request); + let AggIntent::Extension { payload, .. } = point_frequency(AccuracyTarget::Epsilon(0.01)) + else { + unreachable!() + }; + let recommendation = empirical + .offline_frequency_recommendation(&payload) + .unwrap(); + assert_eq!( + recommendation.selected_sketch().unwrap().params, + SketchParams::Cms { + width: 1024, + depth: 5 + } + ); + assert!(recommendation + .candidates + .iter() + .all(|c| c.record_id != "unsupported-cheap-width")); +} + +fn fixture( + model: &ControlPlaneCostModel, + intent: &AggIntent, +) -> (EvidenceArtifact, EvidenceContext) { + let distribution = json!({"id":"synthetic-unit-test", "family":"uniform", "sample_count":1000, + "distinct_count":100, "parameters":{"seed":42}}); + let environment = json!({"id":"test-machine", "cpu":"test", "os":"test", "runtime":"test", + "implementation":"synthetic-test-fixture", "implementation_version":"v1"}); + let records: Vec<_> = [(SketchAlgorithm::Cms,20.0), (SketchAlgorithm::CountSketch,10.0)].into_iter().map(|(algorithm,value)| { + let params = model.size_params(algorithm.clone(), intent, 0.01, 0.01); + json!({"id":format!("test-{algorithm:?}"), "algorithm":algorithm, "params":params, + "distribution":distribution, "environment":environment, + "measured_at_unix_seconds":100, "valid_until_unix_seconds":200, + "provenance":{"command":"synthetic integration fixture, not a real benchmark", "dataset":"test", "source_revision":"test", "repetitions":3}, + "metrics":{"update_cpu_ns":{"value":value,"stddev":1.0,"samples":3}}, + "error":{"metric":"absolute relative frequency error", "mean":0.0, "max":null, "trials":3, + "ground_truth_method":"synthetic fixture", "query":{"kind":"point_frequency"}}}) + }).collect(); + let artifact = serde_json::from_value( + json!({"schema_version":1,"benchmark_version":"synthetic-test-v1", + "model_version":"empirical-update-cpu-v1","records":records}), + ) + .unwrap(); + let context = serde_json::from_value( + json!({"distribution":distribution,"environment":environment,"now_unix_seconds":150}), + ) + .unwrap(); + (artifact, context) +} + +fn intent() -> AggIntent { + AggIntent::Count { + accuracy: AccuracyTarget::Epsilon(0.01), + } +} +fn candidates() -> Vec { + vec![SketchAlgorithm::Cms, SketchAlgorithm::CountSketch] +} +fn model() -> ControlPlaneCostModel { + ControlPlaneCostModel::new(AccuracyTarget::Epsilon(0.01)) +} + +fn bound(model: &ControlPlaneCostModel) -> std::rc::Rc { + let query = parse_query_expr_canonical( + "count_over_time(offline_metric[5m])", + AccuracyTarget::Epsilon(0.01), + ) + .unwrap(); + let PhysicalExpr::Committed(PostAsapPlan::Summary(node)) = + bind_query_expr_with_cost_model(&query, model).unwrap() + else { + panic!("expected summary binding") + }; + node +} + +fn sketch(node: &SummaryNode) -> (&SketchAlgorithm, &SketchParams) { + match &node.expr { + SummaryExpr::SummaryEstimate { summary_input, .. } => sketch(summary_input), + SummaryExpr::SummaryAgg { + family: SummaryFamilyType::Sketch(kind, _), + .. + } => (kind.algorithm(), kind.params()), + other => panic!("expected sketch, got {other:?}"), + } +} + +/// The actual control-plane parser/binder selects the lower measured update +/// cost while preserving the selected algorithm's normal parameter sizing. +#[test] +fn offline_update_evidence_changes_typed_binding() { + let default = model(); + let (artifact, context) = fixture(&default, &intent()); + let empirical = + model().with_offline_evidence(EmpiricalEvidenceProvider::new(artifact, context).unwrap()); + assert_eq!( + default.rank_candidates(&intent(), &candidates()), + candidates() + ); + assert_eq!( + empirical.rank_candidates(&intent(), &candidates()), + vec![SketchAlgorithm::CountSketch, SketchAlgorithm::Cms] + ); + let default_bound = bound(&default); + let measured_bound = bound(&empirical); + assert_eq!(sketch(&default_bound).0, &SketchAlgorithm::Cms); + assert_eq!(sketch(&measured_bound).0, &SketchAlgorithm::CountSketch); + assert_eq!( + sketch(&measured_bound).1, + &default.size_params(SketchAlgorithm::CountSketch, &intent(), 0.01, 0.01) + ); + assert!(default_bound.guarantee.is_some()); + assert!(measured_bound.guarantee.is_some()); + // Observed zero point-frequency error has no effect on formal sizing. + for kind in candidates() { + assert_eq!( + empirical.size_params(kind.clone(), &intent(), 0.01, 0.01), + default.size_params(kind, &intent(), 0.01, 0.01) + ); + } +} + +/// Stale/missing/environment-mismatched data and parameters for another +/// workload accuracy all fall back to the deployment's existing preference. +#[test] +fn incompatible_evidence_preserves_deployment_behavior() { + for scenario in ["missing", "stale", "environment", "workload_sizing"] { + let (mut artifact, mut context) = fixture(&model(), &intent()); + let mut deployment = model(); + match scenario { + "missing" => { + artifact.records.remove(1); + } + "stale" => context.now_unix_seconds = 201, + "environment" => context.environment.cpu = "different CPU".into(), + "workload_sizing" => { + deployment = ControlPlaneCostModel::new(AccuracyTarget::Epsilon(0.001)) + } + _ => unreachable!(), + } + let empirical = deployment + .with_offline_evidence(EmpiricalEvidenceProvider::new(artifact, context).unwrap()); + assert_eq!( + empirical.rank_candidates(&intent(), &candidates()), + candidates(), + "{scenario}" + ); + assert_eq!( + sketch(&bound(&empirical)).0, + &SketchAlgorithm::Cms, + "{scenario}" + ); + } +} + +/// A planner binary summary cannot become a silently executable warm-tier +/// plan before that tier implements binary summary evaluation. +#[test] +fn binary_summary_has_explicit_warm_tier_fallback() { + use control_plane::query_plan::{ + FallbackPolicy, InstantExecution, QueryPlanEntry, QueryPlanNode, + }; + use planner_types::{post_asap::BinaryOperator, pre_asap::BinaryOpKind}; + let child = bound(&model()); + let root = std::rc::Rc::new(SummaryNode { + expr: SummaryExpr::BinaryOp { + lhs: child.clone(), + rhs: child.clone(), + operator: BinaryOperator { + kind: BinaryOpKind::Arithmetic(planner_types::pre_asap::ArithmeticOpKind::Div), + vector_match: None, + }, + }, + schema: child.schema.clone(), + guarantee: None, + }); + let plan = QueryPlanEntry::compile_bound( + "test".into(), + "left / right".into(), + &root, + InstantExecution { + lookback_ms: 300000, + full_history: false, + cumulative_readout: false, + }, + FallbackPolicy::Reject, + |_, _| panic!("unsupported binary plan must not bind a materialization"), + ) + .unwrap(); + assert!( + matches!(&plan.nodes[&plan.root], QueryPlanNode::ExactFallback { reason } if reason.contains("binary operation")) + ); + assert!(plan.materialization_bindings().is_empty()); +} diff --git a/crates/asap_types/Cargo.toml b/crates/asap_types/Cargo.toml index fe3a0270..9ab0e5df 100644 --- a/crates/asap_types/Cargo.toml +++ b/crates/asap_types/Cargo.toml @@ -30,4 +30,4 @@ xxhash-rust = { version = "0.8", features = ["xxh64"] } # exactly (`control_plane/Cargo.toml`) -- two different revs of the same # git dependency in one workspace resolve to two distinct Rust types that # won't unify. -planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "cb50219c582d43f53ab77d3a595bd1ea4a9aa119" } +planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "dfdf6b5c1f7d667394a4ea1f56fb3786af04228d" } diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index a424258b..e809ac8d 100644 --- a/data_plane/Cargo.toml +++ b/data_plane/Cargo.toml @@ -38,7 +38,7 @@ control_plane = { path = "../control_plane" } # reduction: Reduction, .. }`) are `pre_asap` types, in the same crate now # (not a separate `asap-ir` import). Query serving consumes the compiled # QueryPlan; these types are used at physical-plan compilation boundaries. -planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "cb50219c582d43f53ab77d3a595bd1ea4a9aa119" } +planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "dfdf6b5c1f7d667394a4ea1f56fb3786af04228d" } # Shared external (workspace) serde.workspace = true diff --git a/data_plane/src/query_engines/asap_query_engine/post_asap_planner.rs b/data_plane/src/query_engines/asap_query_engine/post_asap_planner.rs index e2b351b8..fb9fc4d3 100644 --- a/data_plane/src/query_engines/asap_query_engine/post_asap_planner.rs +++ b/data_plane/src/query_engines/asap_query_engine/post_asap_planner.rs @@ -386,7 +386,7 @@ fn query_expr_contains_time_range(qe: &planner_types::pre_asap::QueryExpr) -> bo | QueryExpr::PromqlSubquery { child, .. } | QueryExpr::TimeShift { child, .. } | QueryExpr::SQLWindowFunc { child, .. } => query_expr_contains_time_range(child), - QueryExpr::Concat { children } => children.iter().any(query_expr_contains_time_range), + QueryExpr::Concat { children, .. } => children.iter().any(query_expr_contains_time_range), QueryExpr::Join { left, right, .. } | QueryExpr::SetOp { left, right, .. } | QueryExpr::BinaryOp { @@ -415,7 +415,7 @@ fn query_expr_contains_rate(qe: &planner_types::pre_asap::QueryExpr) -> bool { | QueryExpr::TimeRange { child, .. } | QueryExpr::TimeShift { child, .. } | QueryExpr::SQLWindowFunc { child, .. } => query_expr_contains_rate(child), - QueryExpr::Concat { children } => children.iter().any(query_expr_contains_rate), + QueryExpr::Concat { children, .. } => children.iter().any(query_expr_contains_rate), QueryExpr::Join { left, right, .. } | QueryExpr::SetOp { left, right, .. } | QueryExpr::BinaryOp { @@ -441,7 +441,7 @@ fn query_expr_has_filter(qe: &planner_types::pre_asap::QueryExpr) -> bool { | QueryExpr::TimeRange { child, .. } | QueryExpr::TimeShift { child, .. } | QueryExpr::SQLWindowFunc { child, .. } => query_expr_has_filter(child), - QueryExpr::Concat { children } => children.iter().any(query_expr_has_filter), + QueryExpr::Concat { children, .. } => children.iter().any(query_expr_has_filter), QueryExpr::Join { left, right, .. } | QueryExpr::SetOp { left, right, .. } | QueryExpr::BinaryOp { @@ -472,7 +472,7 @@ fn query_expr_max_time_range_ms(qe: &planner_types::pre_asap::QueryExpr) -> Opti | QueryExpr::Limit { child, .. } | QueryExpr::TimeShift { child, .. } | QueryExpr::SQLWindowFunc { child, .. } => child_max(child), - QueryExpr::Concat { children } => children + QueryExpr::Concat { children, .. } => children .iter() .filter_map(query_expr_max_time_range_ms) .max(), diff --git a/data_plane/src/query_engines/asap_query_engine/summary_exec.rs b/data_plane/src/query_engines/asap_query_engine/summary_exec.rs index 13ce252e..3150e635 100644 --- a/data_plane/src/query_engines/asap_query_engine/summary_exec.rs +++ b/data_plane/src/query_engines/asap_query_engine/summary_exec.rs @@ -154,10 +154,19 @@ pub fn execute( SummaryExpr::SummaryAgg { child, family, - col, + input, reduction, .. } => { + let planner_types::post_asap::SummaryUpdate { + item: None, + weight: planner_types::post_asap::SummaryInputExpr::Column(col), + } = input + else { + return Err(ExecError::NotYetSupported( + "keyed or non-column summary update", + )); + }; let tagged = exec.find_candidates(family, col, reduction, child)?; if tagged.is_empty() { return Err(ExecError::NoCandidates); @@ -235,6 +244,7 @@ pub fn execute( SummaryExpr::SummaryJoin { .. } => Err(ExecError::NotYetSupported("SummaryJoin")), SummaryExpr::SummarySubtract { .. } => Err(ExecError::NotYetSupported("SummarySubtract")), SummaryExpr::SummaryDelete { .. } => Err(ExecError::NotYetSupported("SummaryDelete")), + SummaryExpr::BinaryOp { .. } => Err(ExecError::NotYetSupported("BinaryOp")), } } @@ -325,7 +335,7 @@ mod tests { expr: SummaryExpr::SummaryAgg { child, family, - col: ColumnRef::SampleValue, + input: planner_types::post_asap::SummaryUpdate::column(ColumnRef::SampleValue), reduction, grouping: planner_types::post_asap::GroupingStrategy::default(), }, @@ -478,6 +488,63 @@ mod tests { assert_eq!(only(v), 7.0); } + /// Legacy candidate lookup cannot discard keyed items or constant weights + /// when adapting the planner's typed update contract. + #[test] + fn unsupported_update_semantics_fail_before_candidate_lookup() { + use planner_types::post_asap::{SummaryInputExpr, SummaryUpdate}; + let exec = MockExecutor::new(); + exec.register(7.0); + for input in [ + SummaryUpdate { + item: Some(SummaryInputExpr::Column(ColumnRef::Named("key".into()))), + weight: SummaryInputExpr::Column(ColumnRef::SampleValue), + }, + SummaryUpdate { + item: None, + weight: SummaryInputExpr::Constant(1.0), + }, + ] { + let mut tree = agg_node(sum(), logical_node()); + let SummaryExpr::SummaryAgg { input: update, .. } = &mut Rc::make_mut(&mut tree).expr + else { + unreachable!() + }; + *update = input; + assert!(matches!( + execute(&tree, &exec), + Err(ExecError::NotYetSupported( + "keyed or non-column summary update" + )) + )); + } + } + + /// Binary summaries remain explicit unsupported execution until value and + /// label/time matching are implemented by the warm tier. + #[test] + fn binary_summary_is_not_executed_as_one_operand() { + let child = logical_node(); + let tree = SummaryNode { + expr: SummaryExpr::BinaryOp { + lhs: child.clone(), + rhs: child.clone(), + operator: planner_types::post_asap::BinaryOperator { + kind: planner_types::pre_asap::BinaryOpKind::Arithmetic( + planner_types::pre_asap::ArithmeticOpKind::Div, + ), + vector_match: None, + }, + }, + schema: child.schema.clone(), + guarantee: None, + }; + assert!(matches!( + execute(&tree, &MockExecutor::new()), + Err(ExecError::NotYetSupported("BinaryOp")) + )); + } + #[test] fn multiple_candidates_for_one_agg_are_merged() { let exec = MockExecutor::new(); diff --git a/data_plane/src/query_engines/asap_query_engine/summary_executor.rs b/data_plane/src/query_engines/asap_query_engine/summary_executor.rs index 025005cf..cc37b7d9 100644 --- a/data_plane/src/query_engines/asap_query_engine/summary_executor.rs +++ b/data_plane/src/query_engines/asap_query_engine/summary_executor.rs @@ -1142,7 +1142,7 @@ pub(crate) fn find_metric_in_query_expr(qe: &QueryExpr) -> Option { | QueryExpr::TimeRange { child, .. } | QueryExpr::TimeShift { child, .. } | QueryExpr::SQLWindowFunc { child, .. } => find_metric_in_query_expr(child), - QueryExpr::Concat { children } => children.iter().find_map(find_metric_in_query_expr), + QueryExpr::Concat { children, .. } => children.iter().find_map(find_metric_in_query_expr), QueryExpr::Join { left, .. } | QueryExpr::SetOp { left, .. } => { find_metric_in_query_expr(left) } @@ -1221,7 +1221,7 @@ mod tests { planner_types::post_asap::SketchAlgorithm::Kll, planner_types::post_asap::SketchParams::Kll { k: 200 }, ), - col: ColumnRef::SampleValue, + input: planner_types::post_asap::SummaryUpdate::column(ColumnRef::SampleValue), reduction, grouping: planner_types::post_asap::GroupingStrategy::default(), }, @@ -1245,7 +1245,7 @@ mod tests { planner_types::post_asap::SketchAlgorithm::Hll, planner_types::post_asap::SketchParams::Hll { precision: 10 }, ), - col: ColumnRef::SampleValue, + input: planner_types::post_asap::SummaryUpdate::column(ColumnRef::SampleValue), reduction, grouping: planner_types::post_asap::GroupingStrategy::default(), }, @@ -1395,7 +1395,7 @@ mod tests { depth: 4, }, ), - col: ColumnRef::SampleValue, + input: planner_types::post_asap::SummaryUpdate::column(ColumnRef::SampleValue), reduction: Reduction::by(vec![]), grouping: planner_types::post_asap::GroupingStrategy::default(), }, @@ -1458,7 +1458,7 @@ mod tests { heap_size: 10, }, ), - col: ColumnRef::SampleValue, + input: planner_types::post_asap::SummaryUpdate::column(ColumnRef::SampleValue), reduction: Reduction::by(vec![]), grouping: planner_types::post_asap::GroupingStrategy::default(), }, @@ -1502,7 +1502,7 @@ mod tests { planner_types::post_asap::ExactKind::Sum, planner_types::post_asap::ExactParams::Sum, ), - col: ColumnRef::SampleValue, + input: planner_types::post_asap::SummaryUpdate::column(ColumnRef::SampleValue), // Sum is a genuine PromQL aggregation operator -- an empty // `by` always means "reduce fully," never `PerEntity` (see // `resolve_group_key`'s doc). @@ -2327,7 +2327,7 @@ mod tests { planner_types::post_asap::SketchAlgorithm::Kll, planner_types::post_asap::SketchParams::Kll { k: 500 }, ), - col: ColumnRef::SampleValue, + input: planner_types::post_asap::SummaryUpdate::column(ColumnRef::SampleValue), reduction: Reduction::by(vec![]), grouping: planner_types::post_asap::GroupingStrategy::default(), }, @@ -2788,7 +2788,7 @@ mod tests { planner_types::post_asap::ExactKind::MinMax, planner_types::post_asap::ExactParams::MinMax, ), - col: ColumnRef::SampleValue, + input: planner_types::post_asap::SummaryUpdate::column(ColumnRef::SampleValue), reduction: Reduction::by(vec![]), grouping: planner_types::post_asap::GroupingStrategy::default(), }, diff --git a/docs/examples/asapquery-compatibility-demo-snapshot.json b/docs/examples/asapquery-compatibility-demo-snapshot.json index 2870e543..eb62b192 100644 --- a/docs/examples/asapquery-compatibility-demo-snapshot.json +++ b/docs/examples/asapquery-compatibility-demo-snapshot.json @@ -1,7 +1,7 @@ { "snapshot_version": 1, "query_workload": { - "language": "prom_q_l", + "language": "promql", "query_batch": null, "repeating_queries": [ { diff --git a/docs/examples/asapquery-planning-snapshot.json b/docs/examples/asapquery-planning-snapshot.json index b7a3bcb4..fb032673 100644 --- a/docs/examples/asapquery-planning-snapshot.json +++ b/docs/examples/asapquery-planning-snapshot.json @@ -1,7 +1,7 @@ { "snapshot_version": 1, "query_workload": { - "language": "prom_q_l", + "language": "promql", "query_batch": null, "repeating_queries": [ { diff --git a/tools/run-offline-planner-replay.py b/tools/run-offline-planner-replay.py new file mode 100644 index 00000000..40b3a0ec --- /dev/null +++ b/tools/run-offline-planner-replay.py @@ -0,0 +1,51 @@ +#!/usr/bin/env python3 +"""Run the control-plane offline example against an explicit planner checkout.""" +import argparse +import json +from pathlib import Path +import re +import subprocess +import sys + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--planner", type=Path, help="optional local planner override; default uses the pinned git dependency") + parser.add_argument("--queries", required=True, type=Path) + parser.add_argument("--evidence", type=Path) + parser.add_argument("--context", type=Path) + parser.add_argument("--output", required=True, type=Path) + args = parser.parse_args() + if bool(args.evidence) != bool(args.context): + parser.error("--evidence and --context must be provided together") + backend = Path(__file__).resolve().parents[1] + planner = args.planner.resolve() if args.planner else None + command = ["cargo", "run", "--quiet", "-p", "control_plane", "--example", "offline_planner_replay"] + for package, directory in ([("asap-types", "types"), ("asap-aware-mapping", "asap-aware-mapping"), ("asap-frontend-promql", "frontend-promql")] if planner else []): + path = planner / "crates" / directory + if not (path / "Cargo.toml").is_file(): + parser.error(f"missing planner crate: {path}") + command.extend(["--config", f'patch."https://github.com/ProjectASAP/ASAPPlanner".{package}.path={json.dumps(str(path))}']) + command.extend(["--", str(args.queries.resolve())]) + if args.evidence: + command.extend([str(args.evidence.resolve()), str(args.context.resolve())]) + result = subprocess.run(command, cwd=backend, capture_output=True, text=True) + if result.returncode: + sys.stderr.write(result.stderr) + result.check_returncode() + report = json.loads(result.stdout) + report["reproduction"] = { + "command": command, + "backend_revision": subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=backend, text=True).strip(), + "planner_revision": subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=planner, text=True).strip() if planner else re.search(r'asap-aware-mapping = .*rev = "([a-f0-9]+)"', (backend / "control_plane/Cargo.toml").read_text()).group(1), + "backend_dirty": bool(subprocess.check_output(["git", "status", "--porcelain"], cwd=backend, text=True)), + "planner_dirty": bool(subprocess.check_output(["git", "status", "--porcelain"], cwd=planner, text=True)) if planner else False, + "planner_source": "local_override" if planner else "pinned_git_dependency", + } + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(json.dumps(report, indent=2) + "\n") + print(f"wrote {len(report['rows'])} query/mode results to {args.output}") + + +if __name__ == "__main__": + main() From c2a632e1ab4baf5a4183b5209adfc691a3d9aad3 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 8 Sep 2026 12:43:58 -0600 Subject: [PATCH 2/3] docs(bench): own offline evaluation tooling and guides in backend --- .../empirical-o11y-execution-plan.md | 67 ++++ docs/offline-o11y-final-2026-09-08.md | 102 +++++ docs/user_guide/o11y-replay.md | 86 ++++ tools/empirical-bench/.gitignore | 3 + tools/empirical-bench/ARTIFACTS.md | 32 ++ tools/empirical-bench/README.md | 117 ++++++ tools/empirical-bench/memory_probe.rs | 101 +++++ tools/empirical-bench/resource_probe.rs | 138 +++++++ tools/empirical-bench/run.py | 367 ++++++++++++++++++ tools/empirical-bench/test_export.py | 51 +++ 10 files changed, 1064 insertions(+) create mode 100644 docs/design_docs/empirical-o11y-execution-plan.md create mode 100644 docs/offline-o11y-final-2026-09-08.md create mode 100644 docs/user_guide/o11y-replay.md create mode 100644 tools/empirical-bench/.gitignore create mode 100644 tools/empirical-bench/ARTIFACTS.md create mode 100644 tools/empirical-bench/README.md create mode 100644 tools/empirical-bench/memory_probe.rs create mode 100644 tools/empirical-bench/resource_probe.rs create mode 100644 tools/empirical-bench/run.py create mode 100644 tools/empirical-bench/test_export.py diff --git a/docs/design_docs/empirical-o11y-execution-plan.md b/docs/design_docs/empirical-o11y-execution-plan.md new file mode 100644 index 00000000..00e98b74 --- /dev/null +++ b/docs/design_docs/empirical-o11y-execution-plan.md @@ -0,0 +1,67 @@ +# Offline evidence and o11y planning evaluation + +This is the historical offline-evidence milestone plan. The current prototype +adds actual backend-only data-plane execution using the supplied OpenMetrics +dataset, as described in the [current evaluation guide](../user_guide/o11y-replay.md). +Benchmark production and execution tooling now belong to ASAPQuery-backend; +planner-only coverage and synthetic cached-result timing are not its acceptance criteria. + +Audience: developers reproducing issue #322 and the planner/control-plane evaluation. + +## Scope + +Use offline sketch-bench CPU, elapsed time, state/memory, disk (when measured), +and errors against offline ground truth. No runtime ground truth, posterior +feedback, or self-estimated accuracy is required. Offline accuracy observations +do not establish formal guarantees on unseen distributions. + +## Execution and ownership + +1. Pin isolated planner and control-plane worktrees from fetched main branches. + Preserve existing worktrees, including unresolved cost-model changes. +2. Evidence agent: implement the versioned artifact, validation, compatibility + matching, public cost-model integration, and fallback/decision tests. +3. Benchmark agent: run actual sketch-bench algorithms on uniform and Zipf + inputs, preserve raw output, export artifacts with source and environment + provenance, and document reproducible commands. +4. Replay agent: run the existing o11y PromQL corpus through the backend parser, + its ASAPPlanner call, and the backend typed binder (not a planner-only entry), + export binding/fallback coverage and planning latency, and separately identify + supplemental sketch workloads. +5. Integration owner: connect compatible evidence to the downstream control + plane, validate dependency compatibility, review other agents' changes, run + integration checks, and report evidence-supported comparisons. + +Steps 2–4 run concurrently after agreeing the artifact contract. Integration +uses their completed interfaces and measurements. Query-pattern changes are +limited to demonstrated blockers with regression coverage; unsupported query +semantics remain explicit in the report. + +## Acceptance + +- Versioned schema and example; explicit units, algorithm/configuration, + dataset/distribution, environment, collection/validity times, sample count, + dispersion, and benchmark/model provenance. +- At least two actual sketch algorithms measured offline on uniform and skewed + inputs. CPU is distinct from elapsed time; serialization size is distinct + from heap memory and disk I/O. Unmeasured fields stay unavailable. +- Matching evidence affects a public planning cost/ranking/lifecycle boundary. + Tests cover changed decisions, missing/stale/mismatched evidence, invalid + values, and unchanged accuracy guarantees. +- The planner and control-plane evaluation preserve exact fallbacks and expose + unsupported shapes, measured provenance, and limits on benefit estimates. +- Reproducible measurement and replay commands, raw machine-readable results, + and a concise report distinguish actual measurements from modeled totals. + +## Comparison rules + +Compare equivalent tasks, data, parameters, windows, horizons, and evaluation +cadences. Whole-plan estimates include build/update/readout, retained windows, +sharing, and raw residual work where evidence exists. Missing raw baseline or +physical evidence means an unavailable whole-plan speedup, not zero cost. +Microbenchmark algorithm comparisons are labeled separately. Disk usage and +network savings require their own measurements or explicit models. + +Repeated-query break-even can be computed only when compatible exact baseline, +sketch build/maintenance, and readout measurements are present. It is an offline +estimate and does not establish deployed end-to-end latency improvement. diff --git a/docs/offline-o11y-final-2026-09-08.md b/docs/offline-o11y-final-2026-09-08.md new file mode 100644 index 00000000..64ddcf03 --- /dev/null +++ b/docs/offline-o11y-final-2026-09-08.md @@ -0,0 +1,102 @@ +# Historical Offline Sketch Evidence and o11y Report + +This report records the earlier offline experiment, not the backend-only +prototype's execution results. Its synthetic exact-result cache experiment is +out of the current prototype scope; its source tool is not part of the prototype. +Current work uses user-provided OpenMetrics data and queries entering the backend, +followed by ASAPPlanner and actual data-plane execution. No execution speedup or +resource reduction is established by the historical results below. + +Audience: developers. This report covers the offline evidence path for #322, +without real-time error feedback. The original dirty working directory was left unchanged. + +## Completed Integration + +- Versioned artifacts, JSON Schema, and a public CostModel provider match parameters, query semantics, distributions, environments, and validity intervals. Unknown measurements remain unavailable. +- Actual CMS/CountSketch parameter sweeps measure offline error, separate construction/update/read/merge CPU, live heap allocations, serialized size, and allocated file blocks. +- Fixed-snapshot point-frequency comparisons use explicit observed-error thresholds, formal parameter minima, CPU/retained byte-seconds weights, and an exact baseline on the same data. +- The backend filters unsupported layouts before recommending and binding frequency configurations. When exact execution wins or evidence is insufficient, it preserves the original query without silently rounding selected parameters. +- o11y planner/control-plane replay and exact fixed-snapshot reference measurements cover seven explicitly supported queries. + +For interfaces and reproduction instructions, see the [evidence contract](https://github.com/ProjectASAP/ASAPPlanner/blob/codex/empirical-o11y-322/docs/developer_docs/offline-sketch-evidence.md), +[backend benchmark commands](../tools/empirical-bench/README.md), and [backend-entry replay guide](user_guide/o11y-replay.md). + +## Measured Results + +sketch-bench is pinned to revision `87f619e843fd2e4da784160d4e205a0d0d55f032`. +The uniform and Zipf datasets each contain 20,000 i64 keys, with a key space of +1,000, seed 42, and Zipf exponent 1.1. There are 18 sketch configurations and +two exact baselines; CPU measurements use five trials after two warmups. +Accuracy comes from one offline experiment per fixed dataset, not a guarantee +across distributions or real-time ground truth. The comparison assumes one +build, 1,000 queries, 300 seconds of retention, and no merges. + +| Scenario | Uniform / Zipf results | +| --- | --- | +| CPU-only, with a 1% or 5% observed mean relative-error limit | Both select exact execution; no sketch CPU benefit | +| Generic comparator, with memory weighting | CMS 2720×5; retained heap decreases by 81.68%, while CPU increases by approximately 0.037 / 0.182 ms, respectively | +| Backend-compatible configuration, with memory weighting | CMS 4096×5; retained heap decreases by approximately 72.4%, while CPU also increases | + +The memory-weighted objective is `CPU ns + 0.01 × retained byte-seconds`. These +weights express an illustrative preference, not a conversion measured from +hardware. Exact retained heap is 296,976 B; CMS 2720×5 uses 54,400 B, and +4096×5 uses 81,920 B. Memory figures measure requested allocation bytes using +a separate System allocator probe; CPU measurements use jemalloc. These memory +figures are not process RSS. Serialized size and allocated file blocks do not +establish disk-throughput benefits. + +See `results-sweep/MEASUREMENTS.md` in the [verified experiment archive](../tools/empirical-bench/ARTIFACTS.md). +Earlier frequency measurements and control-plane replay in `results/` remain +as the initial baseline. Final frequency comparisons and control-plane results +use `results-sweep/`; costs from the two runs must not be mixed. + +## o11y Coverage and Limitations + +The replay reuses the repository's existing 27 PromQL fixtures. It does not run +upstream LLM-agent scoring or use real scenario data. + +The supported evaluation entry point is now the backend's `offline_planner_replay`: +queries enter its parser, call ASAPPlanner, and return through its typed binder. +The standalone planner-only replay was removed. Its candidate/lifecycle counts +below remain historical diagnostics, not backend support coverage. + +| Check | Result | +| --- | --- | +| Planner candidate coverage | 26/27 have exact summary candidates; 1/27 uses raw fallback; no sketch candidates | +| Complete lifecycle costs in the original replay | Insufficient evidence; all 27/27 conservatively fall back to raw execution | +| Control-plane exact/default/empirical binding | All three modes bind 27/27 successfully; each retains the original query in 5 root plans | +| Seven exact reference queries | Raw execution and cached-result reads are measured on fixed synthetic gauge snapshots, using the public CostModel and actual selection flow | +| Remaining 20 queries | No complete reference implementation; benefits remain unavailable | + +New fallbacks preserve the original IR for selector, sort, and comparison/filter +roots, increasing successful binding from 24/27 to 27/27. Binding does not mean +every plan is executable by the summary executor. Column updates in the new IR +are supported; keyed/non-column updates and warm-tier BinaryOp remain explicitly rejected. + +The seven reference queries cover instant sums, window maxima, and sums of +window averages: 300 series, three jobs, one finite gauge sample per minute, +and 60 repeated reads of the same snapshot. The retained state is the complete +exact query result, not a sliding window with advancing time. Measurements +exclude network, disk, protocol serialization, and output-label materialization; +memory figures count logical value bytes. Benefits from repeated reads cannot +be extrapolated to production end-to-end speedups. Machine-readable results are +available as `results/o11y-exact-snapshot.json` in that archive. + +Quantile measurements, real o11y traces, time drift, post-merge error, online +update and retirement costs, and complete production benefits remain outside +this coverage. Point-frequency error cannot be applied to quantiles or count_over_time. + +## Validation and Delivery + +- Planner mapping: 349 unit tests pass. +- Devtools: 3 exact benchmark tests and 4 replay tests pass; the recommendation CLI is verified with six actual requests. +- Benchmark export: 5 Python tests pass. +- Backend: 633 control-plane and 973 data-plane library tests pass; 7 new integration tests pass. +- Agents cross-reviewed the provider, measurement, and binding code; artifact text, JSON, source hashes, and provenance were checked successfully. + +The backend pins the published planner core commit +`dfdf6b5c1f7d667394a4ea1f56fb3786af04228d`. Normal builds and final replay do not +require local Cargo source patches. Core planner code, benchmark tools, and backend +integration are reviewed in separate PRs. Full generated results are distributed +as an experiment artifact, not embedded in the source diff. No PR is automatically +merged and the issue is not automatically closed. diff --git a/docs/user_guide/o11y-replay.md b/docs/user_guide/o11y-replay.md new file mode 100644 index 00000000..957b9c36 --- /dev/null +++ b/docs/user_guide/o11y-replay.md @@ -0,0 +1,86 @@ +# Offline o11y evaluation through the backend + +For developers evaluating the actual control-plane planning integration. +The canonical path is: + +`PromQL corpus -> ASAPQuery-backend parser -> ASAPPlanner -> backend typed binder/fallback` + +There is no standalone planner-only `o11y_replay` command. Run the backend's +`offline_planner_replay` example through its reproducibility wrapper instead. + +## Run the control-plane replay + +From an ASAPQuery-backend checkout, with its normal sibling dependencies available: + +```sh +python3 tools/run-offline-planner-replay.py \ + --queries /path/to/ASAPPlanner/crates/frontend-promql/tests/observability/data/o11y_bench_promql.txt \ + --evidence /path/to/planner-evidence.json \ + --context /path/to/context.json \ + --output /tmp/control-plane-o11y.json +``` + +The backend uses its pinned planner dependency; a planner source checkout is +only needed here to locate the vendored query fixture. Omit evidence and context +together to run exact/default modes only. Supplying both adds empirical mode. +An optional `--planner /path/to/ASAPPlanner` enables a local source override for +development; it is not the default evaluation path. + +The [backend guide](../../control_plane/docs/offline-sketch-evidence.md) +documents the actual parser/binder and evidence assumptions. +The [benchmark driver](../../tools/empirical-bench/README.md) +and [artifact instructions](../../tools/empirical-bench/ARTIFACTS.md) +also live in the backend repository. + +## Interpret coverage conservatively + +- A parse failure is not a planner or backend execution result. +- A bound plan means the backend accepted a plan returned through its planner integration. +- `root_raw_fallback` means the complete original query remains selected; this is not summary acceleration. +- `raw_subtrees` includes source scans underneath summaries and cannot be counted as root fallback. +- `planning_elapsed_ns` measures the parser/binder path, not query execution. +- Binding does not establish deployable placement, successful data-plane execution, or resource savings. + +The recorded final evaluation bound all 27 fixtures in exact/default/empirical +modes; each mode retained five raw root fallbacks. Earlier standalone planner +candidate counts are historical diagnostics, not backend support coverage. +No collectors, storage services, query servers or upstream agent scoring run. +Point-frequency error measurements do not establish error bounds for these +PromQL queries. + +## Backend-only execution prototype + +The target is real data-plane execution, not just binding coverage: +OpenMetrics data and o11y queries enter ASAPQuery-backend; the backend calls +ASAPPlanner and executes either a valid ASAP plan or a local exact baseline. +No ASAPCollector or Prometheus process is required by the intended standalone +prototype. The implementation and its current limitations belong to +[backend PR #524](https://github.com/ProjectASAP/ASAPQuery-backend/pull/524). + +The supplied dataset is `/mydata/metrics.txt`, an OpenMetrics file with 305,026 +samples across 106 series. Preserve it read-only. The backend adapter owns +timestamp conversion and Remote Write encoding; OpenMetrics text cannot be +posted directly to `/api/v1/write`. The currently pinned upstream task corpus +has 28 query occurrences (24 unique queries); do not equate that corpus with the +older 27-query fixture used for the historical binding report. + +Acceptance requires matching exact/ASAP query results before reporting latency +or resource reductions, with identical data, labels and evaluation times. +Report construction/ingestion costs separately from steady-state queries. +Unsupported queries and exact fallbacks are not accelerated successes. +Do not substitute the earlier synthetic result-cache experiment for this test. + +## Optional backend diagnostic + +The small `offline_recommend` diagnostic also lives in ASAPQuery-backend. +It compares measured fixed-snapshot frequency scenarios, not PromQL execution: + +```sh +cargo run -p control_plane --example offline_recommend -- \ + /path/to/comparison-evidence.json /path/to/request.json +``` + +All benchmark producers and execution tools live in the backend repository. +These evaluation documents are maintained in ASAPQuery-backend. The public +evidence and resource contracts are reviewed separately in +[ASAPPlanner #357](https://github.com/ProjectASAP/ASAPPlanner/pull/357). diff --git a/tools/empirical-bench/.gitignore b/tools/empirical-bench/.gitignore new file mode 100644 index 00000000..81c34145 --- /dev/null +++ b/tools/empirical-bench/.gitignore @@ -0,0 +1,3 @@ +/results/ +/results-sweep/ +__pycache__/ diff --git a/tools/empirical-bench/ARTIFACTS.md b/tools/empirical-bench/ARTIFACTS.md new file mode 100644 index 00000000..4ec53c74 --- /dev/null +++ b/tools/empirical-bench/ARTIFACTS.md @@ -0,0 +1,32 @@ +# Archived experiment results + +Full raw measurements and expanded query plans are distributed as an +[experiment artifact](https://github.com/ProjectASAP/ASAPPlanner/releases/tag/offline-evidence-2026-09-08), +not checked into the tool source tree. This prerelease is an experiment archive, +not a product release. The concise conclusions remain in +[the historical report](../../docs/offline-o11y-final-2026-09-08.md). + +Download and verify from the repository root: + +```sh +gh release download offline-evidence-2026-09-08 --repo ProjectASAP/ASAPPlanner \ + --pattern offline-evidence-2026-09-08.tar.gz --dir /tmp +echo '180f3768203635c31188d19b500455290b96e19a4df4fdac7245af0944bd1b34 /tmp/offline-evidence-2026-09-08.tar.gz' | sha256sum --check +tar -xzf /tmp/offline-evidence-2026-09-08.tar.gz +``` + +Extract only into a checkout without existing result files; extraction restores +`tools/empirical-bench/results/` and `results-sweep/`. These generated directories +are ignored by Git. The archive is 460,710 bytes and contains 43 text files. +It preserves the original source snapshot +`152ca6cb71fc11a8c38d93e69c9fd7dca682f03b`, per-run commands, environment metadata, +raw timing/error reports, and measurement-source checksums. + +- `results/`: historical partial frequency measurements and initial planner/control-plane replay; also the later `o11y-exact-snapshot.json` reference evaluation. +- `results-sweep/`: final frequency matrix, disjoint CPU and heap evidence, six recommendation scenarios, four frequency binding reports, and final o11y control-plane replay. +- Each directory contains `MEASUREMENTS.md` explaining its measurement scope. Do not mix costs from the two runs. + +To produce new results instead of downloading these measurements, follow +[the benchmark guide](README.md) and [the replay guide](../../control_plane/docs/offline-sketch-evidence.md). +The archive preserves offline evidence, not production speedups or real-time +accuracy guarantees. diff --git a/tools/empirical-bench/README.md b/tools/empirical-bench/README.md new file mode 100644 index 00000000..cfb74ade --- /dev/null +++ b/tools/empirical-bench/README.md @@ -0,0 +1,117 @@ +# Offline sketch-bench evidence (developer guide) + +This driver runs ProjectASAP/sketch-bench's real `approxbench` CLI and adapts its +schema-v5 reports to the planner's versioned evidence schema. It measures CMS and +CountSketch using `asap_sketchlib` RegularPath/Vector2D plus an exact Polars +frequency baseline, on deterministic uniform and Zipf streams. No collector, +query backend, runtime ground truth, or runtime error feedback is involved. + +Full historical and final outputs are stored in the [experiment archive](ARTIFACTS.md), +not in this source tree. Download and verify that archive to inspect saved results. +The planner-side conclusions remain in the +[historical report](../../docs/offline-o11y-final-2026-09-08.md). +For control-plane integration and replay, see the +[backend evidence guide](../../control_plane/docs/offline-sketch-evidence.md). + +```bash +git clone https://github.com/ProjectASAP/sketch-bench.git /path/to/sketch-bench +git -C /path/to/sketch-bench checkout 87f619e843fd2e4da784160d4e205a0d0d55f032 +(cd /path/to/sketch-bench && cargo build --release --locked -p aqpbm-cli) +python3 tools/empirical-bench/run.py --bench-repo /path/to/sketch-bench --output /tmp/offline-evidence --sweep +python3 -m unittest discover -s tools/empirical-bench -p 'test_*.py' +``` + +Build from the sketch-bench directory so its `.cargo/config.toml` applies +`target-cpu=native`, as in the archived run. The driver records the source +revision, dirty status, executable checksum, full invocations, compiler, CPU, +OS, implementation, and parameters. It pins Polars to one worker. Five measured +runs follow two warmups per timed operation by default. Upstream accuracy executes +one offline truth comparison for the fixed generated dataset, recorded as +`error.trials=1`; it is run separately because accuracy of insert is undefined. +These timings are repeated observations +within one process, not independent experiments: standard deviations are reported, +but no confidence interval or cross-distribution guarantee is inferred. +`manifest.json.runtime_provenance` distinguishes observed host metadata and the +driver's enforced thread setting from declared build preconditions. The runtime +string's release/native/jemalloc settings describe the documented required build; +the driver does not independently recover compiler or allocator flags from the +executable. Its SHA-256 identifies that binary without proving those build flags. +The archived run used a shared development host without CPU affinity or an +exclusive core reservation; timing estimates should be remeasured on deployment +hardware before using them for capacity decisions. + +Outputs: + +- `raw.json`: unmodified per-invocation sketch-bench flat reports, including wall + times, CPU user/system samples, process RSS/allocator readings, and all errors. +- `manifest.json`: execution and environment provenance. +- `operation-reports.json`: original reports before upstream flattening; preserve + per-operation memory, including exact prepared-index state. +- `memory-probe.json`: requested-allocation samples with each live sketch, using + the same library and upstream data generator on identical inputs. +- `planner-evidence.json`: normalized sketch measurements, with unknowns null. +- `exact-baselines.json`: exact frequency index costs on the same input. +- `resource-probe.json`: disjoint live-state construction/update/read/merge CPU, + actual serialized snapshot size and allocated filesystem blocks. +- `comparison-evidence.json`: query-bound sketch records and exact baselines; + its required `disjoint_live_state_v1` timing contract excludes duplicate setup + and destruction charges. +- `request-*.json`: fixed-snapshot CPU-only comparisons at 1%/5% observed mean + relative error, plus an explicitly illustrative memory-weighted comparison. + +The default sketch parameters match formal planner sizing for epsilon=0.01, delta=0.01: +CMS width 272, depth 5; CountSketch width 30,000, depth 83. Input defaults to +20,000 i64 keys, key-space size 1,000, seed 42; Zipf exponent is 1.1. Actual +distinct-key counts come from the ground-truth probe population. Errors are +average absolute relative point-frequency errors over **all distinct keys**. +They are not error of a total COUNT, or a formal epsilon bound. Aggregate error +over repeated deterministic inputs does not establish a confidence interval. +`--sweep` additionally measures CMS widths 512/2720/4096/27200/32768 (depth 5) +and CountSketch widths 300/3000 (depth 83). The power-of-two CMS widths support +the backend frequency extension's actual parameter constraints. An offline +accuracy threshold does not authorize deployment below formal minimum sizes. + +Original upstream CPU reports use paired user+system samples and preserve wrapper +allocation/destruction overhead. The current comparison exporter instead uses +`resource_probe.rs`, linked to the exact release libraries and jemalloc. It times +live-state phases separately with `CLOCK_PROCESS_CPUTIME_ID`: empty constructor, +updates, exact prepare, reads, and sketch merge. Holder allocation and destruction +are outside the constructor timer; constructors/destructors are outside the other +phase timers. Queries cover every distinct key ten times against an unchanged +snapshot, in a reproducible rotated sorted order. The exact implementation is +the upstream public `PolarsFrequencyCore`, not a replacement algorithm. +The model's scope ends with state retained after reading; retirement is excluded +on both sides. Wall time remains in upstream raw reports and is never called CPU. + +`retained_bytes` and `peak_bytes` use the companion `memory_probe.rs`, linked to +the exact release libraries built by sketch-bench. Its counting System allocator +measures requested heap allocation bytes across construction and insertion, +keeping the sketch alive at the final snapshot; input generation is outside +the measurement. This is independent of allocator-resident pages and does not +measure jemalloc overhead. Five runs must release every sketch allocation after +destruction. CPU results come from a separate uninstrumented jemalloc probe. +The exact heap probe warms Polars twice, then measures construction/ingestion/ +prepare with the exact state alive. Its readings are accepted only when every +post-destruction allocation balance returns to zero; otherwise the exact footprint +stays explicitly formula based. `--export-only --refresh-memory` refreshes this +heap evidence without rerunning or replacing CPU timings. +Upstream counter-storage formula, process RSS and allocator readings remain in +raw reports only. The resource probe calls the actual `serialize_to_bytes` API +and verifies estimate-equivalent deserialization for every input key. It writes +one snapshot to a fresh local temporary file, flushes it, and records allocated +blocks times 512 separately from logical byte length; the file is then removed. +Directory/inode overhead and a runtime write schedule are outside this disk +measurement. State bytes are never substituted for disk usage. Validity is a +30-day reproducibility policy, not a statistical claim of future applicability. + +The exact baseline inserts the input into a buffer, then `prepare` executes +Polars group-by/count and creates a HashMap. Reads query that prepared exact +index. Compare `empty_build + (insert_per_item * N) + prepare + Q * read` with +sketch `empty_build + (update_per_item * N) + Q * read`. All components use +disjoint timed regions. A sketch can save state while losing CPU to this exact index. +These comparisons are offline point-frequency microbenchmarks; they do not +measure an o11y PromQL query or end-to-end deployed speedup. + +`results/` preserves the earlier partial run whose constructor/disk/serialization +were unknown. `results-sweep/` contains the complete frequency component run; +its additional probe must not retroactively change the earlier measurements. diff --git a/tools/empirical-bench/memory_probe.rs b/tools/empirical-bench/memory_probe.rs new file mode 100644 index 00000000..7564fbe4 --- /dev/null +++ b/tools/empirical-bench/memory_probe.rs @@ -0,0 +1,101 @@ +//! Counts requested allocation bytes with each sketch alive; input generation is excluded. +use std::alloc::{GlobalAlloc, Layout, System}; +use std::sync::atomic::{AtomicUsize, Ordering::Relaxed}; +use asap_sketchlib::{Count, CountMin, DataInput, RegularPath, Vector2D}; + +struct Tracking; +static LIVE: AtomicUsize = AtomicUsize::new(0); +static PEAK: AtomicUsize = AtomicUsize::new(0); +fn account(bytes: usize) { + let live = LIVE.fetch_add(bytes, Relaxed) + bytes; + PEAK.fetch_max(live, Relaxed); +} +unsafe impl GlobalAlloc for Tracking { + unsafe fn alloc(&self, layout: Layout) -> *mut u8 { + let ptr = unsafe { System.alloc(layout) }; + if !ptr.is_null() { account(layout.size()); } + ptr + } + unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 { + let ptr = unsafe { System.alloc_zeroed(layout) }; + if !ptr.is_null() { account(layout.size()); } + ptr + } + unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) { + unsafe { System.dealloc(ptr, layout) }; + LIVE.fetch_sub(layout.size(), Relaxed); + } + unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, size: usize) -> *mut u8 { + let next = unsafe { System.realloc(ptr, layout, size) }; + if !next.is_null() { + if size >= layout.size() { account(size - layout.size()); } + else { LIVE.fetch_sub(layout.size() - size, Relaxed); } + } + next + } +} +#[global_allocator] +static ALLOCATOR: Tracking = Tracking; + +fn main() { + let path = std::env::args().nth(1).expect("raw.json path"); + let rows: Vec = serde_json::from_slice(&std::fs::read(path).unwrap()).unwrap(); + let mut results = Vec::new(); + for row in rows { + let description: aqpbm_datagen::TableDescription = serde_json::from_value(row["workload"]["synthetic"]["description"].clone()).unwrap(); + let items = description.generate().unwrap().into_column(0).unwrap().into_i64().unwrap(); + if row["impl"] == "polars" { + use sketch_bench::wrappers::polars_shared::PolarsFrequencyCore; + // Initialize Polars' worker/cache state before attributing bytes to one live index. + for _ in 0..2 { + let mut warm = PolarsFrequencyCore::::default(); + for item in &items { warm.update(item); } + warm.finalize(); + } + let mut samples = Vec::with_capacity(5); + let mut cleanup_deltas = Vec::with_capacity(5); + for _ in 0..5 { + let baseline = LIVE.load(Relaxed); + PEAK.store(baseline, Relaxed); + let mut state = PolarsFrequencyCore::::default(); + for item in &items { state.update(item); } + state.finalize(); + std::hint::black_box(&state); + let sample = (LIVE.load(Relaxed).saturating_sub(baseline), PEAK.load(Relaxed).saturating_sub(baseline)); + drop(state); + cleanup_deltas.push(LIVE.load(Relaxed) as i64 - baseline as i64); + samples.push(sample); + } + results.push(serde_json::json!({"sketch": row["sketch"], "impl": "polars", + "workload": row["workload"], "sketch_config": row["sketch_config"], "samples": samples, + "cleanup_deltas": cleanup_deltas, + "method": "requested allocation bytes using counting System allocator; exact upstream PolarsFrequencyCore construction + insertion + prepare; state alive; input and two startup warmups excluded; valid only when cleanup deltas are zero; not allocator-resident pages"})); + continue; + } + let depth = row["sketch_config"]["params"]["rows"].as_u64().unwrap() as usize; + let width = row["sketch_config"]["params"]["cols"].as_u64().unwrap() as usize; + let cms = row["sketch"].as_str().unwrap().starts_with("cms-"); + let mut samples = Vec::with_capacity(5); + for _ in 0..5 { + let baseline = LIVE.load(Relaxed); + PEAK.store(baseline, Relaxed); + // The sketch remains alive at both snapshots, unlike a consuming benchmark closure. + let (retained, peak) = if cms { + let mut sketch = CountMin::, RegularPath>::with_dimensions(depth, width); + for item in &items { sketch.insert(&DataInput::I64(*item)); } + std::hint::black_box(&sketch); + (LIVE.load(Relaxed) - baseline, PEAK.load(Relaxed) - baseline) + } else { + let mut sketch = Count::, RegularPath>::with_dimensions(depth, width); + for item in &items { sketch.insert(&DataInput::I64(*item)); } + std::hint::black_box(&sketch); + (LIVE.load(Relaxed) - baseline, PEAK.load(Relaxed) - baseline) + }; + assert_eq!(LIVE.load(Relaxed), baseline, "sketch destruction must release its allocations"); + samples.push((retained, peak)); + } + results.push(serde_json::json!({"sketch": row["sketch"], "impl": "lib", "workload": row["workload"], "sketch_config": row["sketch_config"], + "samples": samples, "method": "requested allocation bytes using counting System allocator; construction + insertion; sketch alive; dataset excluded; not allocator-resident pages"})); + } + println!("{}", serde_json::to_string_pretty(&results).unwrap()); +} diff --git a/tools/empirical-bench/resource_probe.rs b/tools/empirical-bench/resource_probe.rs new file mode 100644 index 00000000..e1a17629 --- /dev/null +++ b/tools/empirical-bench/resource_probe.rs @@ -0,0 +1,138 @@ +//! Complements sketch-bench with empty construction CPU and actual persisted snapshot size. +use std::hint::black_box; +use std::io::Write; +use std::os::unix::fs::MetadataExt; +use asap_sketchlib::{Count, CountMin, DataInput, RegularPath, Vector2D}; + +#[global_allocator] +static ALLOCATOR: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc; + +fn cpu_ns() -> u64 { + #[repr(C)] + struct Timespec { tv_sec: std::os::raw::c_long, tv_nsec: std::os::raw::c_long } + extern "C" { fn clock_gettime(clock_id: std::os::raw::c_int, time: *mut Timespec) -> std::os::raw::c_int; } + // Linux CLOCK_PROCESS_CPUTIME_ID; this companion also uses Linux allocated blocks. + assert!(cfg!(target_os = "linux")); + let mut time = Timespec { tv_sec: 0, tv_nsec: 0 }; + assert_eq!(unsafe { clock_gettime(2, &mut time) }, 0); + time.tv_sec as u64 * 1_000_000_000 + time.tv_nsec as u64 +} + +fn construction(mut build: impl FnMut() -> T, batch: usize) -> Vec { + let mut samples = Vec::with_capacity(5); + for run in 0..7 { + let mut alive = Vec::with_capacity(batch); + let start = cpu_ns(); + for _ in 0..batch { alive.push(black_box(build())); } + let elapsed = cpu_ns() - start; + if run >= 2 { samples.push(elapsed as f64 / batch as f64); } + // Empty objects remain alive until after the timed construction batch. + black_box(&alive); + drop(alive); + } + samples +} + +fn live_phases(items: &[i64], build: impl Fn() -> T, insert: impl Fn(&mut T, i64), + read: impl Fn(&T, i64) -> f64, prepare: impl Fn(&mut T), + merge: Option) -> serde_json::Value { + let mut keys: Vec<_> = items.iter().copied().collect::>().into_iter().collect(); + // Fixed rotation avoids sorted probe order without introducing another RNG definition. + if !keys.is_empty() { let mid = keys.len() / 2; keys.rotate_left(mid); } + let mut updates = Vec::with_capacity(5); + let mut prepares = Vec::with_capacity(5); + let mut reads = Vec::with_capacity(5); + let mut merges = Vec::with_capacity(5); + for run in 0..7 { + let mut state = build(); + let start = cpu_ns(); + for item in items { insert(&mut state, black_box(*item)); } + let update = (cpu_ns() - start) as f64 / items.len() as f64; + let start = cpu_ns(); + prepare(&mut state); + let prepare = (cpu_ns() - start) as f64; + let start = cpu_ns(); + for _ in 0..10 { for key in &keys { black_box(read(&state, black_box(*key))); } } + let query = (cpu_ns() - start) as f64 / (keys.len() * 10) as f64; + let merge_time = merge.map(|merge| { + let mut other = build(); + for item in items { insert(&mut other, *item); } + let start = cpu_ns(); + merge(&mut state, &other); + (cpu_ns() - start) as f64 + }); + if run >= 2 { + updates.push(update); prepares.push(prepare); reads.push(query); + if let Some(time) = merge_time { merges.push(time); } + } + black_box(&state); + } + serde_json::json!({"update_cpu_ns_samples": updates, "prepare_cpu_ns_samples": prepares, + "read_cpu_ns_samples": reads, "merge_cpu_ns_samples": merges, + "query_distinct_keys": keys.len(), "query_passes": 10, + "method": "CLOCK_PROCESS_CPUTIME_ID; live state with constructor/destructor excluded; five runs after two warmups; update per input item; read per all-distinct-key lookup across ten repeated fixed-snapshot probe passes; merge per binary merge; fixed rotated sorted key order"}) +} + +fn main() { + let args: Vec = std::env::args().collect(); + let rows: Vec = serde_json::from_slice(&std::fs::read(&args[1]).unwrap()).unwrap(); + let directory = std::path::Path::new(&args[2]); + let mut results = Vec::new(); + for (index, row) in rows.into_iter().enumerate() { + let description: aqpbm_datagen::TableDescription = serde_json::from_value(row["workload"]["synthetic"]["description"].clone()).unwrap(); + let items = description.generate().unwrap().into_column(0).unwrap().into_i64().unwrap(); + if row["impl"] == "polars" { + use sketch_bench::wrappers::polars_shared::PolarsFrequencyCore; + let samples = construction(PolarsFrequencyCore::::default, 4096); + let phases = live_phases(&items, PolarsFrequencyCore::::default, + |s, x| s.update(&x), |s, x| s.query(&x) as f64, |s| s.finalize(), None); + results.push(serde_json::json!({"sketch": row["sketch"], "impl": row["impl"], + "workload": row["workload"], "sketch_config": row["sketch_config"], + "build_cpu_ns_samples": samples, "build_batch": 4096, + "phases": phases, + "build_method": "process CPU around upstream PolarsFrequencyCore::::default(); output holder allocation and destruction excluded"})); + continue; + } + let depth = row["sketch_config"]["params"]["rows"].as_u64().unwrap() as usize; + let width = row["sketch_config"]["params"]["cols"].as_u64().unwrap() as usize; + let batch = (16_000_000 / (depth * width * 4)).clamp(2, 256); + let (samples, serialized, phases) = if row["sketch"].as_str().unwrap().starts_with("cms-") { + let samples = construction(|| CountMin::, RegularPath>::with_dimensions(depth, width), batch); + let phases = live_phases(&items, || CountMin::, RegularPath>::with_dimensions(depth, width), + |s,x| s.insert(&DataInput::I64(x)), |s,x| s.estimate(&DataInput::I64(x)) as f64, |_| {}, Some(CountMin::, RegularPath>::merge)); + let mut sketch = CountMin::, RegularPath>::with_dimensions(depth, width); + for item in &items { sketch.insert(&DataInput::I64(*item)); } + let bytes = sketch.serialize_to_bytes().unwrap(); + let restored = CountMin::, RegularPath>::deserialize_from_bytes(&bytes).unwrap(); + for item in &items { assert_eq!(sketch.estimate(&DataInput::I64(*item)), restored.estimate(&DataInput::I64(*item))); } + (samples, bytes, phases) + } else { + let samples = construction(|| Count::, RegularPath>::with_dimensions(depth, width), batch); + let phases = live_phases(&items, || Count::, RegularPath>::with_dimensions(depth, width), + |s,x| s.insert(&DataInput::I64(x)), |s,x| s.estimate(&DataInput::I64(x)), |_| {}, Some(Count::, RegularPath>::merge)); + let mut sketch = Count::, RegularPath>::with_dimensions(depth, width); + for item in &items { sketch.insert(&DataInput::I64(*item)); } + let bytes = sketch.serialize_to_bytes().unwrap(); + let restored = Count::, RegularPath>::deserialize_from_bytes(&bytes).unwrap(); + for item in &items { assert_eq!(sketch.estimate(&DataInput::I64(*item)), restored.estimate(&DataInput::I64(*item))); } + (samples, bytes, phases) + }; + let path = directory.join(format!("snapshot-{index}.msgpack")); + let mut file = std::fs::OpenOptions::new().write(true).create_new(true).open(&path).unwrap(); + file.write_all(&serialized).unwrap(); + file.sync_all().unwrap(); + let metadata = file.metadata().unwrap(); + assert_eq!(metadata.len(), serialized.len() as u64); + let disk_bytes = metadata.blocks() * 512; + drop(file); + std::fs::remove_file(path).unwrap(); + results.push(serde_json::json!({"sketch": row["sketch"], "impl": row["impl"], + "workload": row["workload"], "sketch_config": row["sketch_config"], + "build_cpu_ns_samples": samples, "build_batch": batch, + "phases": phases, + "build_method": "CLOCK_PROCESS_CPUTIME_ID around empty sketch constructors in batched live objects; holder allocation and all destruction excluded; jemalloc; five runs after two warmups", + "serialized_bytes": serialized.len(), "disk_bytes": disk_bytes, + "disk_method": "actual flushed single MsgPack snapshot file allocated blocks*512 on local temporary filesystem; directory/inode overhead excluded; file removed after measurement; not runtime persistence demand"})); + } + println!("{}", serde_json::to_string_pretty(&results).unwrap()); +} diff --git a/tools/empirical-bench/run.py b/tools/empirical-bench/run.py new file mode 100644 index 00000000..cd56d756 --- /dev/null +++ b/tools/empirical-bench/run.py @@ -0,0 +1,367 @@ +#!/usr/bin/env python3 +"""Run the real sketch-bench CLI and export offline planner evidence (stdlib only).""" +import argparse +import datetime +import hashlib +import json +import math +import os +from pathlib import Path +import platform +import shlex +import statistics +import subprocess +import tempfile + + +BUILD_PROVENANCE = { + "runtime_field_status": "declared build preconditions, not introspected from the executable", + "compiler_status": "rustc --version observed on driver host; executable compiler not independently verified", + "required_build": "run cargo build --release --locked -p aqpbm-cli from the pinned sketch-bench directory with default features", + "declared_settings": "release; repository target-cpu=native; default jemalloc", + "verified_execution_setting": "driver sets POLARS_MAX_THREADS=1", + "binary_identity": "SHA-256 recorded; digest identifies executable but does not verify build settings", +} + + +def command(argv, **kwargs): + return subprocess.check_output(argv, text=True, **kwargs).strip() + + +def measurement(values): + if not values or any(not math.isfinite(x) or x < 0 for x in values): + raise ValueError("invalid or missing measurement") + return {"value": statistics.mean(values), + "stddev": statistics.stdev(values) if len(values) > 1 else None, + "samples": len(values)} + + +def cpu_per_op(row, op): + """Recover each run's work from paired rate and elapsed samples, as upstream does.""" + rate = row.get(op + ("_folds_per_sec" if op == "merge" else "_throughput_items_per_sec")) + cpu = row.get(op + "_cpu_time_ms") + wall = row.get(op + "_wall_time_ms") + if not rate or not cpu or not wall: + return None + samples = [rate["samples"], wall["samples"], cpu["user_ms"]["samples"], cpu["sys_ms"]["samples"]] + if len({len(x) for x in samples}) != 1: + raise ValueError("unaligned CPU/rate/time samples") + values = [] + for r, elapsed, user, system in zip(*samples): + work = r * elapsed / 1000 + if work <= 0: + raise ValueError("zero benchmark work") + values.append((user + system) * 1_000_000 / work) + result = measurement(values) + result["method"] = "mean of paired run (process user+system CPU ns)/(throughput*wall seconds); per " + ("binary merge" if op == "merge" else "input item" if op == "insert" else "point-frequency key lookup") + return result + + +def cpu_batch(row, op): + cpu = row.get(op + "_cpu_time_ms") + if not cpu: + return None + return measurement([(u + s) * 1_000_000 for u, s in + zip(cpu["user_ms"]["samples"], cpu["sys_ms"]["samples"])]) + + +def export(raw, manifest, operation_reports, memory_rows, resource_rows=None): + records, baselines = [], [] + if not (len(raw) == len(manifest["invocations"]) == len(operation_reports)): + raise ValueError("misaligned invocation reports") + for row, invocation, operations in zip(raw, manifest["invocations"], operation_reports): + if row["schema_version"] != 5: + raise ValueError("adapter requires sketch-bench schema 5") + accuracy = row["query_accuracy"] + resource = next((r for r in (resource_rows or []) if r["sketch"] == row["sketch"] + and r["impl"] == row["impl"] and r["sketch_config"] == row["sketch_config"] + and r["workload"] == row["workload"]), None) + if resource_rows is not None and (resource is None or "phases" not in resource): + raise ValueError("complete comparison requires matched disjoint live-state phase measurements for every record") + def phase_metric(phase): + return (dict(measurement(resource["phases"][phase + "_cpu_ns_samples"]), method=resource["phases"]["method"]) + if resource and resource.get("phases") else None) + shape = row["workload"]["synthetic"]["description"] + distribution = {"id": invocation["distribution_id"], + "family": invocation["distribution"], + "sample_count": shape["row_num"], + "distinct_count": int(accuracy["probes_all"]), + "parameters": shape["column_spec"][0]["distribution"]} + if invocation["algorithm"] == "Exact": + exact_memory = next((m for m in memory_rows if m.get("impl") == "polars" and m["workload"] == row["workload"]), None) + baselines.append({"distribution": distribution, "implementation": "polars exact group_by + HashMap", + "insert_cpu_ns_per_item": phase_metric("update") or cpu_per_op(row, "insert"), + "prepare_cpu_ns_per_dataset": phase_metric("prepare") or cpu_batch(row, "prepare"), + "read_cpu_ns_per_key": phase_metric("read") or cpu_per_op(row, "query"), + "retained_bytes": max(r["bench"]["memory_bytes"] for r in operations + if r["bench"].get("operation") == "query"), + "retained_bytes_method": "upstream prepared-query memory_bytes formula: buffer capacity plus HashMap capacity footprint; not measured allocated heap or peak", + "accuracy": accuracy}) + baselines[-1]["empty_build_cpu_ns"] = (dict(measurement(resource["build_cpu_ns_samples"]), method=resource["build_method"]) + if resource else None) + if exact_memory and all(delta == 0 for delta in exact_memory["cleanup_deltas"]): + baselines[-1]["retained_heap_bytes"] = dict(measurement([s[0] for s in exact_memory["samples"]]), method=exact_memory["method"]) + baselines[-1]["peak_heap_bytes"] = dict(measurement([s[1] for s in exact_memory["samples"]]), method=exact_memory["method"]) + continue + algorithm = invocation["algorithm"] + memory = next(m for m in memory_rows if m["sketch"] == row["sketch"] and m["workload"] == row["workload"] + and (m.get("sketch_config", row["sketch_config"]) == row["sketch_config"])) + params = row["sketch_config"]["params"] + timestamp = row["insert_timestamp"] + if not timestamp.endswith("Z"): + raise ValueError("expected upstream UTC timestamp") + measured = int(datetime.datetime.fromisoformat(timestamp[:19] + "+00:00").timestamp()) + environment = dict(manifest["environment"]) + environment["implementation"] = "asap_sketchlib RegularPath Vector2D" + # The algorithm distinguishes families; storage/hash path must also match. + environment["id"] = hashlib.sha256(json.dumps(environment, sort_keys=True).encode()).hexdigest()[:16] + records.append({"id": invocation["id"], "algorithm": algorithm, + "params": {algorithm: {"width": params["cols"], "depth": params["rows"]}}, + "distribution": distribution, "environment": environment, + "measured_at_unix_seconds": measured, + "valid_until_unix_seconds": measured + 30 * 86400, + "provenance": {"command": shlex.join(invocation["argv"]), + "dataset": "deterministic synthetic i64 frequency stream, seed 42", + "source_revision": manifest["source_revision"], + "repetitions": row["runs"]}, + "metrics": {"build_cpu_ns": (dict(measurement(resource["build_cpu_ns_samples"]), method=resource["build_method"]) + if resource else None), + "update_cpu_ns": phase_metric("update") or cpu_per_op(row, "insert"), + "merge_cpu_ns": phase_metric("merge") or cpu_per_op(row, "merge"), + "read_cpu_ns": phase_metric("read") or cpu_per_op(row, "query"), + "retained_bytes": dict(measurement([float(s[0]) for s in memory["samples"]]), method=memory["method"]), + "peak_bytes": dict(measurement([float(s[1]) for s in memory["samples"]]), method=memory["method"]), + "serialized_bytes": (dict(measurement([resource["serialized_bytes"]]), method="actual serialize_to_bytes MsgPack length; estimate-equivalent roundtrip verified for every input key") if resource else None), + "disk_bytes": (dict(measurement([resource["disk_bytes"]]), method=resource["disk_method"]) if resource else None)}, + "error": {"metric": "mean_absolute_relative_frequency_error_all_distinct_keys", + "mean": accuracy["are_all"], "max": None, + "trials": int(accuracy.get("accuracy_runs", 1)), + "ground_truth_method": "exact offline HashMap counts; every distinct key probed", + "query": {"kind": "point_frequency", "value_type": "i64", + "accuracy_stddev": accuracy.get("are_all_stddev"), + "full_accuracy": accuracy}}}) + return {"schema_version": 1, + "benchmark_version": "sketch-bench@" + manifest["source_revision"] + ("; adapter-v2-disjoint" if resource_rows else "; adapter-v1"), + "model_version": "empirical-update-cpu-v1", "records": records}, baselines + + +def comparison_artifact(artifact, baselines, manifest): + query = {"kind": "point_frequency", "value_type": "i64", "probe_set": "all_distinct_keys"} + exact_records = [] + for baseline in baselines: + if baseline["empty_build_cpu_ns"] is None: + continue + reference = next(r for r in artifact["records"] if r["distribution"] == baseline["distribution"]) + environment = dict(manifest["environment"]) + environment["implementation"] = "polars group_by + std HashMap" + environment["implementation_version"] = "polars 0.46.0" + environment["id"] = hashlib.sha256(json.dumps(environment, sort_keys=True).encode()).hexdigest()[:16] + invocation = next(i for i in manifest["invocations"] if i["algorithm"] == "Exact" + and i["distribution_id"] == baseline["distribution"]["id"]) + exact_records.append({"id": invocation["id"], "distribution": baseline["distribution"], + "environment": environment, "query": query, + "measured_at_unix_seconds": reference["measured_at_unix_seconds"], + "valid_until_unix_seconds": reference["valid_until_unix_seconds"], + "provenance": {"command": shlex.join(invocation["argv"]), + "dataset": "deterministic synthetic i64 frequency stream, seed 42; exact zero-error comparison verified offline", + "source_revision": manifest["source_revision"], "repetitions": 5}, + "metrics": {"empty_build_cpu_ns": baseline["empty_build_cpu_ns"], + "update_cpu_ns": baseline["insert_cpu_ns_per_item"], + "prepare_cpu_ns": baseline["prepare_cpu_ns_per_dataset"], + "read_cpu_ns": baseline["read_cpu_ns_per_key"], + "retained_bytes": baseline.get("retained_heap_bytes") or dict(measurement([baseline["retained_bytes"]]), method=baseline["retained_bytes_method"]), + "peak_bytes": baseline.get("peak_heap_bytes")}}) + return {"schema_version": 1, "timing_contract": "disjoint_live_state_v1", "sketch_evidence": artifact, + "query_bindings": [{"record_id": r["id"], "query": query} for r in artifact["records"]], + "exact_records": exact_records} + + +def refresh_memory(bench_repo, raw_path, memory_path): + deps = bench_repo.resolve() / "target/release/deps" + with tempfile.TemporaryDirectory(prefix="asap-memory-probe-") as temporary: + executable = str(Path(temporary) / "memory-probe") + argv = ["rustc", "--edition=2021", "-C", "opt-level=3", "-C", "panic=abort", "-C", "target-cpu=native", + "-L", f"dependency={deps}", str(Path(__file__).with_name("memory_probe.rs")), "-o", executable] + for name in ["asap_sketchlib", "aqpbm_datagen", "serde_json", "sketch_bench"]: + candidates = list(deps.glob(f"lib{name}-*.rlib")) + if len(candidates) != 1: + raise ValueError(f"expected one pinned release library for {name}, got {len(candidates)}") + argv += ["--extern", f"{name}={candidates[0]}"] + for pattern in ["*/out", "*/out/lib"]: + for native in (bench_repo.resolve() / "target/release/build").glob(pattern): + argv += ["-L", f"native={native}"] + subprocess.check_call(argv) + memory_path.write_text(command([executable, str(raw_path)], env=dict(os.environ, POLARS_MAX_THREADS="1")) + "\n") + return {"source_sha256": hashlib.sha256(Path(__file__).with_name("memory_probe.rs").read_bytes()).hexdigest(), + "compile_argv": argv, + "allocator": "System + requested-byte tracking; separate from uninstrumented jemalloc CPU measurements"} + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--bench-repo", required=True, type=Path) + parser.add_argument("--output", required=True, type=Path) + parser.add_argument("--size", type=int, default=20000) + parser.add_argument("--cardinality", type=int, default=1000) + parser.add_argument("--runs", type=int, default=5) + parser.add_argument("--sweep", action="store_true", help="measure three widths per frequency family") + parser.add_argument("--export-only", action="store_true") + parser.add_argument("--refresh-memory", action="store_true", help="with --export-only, remeasure heap without rerunning CPU timings") + parser.add_argument("--resume", action="store_true", help="continue a partially completed run with identical parameters") + args = parser.parse_args() + if args.refresh_memory and not args.export_only: + parser.error("--refresh-memory requires --export-only") + args.output.mkdir(parents=True, exist_ok=True) + manifest_path = args.output / "manifest.json" + raw_path = args.output / "raw.json" + operations_path = args.output / "operation-reports.json" + if args.export_only: + manifest, raw = json.loads(manifest_path.read_text()), json.loads(raw_path.read_text()) + operation_reports = json.loads(operations_path.read_text()) + else: + if (raw_path.exists() or manifest_path.exists()) and not args.resume: + parser.error("output already contains a run; use a new directory or --export-only") + repo = args.bench_repo.resolve() + if 'name = "asap_sketchlib"\nversion = "0.2.2"' not in (repo / "Cargo.lock").read_text(): + parser.error("adapter is validated with asap_sketchlib 0.2.2; do not relabel another version") + binary = repo / "target/release/approxbench" + cpu = next((line.split(":", 1)[1].strip() for line in Path("/proc/cpuinfo").read_text().splitlines() + if line.startswith("model name")), platform.processor()) + manifest = {"source_revision": command(["git", "rev-parse", "HEAD"], cwd=repo), + "source_dirty": bool(command(["git", "status", "--porcelain"], cwd=repo)), + "environment": {"cpu": cpu, "os": platform.platform(), + "runtime": command(["rustc", "--version"]) + "; release; target-cpu=native; jemalloc; POLARS_MAX_THREADS=1", + "implementation_version": "asap_sketchlib 0.2.2"}, + "binary_sha256": hashlib.sha256(binary.read_bytes()).hexdigest(), + "runtime_provenance": BUILD_PROVENANCE, + "invocations": []} + raw, operation_reports = [], [] + if args.resume: + previous = json.loads(manifest_path.read_text()) + if previous["binary_sha256"] != manifest["binary_sha256"] or previous["environment"] != manifest["environment"]: + parser.error("resume requires the same binary and environment") + manifest, raw = previous, json.loads(raw_path.read_text()) + operation_reports = json.loads(operations_path.read_text()) + for invocation in manifest["invocations"]: + argv = invocation["argv"] + if any(argv[argv.index(flag) + 1] != str(value) for flag, value in + [("--size", args.size), ("--cardinality", args.cardinality), ("--runs", args.runs)]): + parser.error("resume requires the same size, cardinality, and runs") + env = dict(os.environ, POLARS_MAX_THREADS="1") + for distribution in ["uniform", "zipf"]: + targets = [ + ("Cms", "cms-regularpath-vector2d", "lib", 5, 272), + ("CountSketch", "countsketch-regularpath-vector2d", "lib", 83, 30000), + ("Exact", "cms", "polars", 5, 272)] + if args.sweep: + targets += [("Cms", "cms-regularpath-vector2d", "lib", 5, width) for width in [2720, 27200, 512, 4096, 32768]] + targets += [("CountSketch", "countsketch-regularpath-vector2d", "lib", 83, width) for width in [300, 3000]] + for algorithm, variant, library, rows, cols in targets: + record_id = f"{algorithm.lower()}-{distribution}-seed42" + (f"-w{cols}-d{rows}" if args.sweep else "") + if any(i["id"] == record_id for i in manifest["invocations"]): + continue + argv = [str(binary), "sketchbench", "--variant", variant, "--library", library, + "--config", f"rows={rows} cols={cols}", "--dataset", distribution, + "--size", str(args.size), "--cardinality", str(args.cardinality), + "--dtype", "i64", "--seed", "42", "--runs", str(args.runs), + "--warmup-runs", "2", "--operations", + "insert,query" if algorithm == "Exact" else "insert,query,merge", + "--metrics", "throughput,cpu,memory"] + if distribution == "zipf": + argv += ["--zipf-s", "1.1"] + print(shlex.join(argv), flush=True) + output = command(argv, env=env) + accuracy_argv = list(argv) + accuracy_argv[accuracy_argv.index("--operations") + 1] = "query" + accuracy_argv[accuracy_argv.index("--metrics") + 1] = "accuracy" + output += "\n" + command(accuracy_argv, env=env) + prepare_argv = None + if algorithm == "Exact": + prepare_argv = list(argv) + prepare_argv[prepare_argv.index("--operations") + 1] = "prepare" + prepare_argv[prepare_argv.index("--metrics") + 1] = "latency,cpu,memory" + output += "\n" + command(prepare_argv, env=env) + operations = [json.loads(line) for line in output.splitlines() if line.strip()] + row = json.loads(command([str(binary), "flatten"], input=output, env=env)) + raw.append(row) + operation_reports.append(operations) + manifest["invocations"].append({"id": record_id, + "distribution_id": f"{distribution}-n{args.size}-c{args.cardinality}-seed42", + "distribution": distribution, "algorithm": algorithm, "argv": argv, + "accuracy_argv": accuracy_argv, "prepare_argv": prepare_argv}) + raw_path.write_text(json.dumps(raw, indent=2) + "\n") + manifest_path.write_text(json.dumps(manifest, indent=2) + "\n") + operations_path.write_text(json.dumps(operation_reports, indent=2) + "\n") + memory_path = args.output / "memory-probe.json" + resources_path = args.output / "resource-probe.json" + if not args.export_only: + # Older partial runs may lack prepare: CPU/memory alone select no upstream timing pass. + for index, (row, invocation) in enumerate(zip(raw, manifest["invocations"])): + if invocation["algorithm"] == "Exact" and not row.get("prepare_cpu_time_ms"): + prepare_argv = list(invocation["argv"]) + prepare_argv[prepare_argv.index("--operations") + 1] = "prepare" + prepare_argv[prepare_argv.index("--metrics") + 1] = "latency,cpu,memory" + output = command(prepare_argv, env=env) + operation_reports[index].extend(json.loads(line) for line in output.splitlines() if line.strip()) + raw[index] = json.loads(command([str(binary), "flatten"], + input="\n".join(json.dumps(r) for r in operation_reports[index]), env=env)) + invocation["prepare_argv"] = prepare_argv + raw_path.write_text(json.dumps(raw, indent=2) + "\n") + operations_path.write_text(json.dumps(operation_reports, indent=2) + "\n") + deps = args.bench_repo.resolve() / "target/release/deps" + manifest["memory_probe"] = refresh_memory(args.bench_repo, raw_path, memory_path) + with tempfile.TemporaryDirectory(prefix="asap-resource-probe-") as temporary: + executable = str(Path(temporary) / "resource-probe") + argv = ["rustc", "--edition=2021", "-C", "opt-level=3", "-C", "panic=abort", "-C", "target-cpu=native", + "-L", f"dependency={deps}", str(Path(__file__).with_name("resource_probe.rs")), "-o", executable] + for name in ["asap_sketchlib", "aqpbm_datagen", "serde_json", "tikv_jemallocator", "sketch_bench"]: + candidates = list(deps.glob(f"lib{name}-*.rlib")) + if len(candidates) != 1: + raise ValueError(f"expected one pinned release library for {name}, got {len(candidates)}") + argv += ["--extern", f"{name}={candidates[0]}"] + for native in (args.bench_repo.resolve() / "target/release/build").glob("tikv-jemalloc-sys-*/out/lib"): + argv += ["-L", f"native={native}"] + for native in (args.bench_repo.resolve() / "target/release/build").glob("*/out"): + argv += ["-L", f"native={native}"] + subprocess.check_call(argv) + resources_path.write_text(command([executable, str(raw_path), temporary], env=env) + "\n") + manifest["resource_probe"] = {"source_sha256": hashlib.sha256(Path(__file__).with_name("resource_probe.rs").read_bytes()).hexdigest(), + "compile_argv": argv, "allocator": "jemalloc", "disk_filesystem_path": tempfile.gettempdir()} + manifest_path.write_text(json.dumps(manifest, indent=2) + "\n") + if args.refresh_memory: + if hashlib.sha256((args.bench_repo.resolve() / "target/release/approxbench").read_bytes()).hexdigest() != manifest["binary_sha256"]: + parser.error("memory refresh requires the original benchmark binary") + manifest["memory_probe"] = refresh_memory(args.bench_repo, raw_path, memory_path) + manifest["runtime_provenance"] = BUILD_PROVENANCE + manifest_path.write_text(json.dumps(manifest, indent=2) + "\n") + artifact, baselines = export(raw, manifest, operation_reports, json.loads(memory_path.read_text()), + json.loads(resources_path.read_text()) if resources_path.exists() else None) + (args.output / "planner-evidence.json").write_text(json.dumps(artifact, indent=2) + "\n") + (args.output / "exact-baselines.json").write_text(json.dumps(baselines, indent=2) + "\n") + if resources_path.exists(): + comparison = comparison_artifact(artifact, baselines, manifest) + (args.output / "comparison-evidence.json").write_text(json.dumps(comparison, indent=2) + "\n") + for exact in comparison["exact_records"]: + row = next(r for r in artifact["records"] if r["distribution"] == exact["distribution"]) + for budget, suffix in [(0.01, "001"), (0.05, "005")]: + request = {"context": {"distribution": row["distribution"], "environment": row["environment"], + "now_unix_seconds": max(r["measured_at_unix_seconds"] for r in artifact["records"])}, + "exact_environment": exact["environment"], "query": exact["query"], + "accuracy": {"metric": row["error"]["metric"], "max_observed_mean": budget, "minimum_trials": 1}, + "workload": {"input_items_per_state": row["distribution"]["sample_count"], + "reads_per_state": 1000, "merges_per_state": 0, "state_instances": 1, "horizon_seconds": 300.0}, + "weights": {"cpu_ns_weight": 1.0, "retained_byte_seconds_weight": 0.0}, "formal_minimums": None} + (args.output / ("request-" + row["distribution"]["family"] + "-are" + suffix + ".json")).write_text(json.dumps(request, indent=2) + "\n") + request["accuracy"]["max_observed_mean"] = 0.01 + request["weights"]["retained_byte_seconds_weight"] = 0.01 + request["formal_minimums"] = [{"algorithm": "Cms", "params": {"Cms": {"width": 512, "depth": 5}}}] + (args.output / ("request-" + row["distribution"]["family"] + "-memory-weighted.json")).write_text(json.dumps(request, indent=2) + "\n") + for row in artifact["records"]: + context = {"distribution": row["distribution"], "environment": row["environment"], + "now_unix_seconds": max(r["measured_at_unix_seconds"] for r in artifact["records"])} + (args.output / ("context-" + row["distribution"]["family"] + ".json")).write_text(json.dumps(context, indent=2) + "\n") + print(f"Exported {len(artifact['records'])} sketch records and {len(baselines)} exact baselines.") + + +if __name__ == "__main__": + main() diff --git a/tools/empirical-bench/test_export.py b/tools/empirical-bench/test_export.py new file mode 100644 index 00000000..ab5389b5 --- /dev/null +++ b/tools/empirical-bench/test_export.py @@ -0,0 +1,51 @@ +"""Unit checks for measurement normalization, independent of running the benchmark.""" +import unittest + +from run import cpu_per_op, export + + +class NormalizationTests(unittest.TestCase): + def test_complete_comparison_rejects_missing_live_phase_measurements(self): + """An old constructor-only probe cannot label consuming-wrapper CPU as disjoint.""" + row = {"schema_version": 5, "query_accuracy": {}, "sketch": "cms", "impl": "lib", + "sketch_config": {}, "workload": {}} + manifest = {"invocations": [{}]} + for resources in [[], [dict(row, build_cpu_ns_samples=[1.0])]]: + with self.subTest(resources=resources): + with self.assertRaisesRegex(ValueError, "disjoint live-state"): + export([row], manifest, [[]], [], resources) + + def test_partial_report_files_are_rejected_before_pairing(self): + """A truncated report file must fail rather than silently lose invocations.""" + for raw, invocations, operations in [([{}], [], []), ([], [{}], []), ([], [], [{}])]: + with self.subTest(lengths=(len(raw), len(invocations), len(operations))): + with self.assertRaisesRegex(ValueError, "misaligned invocation reports"): + export(raw, {"invocations": invocations}, operations, []) + + def test_cpu_normalizes_paired_work_not_wall_time(self): + """Different wall durations for identical work must not change the denominator.""" + row = {"query_throughput_items_per_sec": {"samples": [2000, 1000]}, + "query_wall_time_ms": {"samples": [100, 200]}, + "query_cpu_time_ms": {"user_ms": {"samples": [20, 20]}, + "sys_ms": {"samples": [10, 10]}}} + result = cpu_per_op(row, "query") + self.assertEqual(result["value"], 150000) + self.assertEqual(result["stddev"], 0) + self.assertEqual(result["samples"], 2) + + def test_unmeasured_cpu_is_not_zero(self): + """Missing CPU evidence stays absent instead of looking like free work.""" + self.assertIsNone(cpu_per_op({}, "query")) + + def test_misaligned_samples_rejected(self): + """CPU and rates from different run populations cannot be paired.""" + row = {"insert_throughput_items_per_sec": {"samples": [2000, 1000]}, + "insert_wall_time_ms": {"samples": [100]}, + "insert_cpu_time_ms": {"user_ms": {"samples": [20, 20]}, + "sys_ms": {"samples": [10, 10]}}} + with self.assertRaises(ValueError): + cpu_per_op(row, "insert") + + +if __name__ == "__main__": + unittest.main() From c9aca98a03f8c3efab8a8c6784967788f95c05ca Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 8 Sep 2026 14:40:25 -0600 Subject: [PATCH 3/3] Align the newly merged planner test dependency with production --- data_plane/Cargo.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index 4528b962..1677d4e5 100644 --- a/data_plane/Cargo.toml +++ b/data_plane/Cargo.toml @@ -129,7 +129,7 @@ crc32fast = "1.4" # none of them. [dev-dependencies] -asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c3410d14865497758d212e4265ad25c782187de1" } +asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "abb2f20e27091ac0c60715dc42b9b59c75b3447c" } tempfile = "3.20.0" criterion = { version = "0.5", features = ["html_reports"] } tokio-tungstenite = "0.21"