diff --git a/crates/devtools/src/bin/stage_pipeline.rs b/crates/devtools/src/bin/stage_pipeline.rs index c294ab555..3f127e70e 100644 --- a/crates/devtools/src/bin/stage_pipeline.rs +++ b/crates/devtools/src/bin/stage_pipeline.rs @@ -135,7 +135,17 @@ // the last step. Each choice has its `id` (`E1`, ...), `choice`, `dag` // (`memo::enumeration::compose`; null with `error` when it does not // compose), `budget_share_1` (every chosen candidate sized for the whole -// budget) and `selected` (it is `selected_choice`). `priced` is null. +// budget) and `selected` (it is `selected_choice`). With +// `--price-enumerated ` (off by default), each of the first `n` listed +// choices is also run through Stage 2 and Stage 3 as the selected plan is +// (`memo_pricing::price_choices`, under the run's models): `valid`, +// `rejection` (a Stage 3 rejection cause, or `compose_failed`, +// `invalid_structure`, `stage2_failed`, `stage3_failed`, `rejected`) with +// `rejection_reason`, +// `cost_per_second` (its cheapest valid physical plan's) and +// `cheaper_than_selected` (valid and cheaper than the selected plan); and +// `priced` is `{count, cheaper_than_selected, selected_cost_per_second}`. +// Without it `priced` is null. // `workload.data_workload` is the `DataWorkload` the run planned with, as // declared (`--workload`, `--example`) or built from `--sample-interval-ms` // (`--promql`); each query also carries its `recurrence` and @@ -245,6 +255,7 @@ use std::collections::{BTreeSet, HashSet}; use std::rc::Rc; use asap_devtools::candidate_labels::{candidate_labels, computation_label, query_of_each_target}; +use asap_devtools::memo_pricing::{price_choices, SelectedBaseline}; use asap_devtools::planner_layering_examples::{ example1, example2, pattern_a, pattern_b, PatternARecurrence, SHARED_ROWS_PER_SECOND, }; @@ -283,7 +294,7 @@ const USAGE: &str = [--cost-model usage-over-time|aws-fargate-us-east-1|provisioned-capacity|cluster-capacity [--cluster-memory-bytes \ --cluster-disk-bytes [--cluster-cpu-cores ] --cluster-objective lowest-latency|least-memory|pareto-front]] \ [--plan-selection-settings ] [--enforce-accuracy-guarantees] [--deployment-capabilities ] [--exhaustive [--max-combinations ] | --search candidate-connections] [--milp-time-limit-s ] [--max-displayed ] \ -[--no-memo] [--max-memo-candidates ] [--memo-enumeration-limit ] [--max-enumerated ] --out \n\ +[--no-memo] [--max-memo-candidates ] [--memo-enumeration-limit ] [--max-enumerated ] [--price-enumerated ] --out \n\ --search candidate-connections is the default search."; /// Default evaluator-assignment budget for an exhaustive archive run. @@ -382,6 +393,7 @@ fn run(args: Vec) -> Result<(), Failure> { candidates_per_group: 64, enumeration_limit: 1000, enumerated: 200, + choices_to_price: 0, }; let mut args = args.into_iter(); while let Some(flag) = args.next() { @@ -454,6 +466,9 @@ fn run(args: Vec) -> Result<(), Failure> { "--max-enumerated" => { memo_caps.enumerated = value()?.parse().map_err(|e| format!("{e}"))? } + "--price-enumerated" => { + memo_caps.choices_to_price = value()?.parse().map_err(|e| format!("{e}"))? + } "--search" => { search_algorithm = Some(match value()?.as_str() { "candidate-connections" => SearchAlgorithm::CandidateConnections, @@ -576,6 +591,23 @@ fn run(args: Vec) -> Result<(), Failure> { }, }; plan_selection.ignore_accuracy_guarantees = !enforce_accuracy_guarantees; + if memo_caps.choices_to_price > 0 { + if !write_memo { + return Err("--price-enumerated needs the memo (drop --no-memo)".into()); + } + // The exhaustive search reports no calibration factor to price each + // choice on the selection's scale. + if exhaustive + && plan_selection.frozen_calibration_factor.is_none() + && !plan_selection.measured_operation_costs.operators.is_empty() + { + return Err( + "--price-enumerated with --exhaustive needs a frozen_calibration_factor \ + when measured operation costs are given" + .into(), + ); + } + } let from_files = if workload_paths.is_empty() { None } else if example.is_none() && queries.is_empty() { @@ -743,7 +775,20 @@ fn run(args: Vec) -> Result<(), Failure> { .ok() .and_then(|run| run.memo_steps.as_ref()) .filter(|_| write_memo) - .map(|steps| memo_json(steps, &memo_caps)); + .map(|steps| { + let pricing = Pricing { + stage0_roots: &roots, + entries: &entries, + data: &data, + models, + selected: result + .as_ref() + .ok() + .and_then(|run| run.stage3_plan_selection.as_ref().ok()) + .map(SelectedBaseline::of), + }; + memo_json(steps, &memo_caps, &pricing) + }); let mut document = document_of(described, &roots, &entries, result, max_displayed)?; if let Some(memo) = memo { document["memo"] = memo; @@ -1447,18 +1492,30 @@ fn query_batch( } /// How much of the memo the document writes (`--max-memo-candidates`, -/// `--memo-enumeration-limit`, `--max-enumerated`). +/// `--memo-enumeration-limit`, `--max-enumerated`) and how many listed +/// choices it prices (`--price-enumerated`). struct MemoCaps { candidates_per_group: usize, enumeration_limit: usize, enumerated: usize, + choices_to_price: usize, +} + +/// What pricing an enumerated choice needs: the run's workload, models and +/// selection. +struct Pricing<'a> { + stage0_roots: &'a [QueryRoot], + entries: &'a [QueryWorkloadEntry], + data: &'a DataWorkload, + models: PlanningModels<'a>, + selected: Option, } /// The document's `memo` section: the memo after each step, the selected /// plan's choice over the last step's memo, and the candidate DAGs each /// step's memo stands for (`memo::enumeration::enumerate`; each listed /// choice composed by `memo::enumeration::compose`). -fn memo_json(steps: &MemoSteps, caps: &MemoCaps) -> Value { +fn memo_json(steps: &MemoSteps, caps: &MemoCaps, pricing: &Pricing<'_>) -> Value { let named = [ ( "stage0", @@ -1506,8 +1563,19 @@ fn memo_json(steps: &MemoSteps, caps: &MemoCaps) -> Value { let last = &steps.concrete_states; let (step, memo) = listed.unwrap_or(("concrete_states", last)); let selected = steps.selected_choice.as_ref().ok(); - let choices: Vec<_> = enumerate(memo, caps.enumerated) - .choices + let enumerated = enumerate(memo, caps.enumerated).choices; + let priced = (caps.choices_to_price > 0).then(|| { + price_choices( + memo, + &enumerated[..caps.choices_to_price.min(enumerated.len())], + pricing.stage0_roots, + pricing.entries, + pricing.data, + pricing.models, + pricing.selected, + ) + }); + let mut choices: Vec<_> = enumerated .iter() .enumerate() .map(|(index, choice)| { @@ -1535,6 +1603,20 @@ fn memo_json(steps: &MemoSteps, caps: &MemoCaps) -> Value { entry }) .collect(); + for (entry, priced) in choices.iter_mut().zip(priced.iter().flatten()) { + entry["valid"] = json!(priced.valid); + entry["rejection"] = json!(priced.rejection); + entry["rejection_reason"] = json!(priced.rejection_reason); + entry["cost_per_second"] = json!(priced.cost_per_second); + entry["cheaper_than_selected"] = json!(priced.cheaper_than_selected); + } + let priced = priced.map_or(Value::Null, |priced| { + json!({ + "count": priced.len(), + "cheaper_than_selected": priced.iter().filter(|p| p.cheaper_than_selected).count(), + "selected_cost_per_second": pricing.selected.and_then(|s| s.cost_per_second), + }) + }); let mut memo = json!({ "steps": json_steps, "selected_choice": selected.map_or(Value::Null, choice_json), @@ -1543,7 +1625,7 @@ fn memo_json(steps: &MemoSteps, caps: &MemoCaps) -> Value { "per_step": per_step, "limit": caps.enumeration_limit, "choices": choices, - "priced": null, + "priced": priced, }, }); if let Err(reason) = &steps.selected_choice { diff --git a/crates/devtools/src/lib.rs b/crates/devtools/src/lib.rs index fb66cbff2..65512c128 100644 --- a/crates/devtools/src/lib.rs +++ b/crates/devtools/src/lib.rs @@ -21,6 +21,7 @@ //! missed reuse and sharing behind the `planner_fuzz` bin. pub mod candidate_labels; +pub mod memo_pricing; pub mod planner_fuzz; pub mod planner_layering_examples; pub mod planning_stages; diff --git a/crates/devtools/src/memo_pricing.rs b/crates/devtools/src/memo_pricing.rs new file mode 100644 index 000000000..aa1e9e27d --- /dev/null +++ b/crates/devtools/src/memo_pricing.rs @@ -0,0 +1,228 @@ +//! Price a memo's enumerated choices as the selected plan is priced: each +//! choice's composed DAG (`memo::enumeration::compose`) goes through Stage 2 +//! (`optimize_physical_asap`, every materialization assignment) and Stage 3 +//! (`select_plan`, under the run's models), so a choice cheaper than the +//! selected plan shows that the search missed a plan the memo stands for. + +use std::collections::HashMap; +use std::rc::Rc; + +use asap_logical_asap_optimization::memo::candidates::CandidateMemo; +use asap_logical_asap_optimization::memo::enumeration::{compose, MemoChoice}; +use asap_logical_asap_optimization::memo::MemoGroupId; +use asap_physical_asap_optimization::optimize_physical_asap; +use asap_plan_selection::{select_plan, PlanningModels, Selection, SelectionError}; +use asap_types::ir::{OperatorNode, QueryRoot}; +use asap_types::workload::{DataWorkload, QueryWorkloadEntry}; + +/// One choice's result: valid with its cost, or why it is not valid. +#[derive(Debug, Clone, PartialEq)] +pub struct PricedChoice { + /// Stage 2 implemented it and Stage 3 accepted one of its physical plans. + pub valid: bool, + /// Why it is not valid: the Stage 3 rejection cause of its first + /// physical plan in snake_case (e.g. `accuracy_target_missed`), or + /// `compose_failed`, `invalid_structure`, `stage2_failed`, + /// `stage3_failed`, or `rejected` (no plan and no rejection listed). + pub rejection: Option, + /// The rejection's one-line reason, with the numbers. + pub rejection_reason: Option, + /// The cheapest cost per second of its valid physical plans under the + /// run's cost model; `None` when not valid or under cost model 3. + pub cost_per_second: Option, + /// Valid and costs less than the selected plan (beyond rounding). + pub cheaper_than_selected: bool, +} + +/// What the choices are compared against: the run's selection. +#[derive(Debug, Clone, Copy, PartialEq)] +pub struct SelectedBaseline { + /// The selected plan's cost per second. + pub cost_per_second: Option, + /// The calibration factor the run priced with, frozen for every choice + /// so each is priced on the same scale as the selected plan. + pub calibration_factor: Option, +} + +impl SelectedBaseline { + /// The baseline of `selection`: its selected plan's cost and the + /// calibration factor its search reported. + pub fn of(selection: &Selection) -> Self { + Self { + cost_per_second: selection.costs[&selection.selected].cost_per_second, + calibration_factor: selection + .search + .candidate_connections + .as_ref() + .map(|report| report.calibration_factor), + } + } +} + +/// Costs within this fraction of each other are the same sum of charges +/// rounded differently (Stage 3's relative tie tolerance). +const TIE_TOLERANCE: f64 = 1e-9; + +/// Each of `choices` of `memo`, composed, implemented and priced. Query `i` +/// is `stage0_roots[i]` with workload entry `entries[i]`; its operator +/// references are replaced by their root groups' composed DAGs. +pub fn price_choices( + memo: &CandidateMemo, + choices: &[MemoChoice], + stage0_roots: &[QueryRoot], + entries: &[QueryWorkloadEntry], + data: &DataWorkload, + models: PlanningModels<'_>, + selected: Option, +) -> Vec { + let mut settings = models.plan_selection.clone(); + if settings.frozen_calibration_factor.is_none() { + settings.frozen_calibration_factor = selected.and_then(|s| s.calibration_factor); + } + let models = models.with_plan_selection(&settings); + let selected_cost = selected.and_then(|s| s.cost_per_second); + // Which root group each query's operator references name is the same + // for every choice. + let root_groups = root_groups_of_queries(memo, stage0_roots); + choices + .iter() + .enumerate() + .map(|(index, choice)| { + let roots = compose(memo, choice) + .map_err(|error| error.to_string()) + .and_then(|composed| query_roots(&composed, stage0_roots, &root_groups)); + let mut priced = price_choice(roots, index, entries, data, models); + priced.cheaper_than_selected = match (priced.cost_per_second, selected_cost) { + (Some(cost), Some(selected)) => cost < selected - TIE_TOLERANCE * selected.abs(), + _ => false, + }; + priced + }) + .collect() +} + +/// One choice, its query roots composed (or why they do not compose), +/// implemented and priced. +fn price_choice( + roots: Result, String>, + index: usize, + entries: &[QueryWorkloadEntry], + data: &DataWorkload, + models: PlanningModels<'_>, +) -> PricedChoice { + let invalid = |cause: &str, reason: String| PricedChoice { + valid: false, + rejection: Some(cause.into()), + rejection_reason: Some(reason), + cost_per_second: None, + cheaper_than_selected: false, + }; + let roots = match roots { + Ok(roots) => roots, + Err(error) => return invalid("compose_failed", error), + }; + if let Err(error) = roots.iter().try_for_each(QueryRoot::validate_structure) { + return invalid("invalid_structure", error.to_string()); + } + let id = format!("E{}", index + 1); + let mut physical = + match optimize_physical_asap(&id, &roots, entries, data, models.deployment_capabilities) { + Ok(physical) if !physical.is_empty() => physical, + Ok(_) => return invalid("stage2_failed", "no physical plan".into()), + Err(error) => return invalid("stage2_failed", error.to_string()), + }; + for (k, plan) in physical.iter_mut().enumerate() { + plan.id = format!("{id}-{k}"); + } + match select_plan(&physical, entries, data, models) { + Ok(selection) => PricedChoice { + valid: true, + rejection: None, + rejection_reason: None, + cost_per_second: selection.costs[&selection.selected].cost_per_second, + cheaper_than_selected: false, + }, + Err(SelectionError::NoValidCandidate(rejected)) => match rejected.first() { + Some(first) => invalid( + serde_json::to_value(first.cause) + .ok() + .and_then(|v| v.as_str().map(str::to_owned)) + .as_deref() + .unwrap_or("rejected"), + first.reason.clone(), + ), + None => invalid("rejected", "no valid physical plan".into()), + }, + Err(error) => invalid("stage3_failed", error.to_string()), + } +} + +/// For each query, the root group of each of its operator references, in +/// `operator_refs` order: the one group whose `roots_of` names the query when +/// its root is one operator, else the group whose node is the reference. +fn root_groups_of_queries( + memo: &CandidateMemo, + stage0_roots: &[QueryRoot], +) -> Result>, String> { + stage0_roots + .iter() + .enumerate() + .map(|(query, root)| { + let groups: Vec<_> = (0..memo.groups.len()) + .filter(|&g| memo.groups[g].roots_of.contains(&query)) + .map(|g| MemoGroupId(g as u32)) + .collect(); + match root { + QueryRoot::Operator(_) if groups.len() == 1 => Ok(groups), + QueryRoot::Operator(node) => Ok(vec![group_of(memo, &groups, node, query)?]), + QueryRoot::Scalar(expression) => expression + .operator_refs() + .into_iter() + .map(|node| group_of(memo, &groups, node, query)) + .collect(), + } + }) + .collect() +} + +/// The group among `groups` whose node is `node`: the same node, else an +/// equal one (the memo's group nodes are the frontends' nodes with common +/// sub-DAGs shared). +fn group_of( + memo: &CandidateMemo, + groups: &[MemoGroupId], + node: &Rc, + query: usize, +) -> Result { + let node_of = |g: &&MemoGroupId| &memo.groups[g.0 as usize].node; + groups + .iter() + .find(|g| Rc::ptr_eq(node_of(g), node)) + .or_else(|| groups.iter().find(|g| **node_of(g) == **node)) + .copied() + .ok_or_else(|| format!("query {query}'s root has no root group")) +} + +/// Each query's root with every operator reference replaced by the composed +/// DAG of its root group (`root_groups`). +fn query_roots( + composed: &[(MemoGroupId, Rc)], + stage0_roots: &[QueryRoot], + root_groups: &Result>, String>, +) -> Result, String> { + let root_groups = root_groups.as_ref().map_err(String::clone)?; + let composed: HashMap<_, _> = composed.iter().cloned().collect(); + stage0_roots + .iter() + .zip(root_groups) + .map(|(root, groups)| { + let mut next = groups.iter().map(|group| Rc::clone(&composed[group])); + Ok(match root { + QueryRoot::Operator(_) => QueryRoot::Operator(next.next().expect("one group")), + QueryRoot::Scalar(expression) => QueryRoot::Scalar( + expression.map_operator_refs(&mut |_| next.next().expect("a group per ref")), + ), + }) + }) + .collect() +} diff --git a/crates/devtools/tests/memo_pricing.rs b/crates/devtools/tests/memo_pricing.rs new file mode 100644 index 000000000..902cf3496 --- /dev/null +++ b/crates/devtools/tests/memo_pricing.rs @@ -0,0 +1,352 @@ +//! Pricing a memo's enumerated choices: on small workloads every choice is +//! priced, and none that is valid is cheaper than the plan the search +//! selected. +use asap_devtools::memo_pricing::{price_choices, PricedChoice, SelectedBaseline}; +use asap_devtools::planner_layering_examples::example1; +use asap_devtools::planning_stages::{lower_promql_roots, run_default_search_over}; +use asap_logical_asap_optimization::deployment_capabilities::DeploymentCapabilities; +use asap_logical_asap_optimization::memo::candidates::{try_populate_pass1, CandidateMemo}; +use asap_logical_asap_optimization::memo::enumeration::enumerate; +use asap_logical_asap_optimization::memo::LogicalMemo; +use asap_logical_asap_optimization::per_computation_candidates::generate_per_computation_candidates; +use asap_plan_selection::{PlanSelectionSettings, PlanningModels}; +use asap_planner::PlannerExecutorCompileCheck; +use asap_types::ir::properties::AccuracyTarget; +use asap_types::ir::QueryRoot; +use asap_types::workload::{ + DataWorkload, DurationMs, PlanningWorkload, Predictability, Query, QueryLanguage, + QueryRequirements, QueryWorkload, QueryWorkloadEntry, RepeatedDemand, RepeatingEntry, + RepetitionInterval, +}; +use serde_json::{json, Value}; +use std::process::Command; + +/// The issue's workload file: the query every minute, ε = 0.01, δ = 0.001. +const RATIO_OF_QUANTILES_WORKLOAD: &str = + "../../tools/dag-viewer/memo-examples/promql/ratio-of-quantiles-5m.json"; + +const RATIO_OF_QUANTILES: &str = "avg by (job) (quantile_over_time(0.5, data[5m]) \ + / quantile_over_time(0.5, data[5m]) + quantile_over_time(0.9, data[5m]))"; + +/// One query evaluated every minute at max_error 0.01, failure probability +/// 0.001, over 1,000 series of `data`, each sampled every 15 s. +fn ratio_of_quantiles() -> PlanningWorkload { + PlanningWorkload { + query_workload: QueryWorkload { + language: QueryLanguage::PromQL, + query_batch: None, + repeating_queries: Some(vec![RepeatingEntry { + query: Query(RATIO_OF_QUANTILES.into()), + demand: RepeatedDemand::FixedInterval(RepetitionInterval(60_000)), + requirements: QueryRequirements { + accuracy: AccuracyTarget::max_error(0.01).failure_probability(0.001), + max_latency_ms: None, + }, + predictability: Predictability::Unknown, + range_evaluation: None, + }]), + }, + data_workload: Some({ + let mut data = DataWorkload::sampled_every(["data".to_string()], DurationMs(15_000)); + for facts in data.sources.values_mut() { + facts.series_count = Some(1_000); + } + data + }), + } +} + +/// The workload's roots, entries and data, and its Pass 1 memo. +struct Planned { + roots: Vec, + entries: Vec, + data: DataWorkload, + memo: CandidateMemo, +} + +/// `stage_pipeline` lowers each series' identity as a column +/// (`lower_promql_roots`); without it (`series_identity` false, as the +/// memo-enumeration tests lower), the ratio's two `quantile 0.5` have no +/// provable unique key and stay separate groups. +fn planned(workload: &PlanningWorkload, series_identity: bool) -> Planned { + let roots = if series_identity { + lower_promql_roots(workload).unwrap() + } else { + asap_frontend_promql::lower_promql_query_workload(workload) + .unwrap() + .into_iter() + .collect() + }; + let indexed: Vec<_> = roots.iter().cloned().enumerate().collect(); + let initial = LogicalMemo::from_roots(&indexed); + let inventory = generate_per_computation_candidates(indexed).unwrap(); + Planned { + roots, + entries: workload.query_workload.entries().collect(), + data: workload.data_workload.clone().unwrap(), + memo: try_populate_pass1(initial, &inventory).unwrap(), + } +} + +/// `stage_pipeline`'s models: the executor's summary families and compile +/// check, `settings` (accuracy guarantees ignored unless enforced). +fn models<'a>( + settings: &'a PlanSelectionSettings, + deployment: &'a DeploymentCapabilities, +) -> PlanningModels<'a> { + PlanningModels::builtin() + .with_plan_selection(settings) + .with_deployment_capabilities(deployment) + .with_executor_compile_check(&PlannerExecutorCompileCheck) +} + +fn deployment() -> DeploymentCapabilities { + DeploymentCapabilities { + supported_summaries: Some(asap_executor::supported_summary_families()), + ..DeploymentCapabilities::default() + } +} + +fn settings(enforce_accuracy_guarantees: bool) -> PlanSelectionSettings { + PlanSelectionSettings { + ignore_accuracy_guarantees: !enforce_accuracy_guarantees, + ..PlanSelectionSettings::default() + } +} + +/// The default search's selection of `planned`'s workload, as +/// `stage_pipeline` runs it. +fn selected(planned: &Planned, models: PlanningModels<'_>) -> SelectedBaseline { + let run = run_default_search_over( + planned.roots.clone(), + &planned.entries, + &planned.data, + models, + ) + .unwrap(); + SelectedBaseline::of(run.stage3_plan_selection.as_ref().expect("a selection")) +} + +/// Every choice of `planned`'s Pass 1 memo (the enumeration is complete), +/// priced against `baseline`. +fn price_every_choice( + planned: &Planned, + models: PlanningModels<'_>, + baseline: SelectedBaseline, +) -> Vec { + let enumeration = enumerate(&planned.memo, 100_000); + assert!(enumeration.complete); + price_choices( + &planned.memo, + &enumeration.choices, + &planned.roots, + &planned.entries, + &planned.data, + models, + Some(baseline), + ) +} + +fn assert_none_cheaper(priced: &[PricedChoice], baseline: SelectedBaseline) { + let cheaper: Vec<_> = priced + .iter() + .enumerate() + .filter(|(_, p)| p.cheaper_than_selected) + .map(|(i, p)| (format!("E{}", i + 1), p.cost_per_second)) + .collect(); + assert!( + cheaper.is_empty(), + "selected {:?}; cheaper: {cheaper:?}", + baseline.cost_per_second + ); + assert!(priced.iter().any(|p| p.valid)); +} + +/// `stage_pipeline --price-enumerated 200` on `args`' workload: the listed +/// choices of the enumeration's step, each priced. +fn priced_document(args: &[&str]) -> Value { + static CALLS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0); + let call = CALLS.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let out = std::env::temp_dir().join(format!( + "stage_pipeline_{}_{call}_priced.json", + std::process::id() + )); + let status = Command::new(env!("CARGO_BIN_EXE_stage_pipeline")) + .args(args) + .args(["--price-enumerated", "200", "--out"]) + .arg(&out) + .status() + .unwrap(); + assert!(status.success()); + let document = serde_json::from_str(&std::fs::read_to_string(&out).unwrap()).unwrap(); + std::fs::remove_file(out).unwrap(); + document +} + +/// Every listed choice is priced; the enumeration of its step is complete, +/// and no valid choice is cheaper than the selected plan, whose cost is +/// the one Stage 3 reports. Returns the choices. +fn assert_no_listed_choice_is_cheaper(document: &Value) -> Vec { + let enumeration = &document["memo"]["enumeration"]; + let step = enumeration["step"].as_str().unwrap(); + assert_eq!(enumeration["per_step"][step]["complete"], true); + let choices = enumeration["choices"].as_array().unwrap().clone(); + assert_eq!( + enumeration["per_step"][step]["count"].as_u64().unwrap(), + choices.len() as u64 + ); + let stage3 = &document["stage3_plan_selection"]; + let selected = &stage3["costs"][stage3["selected"].as_str().unwrap()]["cost_per_second"]; + assert_eq!( + enumeration["priced"], + json!({"count": choices.len(), "cheaper_than_selected": 0, "selected_cost_per_second": selected}) + ); + for choice in &choices { + assert_eq!(choice["cheaper_than_selected"], false, "{}", choice["id"]); + if choice["valid"] == true { + assert!(choice["cost_per_second"].as_f64().unwrap() >= selected.as_f64().unwrap()); + assert_eq!(choice["rejection"], Value::Null); + } else { + assert!(choice["rejection"].is_string(), "{}", choice["id"]); + } + } + assert!(choices.iter().any(|c| c["valid"] == true)); + choices +} + +/// #509 Example 1: its Pass 1 memo (the listed step) stands for 5 DAGs; +/// Q2's summary reads miss its 100 ms latency bound, and no valid DAG is +/// cheaper than the selection. +#[test] +fn example1_no_listed_choice_is_cheaper_than_the_selected_plan() { + let document = priced_document(&["--example", "planner-layering-1"]); + let choices = assert_no_listed_choice_is_cheaper(&document); + assert_eq!(document["memo"]["enumeration"]["step"], "pass1"); + assert_eq!(choices.len(), 5); + assert!(choices + .iter() + .any(|c| c["rejection"] == "latency_bound_exceeded")); +} + +/// The issue's ratio of quantiles as `stage_pipeline` lowers it (series +/// identity: the two `quantile 0.5` share one group): no listed DAG is +/// cheaper than the selection, with accuracy guarantees ignored (the +/// default) or enforced. +#[test] +fn ratio_of_quantiles_no_listed_choice_is_cheaper_than_the_selected_plan() { + for enforce in [&[][..], &["--enforce-accuracy-guarantees"][..]] { + let mut args = vec!["--workload", RATIO_OF_QUANTILES_WORKLOAD]; + args.extend_from_slice(enforce); + let choices = assert_no_listed_choice_is_cheaper(&priced_document(&args)); + assert!(choices.len() >= 49); + } +} + +/// Lowered without series identity, the two `quantile 0.5` stay separate +/// groups: 7 × 7 × 7 = 343 DAGs, 27 of them sized for the whole budget. The +/// executor compiles none of their per-series summaries without the series +/// identity, so only the exact DAG is valid, and it is not cheaper than the +/// plan selected for the same query as `stage_pipeline` lowers it. (The +/// default search finds no plan for this lowering in its budget, so the +/// selection comes from the series-identity lowering.) +#[test] +fn ratio_of_quantiles_343_pass1_choices_none_cheaper_than_the_selected_plan() { + let without_identity = planned(&ratio_of_quantiles(), false); + let enumeration = enumerate(&without_identity.memo, 100_000); + let whole_budget = enumeration + .choices + .iter() + .filter(|choice| { + choice.iter().all(|(group, &index)| { + without_identity.memo.candidates[group.0 as usize][index].budget_share == 1 + }) + }) + .count(); + assert_eq!((enumeration.choices.len(), whole_budget), (343, 27)); + let with_identity = planned(&ratio_of_quantiles(), true); + let deployment = deployment(); + for enforce in [false, true] { + let settings = settings(enforce); + let models = models(&settings, &deployment); + let baseline = selected(&with_identity, models); + let started = std::time::Instant::now(); + let priced = price_every_choice(&without_identity, models, baseline); + eprintln!( + "priced {} choices in {:.2?}", + priced.len(), + started.elapsed() + ); + assert_none_cheaper(&priced, baseline); + assert_eq!(priced.iter().filter(|p| p.valid).count(), 1); + } +} + +/// A selection one step worse than the cheapest valid choice (the +/// next-cheapest cost) makes exactly the cheaper choices +/// `cheaper_than_selected`. +#[test] +fn a_worse_selection_is_caught_by_cheaper_than_selected() { + let planned = planned(&example1(), true); + let (settings, deployment) = (settings(false), deployment()); + let models = models(&settings, &deployment); + let choices = enumerate(&planned.memo, 100).choices; + let price = |selected| { + price_choices( + &planned.memo, + &choices, + &planned.roots, + &planned.entries, + &planned.data, + models, + selected, + ) + }; + let unpriced = price(None); + assert!(unpriced.iter().all(|p| !p.cheaper_than_selected)); + let mut costs: Vec = unpriced.iter().filter_map(|p| p.cost_per_second).collect(); + costs.sort_by(f64::total_cmp); + costs.dedup(); + assert!(costs.len() >= 2, "two valid choices of different cost"); + let worse = SelectedBaseline { + cost_per_second: Some(costs[1]), + calibration_factor: None, + }; + let priced = price(Some(worse)); + for (before, after) in unpriced.iter().zip(&priced) { + assert_eq!(before.cost_per_second, after.cost_per_second); + assert_eq!( + after.cheaper_than_selected, + after.cost_per_second == Some(costs[0]) + ); + } + assert!(priced.iter().any(|p| p.cheaper_than_selected)); +} + +/// A query whose root is a scalar expression reading two operator DAGs: +/// each reference is replaced by its own root group's composed DAG, so +/// every choice composes and is priced. +#[test] +fn scalar_root_choices_are_composed_and_priced() { + let mut workload = ratio_of_quantiles(); + workload.query_workload.repeating_queries.as_mut().unwrap()[0].query = Query( + "scalar(sum(quantile_over_time(0.5, data[5m]))) + scalar(max(rate(data[1m])))".into(), + ); + let planned = planned(&workload, true); + assert!(matches!(planned.roots[0], QueryRoot::Scalar(_))); + let (settings, deployment) = (settings(false), deployment()); + let enumeration = enumerate(&planned.memo, 100_000); + assert!(enumeration.complete && enumeration.choices.len() > 1); + let priced = price_choices( + &planned.memo, + &enumeration.choices, + &planned.roots, + &planned.entries, + &planned.data, + models(&settings, &deployment), + None, + ); + for p in &priced { + assert_ne!(p.rejection.as_deref(), Some("compose_failed"), "{p:?}"); + } + assert!(priced.iter().any(|p| p.valid)); +} diff --git a/crates/devtools/tests/stage_pipeline.rs b/crates/devtools/tests/stage_pipeline.rs index d181f544c..a7fa547c0 100644 --- a/crates/devtools/tests/stage_pipeline.rs +++ b/crates/devtools/tests/stage_pipeline.rs @@ -2299,7 +2299,10 @@ fn memo_section_lists_each_step_the_selected_choice_and_the_enumeration() { assert!(!choices.is_empty() && choices.len() <= 200); for choice in choices { validated_dag(&choice["dag"], 1); + // Pricing is off without `--price-enumerated`. + assert!(choice.get("valid").is_none()); } + assert_eq!(enumeration["priced"], Value::Null); // `--no-memo` leaves the section out. let out = std::env::temp_dir().join(format!( "stage_pipeline_{}_no_memo_ratio.json",