From 9b6f89fabdcafcf700d0a228157bfee875e371fb Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Sun, 4 Oct 2026 21:38:02 +0000 Subject: [PATCH 1/2] feat(devtools): stage_pipeline examples planner-layering-2 and 4b Example 2 is the SQL workload over `flows` (distinct, entropy, L2), lowered with the SQL frontend; workload queries report their language. Example 4b is the Q49 crossover: Pattern B's hourly p99 every 10 min over 1,000 series sampled every second, on a deployment that does not keep raw data, where the ingestion-time tumbling KLL (B1) is selected. Co-Authored-By: Claude Opus 5.5 --- crates/devtools/src/bin/stage_pipeline.rs | 166 ++++++++++++++++++---- crates/devtools/tests/stage_pipeline.rs | 82 ++++++++++- 2 files changed, 222 insertions(+), 26 deletions(-) diff --git a/crates/devtools/src/bin/stage_pipeline.rs b/crates/devtools/src/bin/stage_pipeline.rs index f66845aa..1696a766 100644 --- a/crates/devtools/src/bin/stage_pipeline.rs +++ b/crates/devtools/src/bin/stage_pipeline.rs @@ -1,8 +1,10 @@ // cargo run -p asap-devtools --bin stage_pipeline -- \ // --example planner-layering-1 --max-candidates 128 --out planner-layering-example1.json -// (also planner-layering-3a and planner-layering-3b: #509 Example 3, -// Patterns A and B; planner-layering-4a: Example 4, Pattern A repeated -// monthly) +// (also planner-layering-2: #509 Example 2, three SQL statistics over +// `flows`; planner-layering-3a and planner-layering-3b: Example 3, Patterns A +// and B; planner-layering-4a: Example 4, Pattern A repeated monthly; +// planner-layering-4b: Example 4's Q49 crossover, Pattern B's hourly p99 +// every 10 min on a deployment that does not keep raw data) // cargo run -p asap-devtools --bin stage_pipeline -- \ // --promql "topk by (job) (10, rate(x[1m]))" --epsilon 0.01 --delta 0.001 --out run.json // @@ -28,7 +30,8 @@ // be built) or costlier. // // The deployment inputs are the built-in cost and accuracy models and the -// reference executor's capabilities (`asap_executor::capabilities`). +// reference executor's capabilities (`asap_executor::capabilities`; without +// raw data retention for planner-layering-4b). // // - deployment: the deployment inputs Stage 3 used (the executor's // capabilities, summarized; the cost model and its Stage3Calibration; @@ -49,14 +52,14 @@ use asap_logical_optimizer::pass1::logical_candidates::{ choice_index, combination_count, LocalLogicalCandidates, }; use asap_logical_optimizer::Realization; -use asap_plan_selection::PlanningModels; use asap_plan_selection::{ plan_stages, Selection, Sharing, COST_MODEL, COST_PER_SECOND, MAX_ENUMERATED_CANDIDATES, }; +use asap_plan_selection::{DeploymentCapabilities, PlanningModels}; use asap_types::ir::export::{ compile_logical_asap_workload, LogicalASAPDAG, LogicalASAPDAGDocument, }; -use asap_types::ir::schema::{GroupingStrategy, SketchAlgorithm}; +use asap_types::ir::schema::{DataType, Field, GroupingStrategy, Schema, SketchAlgorithm}; use asap_types::ir::schema_support::with_promql_series_identity; use asap_types::ir::{OperatorNode, QueryRoot}; use asap_types::types::AccuracyTarget; @@ -64,12 +67,13 @@ use asap_types::workload::{ AccuracyRequirement, BatchEntry, DataArrival, DataDistribution, DataWorkload, DurationMs, Evidence, EvidenceSource, LatencyRequirement, MetricType, PlanningWorkload, Predictability, Query, QueryLanguage, QueryRecurrence, QueryRequirements, QueryTimeScope, QueryWorkload, Rate, - RepeatedDemand, RepeatingEntry, RepetitionInterval, RootDemand, TimeSelection, TimestampMs, + RepeatedDemand, RepeatingEntry, RepetitionInterval, RootDemand, SqlDialect, TimeSelection, + TimestampMs, }; use serde_json::{json, Value}; const USAGE: &str = - "usage: stage_pipeline (--example planner-layering-{1,3a,3b,4a} | --promql ... \ + "usage: stage_pipeline (--example planner-layering-{1,2,3a,3b,4a,4b} | --promql ... \ [--epsilon --delta ] [--interval-ms ]) [--max-candidates ] --out "; fn main() { @@ -102,9 +106,11 @@ fn run(args: Vec) -> Result<(), String> { } let workload = match (example.as_deref(), queries.is_empty()) { (Some("planner-layering-1"), true) => planner_layering_example1(), + (Some("planner-layering-2"), true) => planner_layering_example2(), (Some("planner-layering-3a"), true) => planner_layering_example3a(), (Some("planner-layering-3b"), true) => planner_layering_example3b(), (Some("planner-layering-4a"), true) => planner_layering_example4a(), + (Some("planner-layering-4b"), true) => planner_layering_example4b(), (Some(other), true) => return Err(format!("unknown example {other}")), (None, false) => { let accuracy = match (epsilon, delta) { @@ -118,24 +124,49 @@ fn run(args: Vec) -> Result<(), String> { _ => return Err("give exactly one of --example or --promql".into()), }; let out = out.ok_or("--out is required")?; - let document = stage_pipeline(&workload, max_candidates)?; + let capabilities = DeploymentCapabilities { + // Example 4b's crossover needs a deployment that does not keep raw data. + raw_data_retained: example.as_deref() != Some("planner-layering-4b"), + ..asap_executor::capabilities() + }; + let document = stage_pipeline(&workload, &capabilities, max_candidates)?; let text = serde_json::to_string_pretty(&document).map_err(|e| e.to_string())? + "\n"; std::fs::write(&out, text).map_err(|e| format!("{out}: {e}")) } -fn stage_pipeline(workload: &PlanningWorkload, max_candidates: usize) -> Result { - // PromQL rows carry each series' full identity as a column: the row - // representation per-series state needs at runtime. - let roots = asap_frontend_promql::lower_promql_query_workload(workload, 0) - .map_err(|e| format!("lowering: {e}"))? - .into_iter() - .map(|root| match root { - QueryRoot::Operator(node) => with_promql_series_identity(&node) - .map(QueryRoot::Operator) - .map_err(|e| format!("series identity: {e}")), - scalar => Ok(scalar), - }) - .collect::, _>>()?; +fn stage_pipeline( + workload: &PlanningWorkload, + capabilities: &DeploymentCapabilities, + max_candidates: usize, +) -> Result { + let roots = match workload.query_workload.language { + // The only SQL workload here is Example 2's, over `flows`. + QueryLanguage::SQL(_) => tokio::runtime::Builder::new_current_thread() + .build() + .map_err(|e| e.to_string())? + .block_on(asap_frontend_sql::lower_sql_batch( + &workload.query_workload, + &flows_catalog(), + )) + .into_iter() + .map(|root| { + root.map(QueryRoot::Operator) + .map_err(|e| format!("lowering: {e}")) + }) + .collect::, _>>()?, + // PromQL rows carry each series' full identity as a column: the row + // representation per-series state needs at runtime. + _ => asap_frontend_promql::lower_promql_query_workload(workload, 0) + .map_err(|e| format!("lowering: {e}"))? + .into_iter() + .map(|root| match root { + QueryRoot::Operator(node) => with_promql_series_identity(&node) + .map(QueryRoot::Operator) + .map_err(|e| format!("series identity: {e}")), + scalar => Ok(scalar), + }) + .collect::, _>>()?, + }; let stage0 = export(&roots)?; let demand: Vec = workload .query_workload @@ -143,8 +174,7 @@ fn stage_pipeline(workload: &PlanningWorkload, max_candidates: usize) -> Result< .map(|entry| RootDemand::from(&entry)) .collect(); let data = workload.data_workload.clone().unwrap_or_default(); - let capabilities = asap_executor::capabilities(); - let models = PlanningModels::builtin().with_capabilities(&capabilities); + let models = PlanningModels::builtin().with_capabilities(capabilities); let run = plan_stages( roots.into_iter().enumerate().collect(), &demand, @@ -372,6 +402,10 @@ fn label(inventory: &LocalLogicalCandidates, owners: &[usize], choice: &[ } fn workload_queries(workload: &PlanningWorkload) -> Vec { + let language = match workload.query_workload.language { + QueryLanguage::SQL(_) => "sql", + _ => "promql", + }; workload .query_workload .entries() @@ -395,7 +429,7 @@ fn workload_queries(workload: &PlanningWorkload) -> Vec { } json!({ "id": format!("q{}", index + 1), - "language": "promql", + "language": language, "text": entry.query.0, "requirements": requirements, }) @@ -494,6 +528,67 @@ fn planner_layering_example1() -> PlanningWorkload { } } +/// #509 Example 2's `flows` table. +fn flows_catalog() -> asap_frontend_sql::SqlCatalog { + asap_frontend_sql::SqlCatalog::new().with_table( + "flows", + Schema::new(vec![ + Field::plain("ts", DataType::Timestamp, false), + Field::plain("src_ip", DataType::Utf8, false), + ]), + ) +} + +/// #509 Example 2: distinct count, entropy and L2 of `src_ip` over the last +/// minute, one-time, as in `planner_layering_example2.rs`. Q3 casts its +/// product to DOUBLE: the design's Int64 `c * c` can overflow, so the L2 rule +/// does not accept it. +fn planner_layering_example2() -> PlanningWorkload { + const WINDOW: &str = "WHERE ts >= now() - INTERVAL '1 minute'"; + let queries = [ + (format!("SELECT COUNT(DISTINCT src_ip) FROM flows {WINDOW}"), 0.02), + ( + format!("SELECT -SUM(p * LN(p)) FROM (SELECT COUNT(*) * 1.0 / SUM(COUNT(*)) OVER () AS p FROM flows {WINDOW} GROUP BY src_ip)"), + 0.05, + ), + ( + format!("SELECT SQRT(SUM(CAST(c AS DOUBLE) * CAST(c AS DOUBLE))) FROM (SELECT src_ip, COUNT(*) AS c FROM flows {WINDOW} GROUP BY src_ip)"), + 0.01, + ), + ]; + PlanningWorkload { + query_workload: QueryWorkload { + language: QueryLanguage::SQL(SqlDialect::DataFusionSQL), + query_batch: Some( + queries + .into_iter() + .map(|(query, epsilon)| BatchEntry { + query: Query(query), + requirements: QueryRequirements { + accuracy: AccuracyRequirement::Explicit(AccuracyTarget::EpsilonDelta { + epsilon, + delta: 0.01, + }), + ..Default::default() + }, + predictability: Default::default(), + invocations: 1, + execute_at: None, + time_selection: Default::default(), + }) + .collect(), + ), + repeating_queries: None, + }, + data_workload: Some(DataWorkload { + arrival: DataArrival::ContinuouslyIngesting, + ingestion_rate: declared(Rate(100_000.0)), + input_cardinality: declared(10_000_000), + ..Default::default() + }), + } +} + /// The shared data workload of #509 with `arrival`. fn shared_data_workload(arrival: DataArrival) -> DataWorkload { DataWorkload { @@ -631,3 +726,24 @@ fn planner_layering_example3b() -> PlanningWorkload { data_workload: Some(shared_data_workload(DataArrival::ContinuouslyIngesting)), } } + +/// #509 Example 4's Q49 crossover (`stage3_b_panes_smaller_than_their_raw_data_win`): +/// Pattern B's p99 over the last hour every 10 min, over 1,000 series sampled +/// every second. Run on a deployment that does not keep raw data, where the +/// 10-min KLL panes are smaller than the samples they cover. +fn planner_layering_example4b() -> PlanningWorkload { + let mut workload = planner_layering_example3b(); + let entry = &mut workload + .query_workload + .repeating_queries + .as_mut() + .expect("Pattern B repeats")[0]; + entry.query = Query("quantile_over_time(0.99, latency_ms[60m])".into()); + entry.demand = RepeatedDemand::FixedInterval(RepetitionInterval(600_000)); + entry.time_selection.lookback = Some(DurationMs(3_600_000)); + let data = workload.data_workload.as_mut().expect("Pattern B's data"); + data.data_ingestion_interval = declared(DurationMs(1_000)); + data.ingestion_rate = declared(Rate(1_000.0)); + data.input_cardinality = declared(1_000); + workload +} diff --git a/crates/devtools/tests/stage_pipeline.rs b/crates/devtools/tests/stage_pipeline.rs index 824df584..14fcd882 100644 --- a/crates/devtools/tests/stage_pipeline.rs +++ b/crates/devtools/tests/stage_pipeline.rs @@ -1,4 +1,4 @@ -//! `stage_pipeline` writes a valid four-stage viewer document for #509 Example 1. +//! `stage_pipeline` writes valid four-stage viewer documents for #509's examples. use std::collections::HashSet; use std::path::PathBuf; use std::process::Command; @@ -195,6 +195,86 @@ fn example4a_repeats_monthly_with_nothing_maintainable() { assert!(cost(&monthly) < cost(&once)); } +/// Every exported DAG in `document` validates with one root per query, and +/// the selected candidate is a Stage 2 candidate with a cost. +fn assert_valid_document(document: &Value, queries: usize) { + assert_eq!(document["format"], "asap-stage-pipeline/v1"); + assert_eq!( + document["workload"]["queries"].as_array().unwrap().len(), + queries + ); + validated_dag(&document["stage0_logical"]["dag"], queries); + for candidate in document["stage1_logical_asap"]["candidates"] + .as_array() + .unwrap() + { + validated_dag(&candidate["dag"], queries); + } + let physical = document["stage2_physical_asap"]["candidates"] + .as_array() + .unwrap(); + for candidate in physical { + let dag: PhysicalASAPDAG = serde_json::from_value(candidate["dag"].clone()).unwrap(); + assert_eq!(dag.roots.len(), queries); + PhysicalASAPDAGDocument::new(dag).validate().unwrap(); + } + let stage3 = &document["stage3_selection"]; + let selected = stage3["selected"].as_str().unwrap(); + assert!(physical.iter().any(|p| p["id"] == selected)); + assert!(stage3["costs"][selected]["total"].as_f64().is_some()); +} + +/// The selected Stage 2 candidate's label. +fn selected_label(document: &Value) -> String { + let selected = &document["stage3_selection"]["selected"]; + document["stage2_physical_asap"]["candidates"] + .as_array() + .unwrap() + .iter() + .find(|p| &p["id"] == selected) + .unwrap()["label"] + .as_str() + .unwrap() + .to_string() +} + +/// Example 2: the three SQL statistics over `flows` lower through the SQL +/// frontend, are reported as SQL, and plan. The built-in models have no +/// UnivMon accuracy model, so the selected plan is the exact one with the +/// shared input (recorded, not required by the design). +#[test] +fn example2_plans_the_sql_workload() { + let document = generate(&["--example", "planner-layering-2", "--max-candidates", "200"]); + assert_valid_document(&document, 3); + assert!(document["workload"]["queries"] + .as_array() + .unwrap() + .iter() + .all(|q| q["language"] == "sql")); + assert_eq!(document["stage1_logical_asap"]["capped"], false); + assert_eq!( + selected_label(&document), + "Q1 exact · Q2 exact (Count acc) · Q3 exact (Count acc) · shared summary" + ); +} + +/// Example 4b (Q49): on a deployment that does not keep raw data, the hourly +/// p99 every 10 min over 1,000 series sampled every second selects the +/// tumbling KLL maintained at ingestion time (B1), and the document says the +/// deployment keeps no raw data. +#[test] +fn example4b_selects_ingestion_time_panes_without_raw_data() { + let document = generate(&["--example", "planner-layering-4b"]); + assert_valid_document(&document, 1); + assert_eq!( + document["deployment"]["capabilities"]["raw_data_retained"], + false + ); + let label = selected_label(&document); + assert!(label.contains("ingestion time"), "{label}"); + assert!(label.contains("tumbling 10m panes"), "{label}"); +} + /// The document records the deployment inputs Stage 3 used: the executor's /// capabilities (Hydra summaries named by their kind), the cost model with /// its calibration, including the memory weight and raw-retention settings, From 727cf892933a6332c7f6ff9dfc4f2311472b202b Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Sun, 4 Oct 2026 21:46:57 +0000 Subject: [PATCH 2/2] fix(devtools): price every candidate; --max-candidates only limits the plans written stage_pipeline selected among the first --max-candidates combinations, so a cap below the combination count could select a costlier plan (Examples 3a and 4a selected P122 at 128 of 486; the cheapest is P365). Stage 3 now prices up to 4096 combinations, and the document carries the cheapest --max-candidates plans with their logical candidates, then invalid ones, and a shown_of section with the totals when not every plan is written. Co-Authored-By: Claude Opus 5.5 --- crates/devtools/src/bin/stage_pipeline.rs | 79 +++++++++++++++++++---- crates/devtools/tests/stage_pipeline.rs | 19 +++++- 2 files changed, 85 insertions(+), 13 deletions(-) diff --git a/crates/devtools/src/bin/stage_pipeline.rs b/crates/devtools/src/bin/stage_pipeline.rs index 1696a766..96f0bca2 100644 --- a/crates/devtools/src/bin/stage_pipeline.rs +++ b/crates/devtools/src/bin/stage_pipeline.rs @@ -17,7 +17,7 @@ // merges something, again with identical sub-DAGs shared ("· shared // input"), and, when the summary-capability rule applies, again with // one summary sized for its strictest consumer ("· shared summary"); in -// enumeration order and capped by `--max-candidates` (default 64). A +// enumeration order, only those with a written physical candidate. A // repeating query's mergeable alternatives also come in tumbling panes // (Pass 2's window-composition rule), e.g. "Q1 Kll · tumbling 1m panes"; // - stage2_physical_asap: per logical candidate, its physical candidates @@ -25,9 +25,15 @@ // down-closed set of summaries maintained at ingestion time, labeled // e.g. "· ingestion time: Kll ×5 panes"), no cost; // - stage3_selection: per-candidate costs, the selected candidate, and -// every other candidate as rejected (`valid: false`: inaccurate, over a -// latency bound, needing a capability the deployment lacks, or could not -// be built) or costlier. +// every other written candidate as rejected (`valid: false`: inaccurate, +// over a latency bound, needing a capability the deployment lacks, or +// could not be built) or costlier. +// - shown_of: present when not every plan is written, the totals. +// +// Stage 3 prices every candidate (up to PRICE_LIMIT combinations), so the +// selection does not depend on `--max-candidates`; that flag only limits +// how many plans the document carries (default 64): the cheapest first, +// then invalid ones in enumeration order. // // The deployment inputs are the built-in cost and accuracy models and the // reference executor's capabilities (`asap_executor::capabilities`; without @@ -46,6 +52,7 @@ // query; without them the queries are exact. `--interval-ms` is the source // cadence PromQL needs (default 15000). +use std::collections::HashSet; use std::rc::Rc; use asap_logical_optimizer::pass1::logical_candidates::{ @@ -134,6 +141,10 @@ fn run(args: Vec) -> Result<(), String> { std::fs::write(&out, text).map_err(|e| format!("{out}: {e}")) } +/// Most Stage 1 combinations built and priced; beyond it the selection is +/// over the first PRICE_LIMIT only, and the tool warns. +const PRICE_LIMIT: usize = 4096; + fn stage_pipeline( workload: &PlanningWorkload, capabilities: &DeploymentCapabilities, @@ -180,11 +191,34 @@ fn stage_pipeline( &demand, &data, models, - max_candidates.max(1), + PRICE_LIMIT, ) .map_err(|e| format!("planning: {e}"))?; let enumeration = run.enumeration.expect("display was requested"); let combinations = enumeration.combinations; + if combinations > PRICE_LIMIT { + eprintln!("stage_pipeline: priced the first {PRICE_LIMIT} of {combinations} combinations"); + } + let selection = &enumeration.selection; + // Every plan id, physical candidates then those that could not be + // built; the cheapest `max_candidates` are written. + let physical_ids: Vec<&str> = enumeration + .candidates + .iter() + .flat_map(|c| c.physical.iter().map(|p| p.id.as_str())) + .collect(); + let mut ranked = physical_ids.clone(); + let known: HashSet<&str> = ranked.iter().copied().collect(); + ranked.extend( + selection + .rejected + .iter() + .map(|r| r.id.as_str()) + .filter(|id| !known.contains(id)), + ); + let total = |id: &str| selection.costs.get(id).map_or(f64::INFINITY, |c| c.total); + ranked.sort_by(|a, b| total(a).total_cmp(&total(b))); + let shown: HashSet<&str> = ranked.iter().take(max_candidates.max(1)).copied().collect(); let mut candidates = Vec::new(); let mut stage2 = Vec::new(); for candidate in &enumeration.candidates { @@ -209,12 +243,24 @@ fn stage_pipeline( Sharing::IdenticalExpressions => " · shared input", Sharing::SummaryCapability => " · shared summary", }; + let written = candidate + .physical + .iter() + .any(|p| shown.contains(p.id.as_str())) + || shown.contains(format!("P{index}").as_str()); + if !written { + continue; + } if let Some(logical) = &candidate.logical { let roots: Vec<_> = logical.iter().map(|(_, root)| root.clone()).collect(); candidates .push(json!({ "id": format!("L{index}"), "label": label, "dag": export(&roots)? })); } - for p in &candidate.physical { + for p in candidate + .physical + .iter() + .filter(|p| shown.contains(p.id.as_str())) + { let label = match p.materialization.as_str() { "" => label.clone(), m => format!("{label} · {m}"), @@ -224,19 +270,27 @@ fn stage_pipeline( ); } } - Ok(json!({ + let mut document = json!({ "format": "asap-stage-pipeline/v1", "workload": { "queries": workload_queries(workload) }, "deployment": deployment_json(&models), "stage0_logical": { "dag": stage0 }, "stage1_logical_asap": { "combinations": combinations, - "capped": combinations > max_candidates, + "capped": shown.len() < ranked.len(), "candidates": candidates, }, "stage2_physical_asap": { "candidates": stage2 }, - "stage3_selection": stage3_json(&enumeration.selection), - })) + "stage3_selection": stage3_json(selection, &shown), + }); + if shown.len() < ranked.len() { + document["shown_of"] = json!({ + "logical": enumeration.candidates.len(), + "physical": physical_ids.len(), + "priced": selection.costs.len(), + }); + } + Ok(document) } /// The deployment inputs Stage 3 used: the executor's capabilities, one @@ -280,10 +334,12 @@ fn deployment_json(models: &PlanningModels<'_>) -> Value { }) } -fn stage3_json(selection: &Selection) -> Value { +/// Stage 3's outcome for the written plans `shown`. +fn stage3_json(selection: &Selection, shown: &HashSet<&str>) -> Value { let costs: serde_json::Map<_, _> = selection .costs .iter() + .filter(|(id, _)| shown.contains(id.as_str())) .map(|(id, cost)| { let per_node: serde_json::Map<_, _> = cost .per_node @@ -304,6 +360,7 @@ fn stage3_json(selection: &Selection) -> Value { let rejected: Vec<_> = selection .rejected .iter() + .filter(|r| shown.contains(r.id.as_str())) .map(|r| json!({ "id": r.id, "valid": r.valid, "reason": r.reason })) .collect(); json!({ "costs": costs, "selected": selection.selected, "rejected": rejected }) diff --git a/crates/devtools/tests/stage_pipeline.rs b/crates/devtools/tests/stage_pipeline.rs index 14fcd882..6dc4edec 100644 --- a/crates/devtools/tests/stage_pipeline.rs +++ b/crates/devtools/tests/stage_pipeline.rs @@ -105,14 +105,29 @@ fn example1_document_is_valid_and_committed_fixture_is_current() { assert_eq!(accounted, all); } -/// The Cartesian product is cut at `--max-candidates` and says so. +/// `--max-candidates` limits the plans written, cheapest first, and says +/// so; Stage 3 still prices every plan, so the selection does not change. #[test] fn candidate_cap_is_recorded() { let document = generate(&["--example", "planner-layering-1", "--max-candidates", "5"]); + let full = generate(&EXAMPLE1); let stage1 = &document["stage1_logical_asap"]; assert_eq!(stage1["capped"], true); - assert_eq!(stage1["candidates"].as_array().unwrap().len(), 5); + assert!(stage1["candidates"].as_array().unwrap().len() <= 5); assert!(stage1["combinations"].as_u64().unwrap() > 5); + let physical = document["stage2_physical_asap"]["candidates"] + .as_array() + .unwrap(); + assert_eq!(physical.len(), 5); + let selection = &document["stage3_selection"]; + assert_eq!(selection["selected"], full["stage3_selection"]["selected"]); + let costs = selection["costs"].as_object().unwrap(); + assert_eq!(costs.len(), 5, "the five cheapest plans are written"); + let full_physical = full["stage2_physical_asap"]["candidates"] + .as_array() + .unwrap(); + assert_eq!(document["shown_of"]["physical"], full_physical.len()); + assert!(full.get("shown_of").is_none()); } /// Example 3, Pattern B: the 5-min p99 every minute offers KLL and DDSketch