From f434db787e7221b0827b3d8a84768f0ee1ced826 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Tue, 13 Oct 2026 18:07:04 +0000 Subject: [PATCH] refactor(planner): name the planner pipeline Stage 1 to Stage 3 and its fields by stage Follow-up to the review on #674. Pure rename plus docs; behavior unchanged. - pass/stage_pipeline.rs -> pass/stage1_to_stage3.rs; plan_stages -> run_stage1_to_stage3; StagePipelineRun -> Stage1ToStage3Run; StagePipeline -> Stage1ToStage3Pass, registered as "stage1-to-stage3". - Run fields named by stage: stage1_pass1_per_computation, stage1_logical_asap_optimization, stage2_physical_asap_optimization, stage2_pruned, stage3_plan_selection; combinations documented as Stage 1. - Candidate label -> display_name (Logical/Physical/PrunedCandidate and the stage_pipeline JSON key); target_owners -> query_of_target. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01W7qG9aFyPij5uWsyAJCxDW --- crates/asap-aware-mapping/src/lib.rs | 17 ++- crates/asap-aware-mapping/src/pass/mod.rs | 22 +-- ...{stage_pipeline.rs => stage1_to_stage3.rs} | 141 ++++++++++-------- .../src/stage2_physical_asap_optimization.rs | 8 +- .../src/physical_planner/candidates.rs | 2 +- .../tests/common/mod.rs | 14 +- crates/devtools/src/bin/stage_pipeline.rs | 12 +- crates/devtools/src/planning_stages.rs | 16 +- crates/frontend-promql/tests/support.rs | 18 +-- .../tests/univmon_candidates.rs | 8 +- crates/frontend-sql/tests/pearson_corr.rs | 2 +- crates/integration-tests/src/lib.rs | 14 +- .../tests/operator_sharing.rs | 2 +- .../tests/planner_layering_common/mod.rs | 10 +- .../tests/planner_layering_example1.rs | 2 +- crates/planner/src/lib.rs | 16 +- crates/planner/tests/e2e_plan.rs | 22 ++- crates/planner/tests/plan_stages.rs | 35 +++-- crates/planner/tests/summary_sharing.rs | 14 +- ...d_interface_with_pluggable_optimization.md | 33 ++-- .../examples/planner-layering-example1.json | 24 +-- 21 files changed, 239 insertions(+), 193 deletions(-) rename crates/asap-aware-mapping/src/pass/{stage_pipeline.rs => stage1_to_stage3.rs} (69%) diff --git a/crates/asap-aware-mapping/src/lib.rs b/crates/asap-aware-mapping/src/lib.rs index 81cfc75e9..e7f7dabcf 100644 --- a/crates/asap-aware-mapping/src/lib.rs +++ b/crates/asap-aware-mapping/src/lib.rs @@ -29,14 +29,15 @@ //! //! ## Planning workflows //! -//! The planner runs the planning stages in order: logical ASAP-aware -//! optimization ([`per_computation_candidates`]: Pass 1 and composition), -//! physical ASAP-aware optimization ([`stage2_physical_asap_optimization`]) and plan -//! selection ([`stage3_plan_selection`]). +//! The planner runs Stage 1 to Stage 3 in order: Stage 1, logical ASAP-aware +//! optimization ([`per_computation_candidates`]: Pass 1 and composition); +//! Stage 2, physical ASAP-aware optimization +//! ([`stage2_physical_asap_optimization`]); Stage 3, plan selection +//! ([`stage3_plan_selection`]). //! //! - Run them for a parsed workload through [`optimize`] with -//! [`StagePipeline`], the default pass. -//! - Call [`plan_stages`] over the frontends' roots to read every stage's +//! [`Stage1ToStage3Pass`], the default pass. +//! - Call [`run_stage1_to_stage3`] over the frontends' roots to read every stage's //! candidates as well as the selected plan. //! //! The planner no longer calls the legacy candidate search @@ -187,9 +188,9 @@ pub use explanation::{ }; pub use grouping::{has_subpopulations, HydraGroupingStrategy}; pub use pass::{ - optimize, plan_stages, OptimizationInput, OptimizationInputError, OptimizationPass, + optimize, run_stage1_to_stage3, OptimizationInput, OptimizationInputError, OptimizationPass, OptimizeError, PassNameConflict, PassRegistry, PlanOutput, PlanningModels, QueryPlan, - StagePipeline, StagePipelineRun, + Stage1ToStage3Pass, Stage1ToStage3Run, }; pub use recurrence::{ evaluation_rate_of, total_cost, update_rate_from_data_workload, CostRate, EvaluationRate, diff --git a/crates/asap-aware-mapping/src/pass/mod.rs b/crates/asap-aware-mapping/src/pass/mod.rs index c18d91614..5b485a224 100644 --- a/crates/asap-aware-mapping/src/pass/mod.rs +++ b/crates/asap-aware-mapping/src/pass/mod.rs @@ -6,14 +6,14 @@ //! `TargetSubDAGCandidates`, no `ReplacementStrategy` — so an algorithm with no //! candidate-generation phase at all (a greedy MQO loop, say) can implement it //! without pretending to have phases it does not have. The shipped algorithm is -//! one implementation, [`StagePipeline`]: the planning stages in order -//! ([`plan_stages`]). +//! one implementation, [`Stage1ToStage3Pass`]: the planning stages in order +//! ([`run_stage1_to_stage3`]). //! //! Call [`optimize`] rather than [`OptimizationPass::optimize`] directly: it //! validates the input once for every pass and checks the output contract that //! downstream consumers rely on. -mod stage_pipeline; +mod stage1_to_stage3; use std::collections::BTreeMap; use std::rc::Rc; @@ -34,8 +34,8 @@ use crate::cost_model::{CostModel, DefaultCostModel}; use crate::per_computation_candidates::PerComputationCandidateError; use crate::stage3_plan_selection::{Selection, SelectionError}; -pub use stage_pipeline::{ - plan_stages, LogicalCandidate, PrunedCandidate, StagePipeline, StagePipelineRun, +pub use stage1_to_stage3::{ + run_stage1_to_stage3, LogicalCandidate, PrunedCandidate, Stage1ToStage3Pass, Stage1ToStage3Run, }; static DEFAULT_COST_MODEL: DefaultCostModel = DefaultCostModel; @@ -48,7 +48,7 @@ static NO_ACCURACY_EVIDENCE: NoAccuracyEvidence = NoAccuracyEvidence; /// a built-in default, evidence about a particular deployment cannot. /// /// Plan selection reads `accuracy` and `evidence`. It prices candidates with -/// its own analytical model, so [`StagePipeline`] does not read `cost`. +/// its own analytical model, so [`Stage1ToStage3Pass`] does not read `cost`. #[derive(Clone, Copy)] #[non_exhaustive] pub struct PlanningModels<'a> { @@ -401,11 +401,11 @@ impl PassRegistry { Self::default() } - /// Only [`StagePipeline`], under the name `stage-pipeline`. + /// Only [`Stage1ToStage3Pass`], under the name `stage1-to-stage3`. pub fn with_builtin() -> Self { let mut registry = Self::new(); registry - .register(Box::new(StagePipeline)) + .register(Box::new(Stage1ToStage3Pass)) .expect("empty registry cannot conflict"); registry } @@ -589,17 +589,17 @@ mod tests { assert_eq!(err.0, "greedy"); } - /// The builtin registry resolves `stage-pipeline`, and names come back + /// The builtin registry resolves `stage1-to-stage3`, and names come back /// sorted so a sweep over every registered pass is reproducible. #[test] fn registry_resolves_builtin_and_lists_names_in_order() { let mut registry = PassRegistry::with_builtin(); registry.register(Box::new(Stub("alpha"))).unwrap(); - assert!(registry.get("stage-pipeline").is_some()); + assert!(registry.get("stage1-to-stage3").is_some()); assert!(registry.get("absent").is_none()); assert_eq!( registry.names().collect::>(), - vec!["alpha", "stage-pipeline"] + vec!["alpha", "stage1-to-stage3"] ); } } diff --git a/crates/asap-aware-mapping/src/pass/stage_pipeline.rs b/crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs similarity index 69% rename from crates/asap-aware-mapping/src/pass/stage_pipeline.rs rename to crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs index f4bbe6173..525f3afb3 100644 --- a/crates/asap-aware-mapping/src/pass/stage_pipeline.rs +++ b/crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs @@ -1,18 +1,21 @@ -//! The planner pipeline (#509 MVP): the planning stages in order over the -//! frontends' query roots, and [`StagePipeline`], the -//! [`OptimizationPass`](super::OptimizationPass) that runs them. +//! Stage 1 (logical ASAP-aware optimization) → Stage 2 (physical ASAP-aware +//! optimization) → Stage 3 (plan selection), starting from the query-language +//! frontends' roots; Stage 0 runs before it in the facade. //! -//! [`plan_stages`] runs, for one workload: -//! 1. logical ASAP-aware optimization: Pass 1 per-computation candidate +//! This is the planner pipeline (#509 MVP). [`Stage1ToStage3Pass`] is the +//! [`OptimizationPass`](super::OptimizationPass) that runs it. +//! +//! [`run_stage1_to_stage3`] runs, for one workload: +//! 1. Stage 1, logical ASAP-aware optimization: Pass 1 per-computation candidate //! generation, then composition of one whole-workload candidate per choice //! of a per-computation candidate for every target (the Cartesian product, //! in mixed-radix order, the last target varying fastest); -//! 2. physical ASAP-aware optimization of every composed candidate +//! 2. Stage 2, physical ASAP-aware optimization of every composed candidate //! ([`optimize_physical_asap`]); -//! 3. plan selection over the physical candidates ([`select_plan`]). +//! 3. Stage 3, plan selection over the physical candidates ([`select_plan`]). //! -//! The query-language frontends run before it (the `asap-planner` facade -//! lowers the workload). Pass 2 (ASAP-aware CSE) is not planned yet. +//! Stage 0, the query-language frontends, runs before it (the `asap-planner` +//! facade lowers the workload). Pass 2 (ASAP-aware CSE) is not planned yet. //! //! Every combination is enumerated and plan selection prices every //! candidate; nothing is capped. A search that avoids building every @@ -42,8 +45,9 @@ use crate::PlanningModels; pub struct LogicalCandidate { /// `L`, numbered in enumeration order. pub id: String, - /// E.g. "Q1 exact · Q2 CMS+heap"; exact accumulators in parentheses. - pub label: String, + /// Human-readable name of the candidate, e.g. "Q1 exact · Q2 CMS+heap", used only for + /// display. + pub display_name: String, pub choice: Vec, pub roots: Result, PerComputationCandidateError>, } @@ -53,32 +57,42 @@ pub struct LogicalCandidate { #[derive(Debug, Clone, PartialEq)] pub struct PrunedCandidate { pub from_logical: String, - pub label: String, + /// Human-readable name of the candidate, e.g. "Q1 exact · Q2 CMS+heap", used only for + /// display. + pub display_name: String, pub reason: String, } -/// Every stage's output for one workload ([`plan_stages`]). +/// The output of Stage 1, Stage 2 and Stage 3 for one workload +/// ([`run_stage1_to_stage3`]). #[derive(Debug)] -pub struct StagePipelineRun { - /// Pass 1's per-computation candidates, with the roots they read. - pub per_computation: PerComputationCandidates, - /// Size of the Cartesian product of the targets' candidates, saturating - /// at `usize::MAX`. +pub struct Stage1ToStage3Run { + /// Stage 1, Pass 1: the per-computation candidates, with the roots they + /// read. + pub stage1_pass1_per_computation: PerComputationCandidates, + /// Stage 1: the number of whole-workload candidates, the size of the + /// Cartesian product of the targets' candidates, saturating at + /// `usize::MAX`. pub combinations: usize, - /// Every logical candidate, in enumeration order. - pub logical: Vec, - /// One physical candidate (`P` from `L`) per logical candidate that - /// composed and was implemented. - pub physical: Vec, - pub pruned: Vec, - pub selection: Result, + /// Stage 1: every whole-workload logical candidate, in enumeration order. + pub stage1_logical_asap_optimization: Vec, + /// Stage 2: one physical candidate (`P` from `L`) per Stage 1 + /// candidate that composed and was implemented. + pub stage2_physical_asap_optimization: Vec, + /// Stage 2: the Stage 1 candidates Stage 2 could not implement, with the + /// reasons. + pub stage2_pruned: Vec, + /// Stage 3: plan selection over the Stage 2 candidates. + pub stage3_plan_selection: Result, } -impl StagePipelineRun { +impl Stage1ToStage3Run { /// The selected physical candidate, if selection succeeded. pub fn selected(&self) -> Option<&PhysicalCandidate> { - let selected = &self.selection.as_ref().ok()?.selected; - self.physical.iter().find(|p| &p.id == selected) + let selected = &self.stage3_plan_selection.as_ref().ok()?.selected; + self.stage2_physical_asap_optimization + .iter() + .find(|p| &p.id == selected) } /// The composed roots of the logical candidate the selected physical @@ -86,55 +100,58 @@ impl StagePipelineRun { /// implementation and materialization. pub fn selected_logical(&self) -> Option<&[QueryRoot]> { let from = &self.selected()?.from_logical; - let logical = self.logical.iter().find(|l| &l.id == from)?; + let logical = self + .stage1_logical_asap_optimization + .iter() + .find(|l| &l.id == from)?; logical.roots.as_deref().ok() } } -/// Run the planning stages over the frontends' `roots`, enumerating every +/// Run Stage 1 → Stage 2 → Stage 3 over the frontends' `roots`, enumerating every /// whole-workload candidate. `targets[i]` is the accuracy /// requirement of `roots[i]` (`None` imposes none). Only a Pass 1 failure is /// an error; a candidate that does not compose or cannot be implemented is /// listed with its reason, and a failed selection is reported in the run. -pub fn plan_stages( +pub fn run_stage1_to_stage3( roots: Vec<(Id, QueryRoot)>, targets: &[Option], data: &DataWorkload, models: PlanningModels<'_>, -) -> Result, PerComputationCandidateError> { +) -> Result, PerComputationCandidateError> { let per_computation = generate_per_computation_candidates(roots)?; let combinations = per_computation .targets .iter() .map(|target| target.choices.candidates.len()) .fold(1usize, usize::saturating_mul); - let owners = target_owners(&per_computation); + let query_of_target = query_of_target(&per_computation); let mut logical = Vec::new(); let mut physical = Vec::new(); let mut pruned = Vec::new(); let mut choice = vec![0; per_computation.targets.len()]; for index in 0usize.. { let id = format!("L{}", index + 1); - let label = label(&per_computation, &owners, &choice); + let display_name = display_name(&per_computation, &query_of_target, &choice); let roots = compose_logical_candidate(&per_computation, &choice) .map(|roots| roots.into_iter().map(|(_, root)| root).collect::>()); if let Ok(roots) = &roots { match optimize_physical_asap(&id, roots) { Ok(mut candidate) => { candidate.id = format!("P{}", index + 1); - candidate.label = label.clone(); + candidate.display_name = display_name.clone(); physical.push(candidate); } Err(error) => pruned.push(PrunedCandidate { from_logical: id.clone(), - label: label.clone(), + display_name: display_name.clone(), reason: error.to_string(), }), } } logical.push(LogicalCandidate { id, - label, + display_name, choice: choice.clone(), roots, }); @@ -154,18 +171,20 @@ pub fn plan_stages( } } let selection = select_plan(&physical, targets, data, models); - Ok(StagePipelineRun { - per_computation, + Ok(Stage1ToStage3Run { + stage1_pass1_per_computation: per_computation, combinations, - logical, - physical, - pruned, - selection, + stage1_logical_asap_optimization: logical, + stage2_physical_asap_optimization: physical, + stage2_pruned: pruned, + stage3_plan_selection: selection, }) } -/// The first root that reaches each target (enumeration order). -fn target_owners(inventory: &PerComputationCandidates) -> Vec { +/// `query_of_target[t]`: the index of the query root that first reads target +/// `t` (the first root, in input order, from which the target is reachable). +/// [`display_name`] lists each target's choice under that query. +fn query_of_target(inventory: &PerComputationCandidates) -> Vec { inventory .targets .iter() @@ -192,20 +211,24 @@ fn root_operators(root: &QueryRoot) -> Vec<&Rc> { } } -/// E.g. "Q1 exact · Q2 CMS+heap"; exact accumulators are listed in parentheses. -fn label( +/// A human-readable name of a candidate, e.g. "Q1 exact · Q2 CMS+heap" +/// (exact accumulators are listed in parentheses), used only for display in +/// the `stage_pipeline` devtool and the viewer; planning never reads it. +fn display_name( inventory: &PerComputationCandidates, - owners: &[usize], + query_of_target: &[usize], choice: &[usize], ) -> String { (0..inventory.roots.len()) .map(|query| { let mut sketches = Vec::new(); let mut accumulators = Vec::new(); - for (target, (&owner, &index)) in - inventory.targets.iter().zip(owners.iter().zip(choice)) + for (target, (&target_query, &index)) in inventory + .targets + .iter() + .zip(query_of_target.iter().zip(choice)) { - if owner != query { + if target_query != query { continue; } match &target.choices.candidates[index] { @@ -242,14 +265,14 @@ fn label( /// workload's roots in entry order, PromQL roots with each series' full /// identity as a column (the row representation per-series state needs at /// runtime; a root with no such representation is planned over its labels -/// alone), through [`plan_stages`] with every entry's accuracy target. The +/// alone), through [`run_stage1_to_stage3`] with every entry's accuracy target. The /// output is the selected physical candidate's roots. #[derive(Debug, Default, Clone, Copy)] -pub struct StagePipeline; +pub struct Stage1ToStage3Pass; -impl OptimizationPass for StagePipeline { +impl OptimizationPass for Stage1ToStage3Pass { fn name(&self) -> &'static str { - "stage-pipeline" + "stage1-to-stage3" } fn optimize(&self, input: OptimizationInput<'_>) -> Result { @@ -280,10 +303,10 @@ impl OptimizationPass for StagePipeline { .map(|entry| Some(entry.requirements.accuracy.target())) .collect(); let data = workload.data_workload().cloned().unwrap_or_default(); - let run = plan_stages(roots, &targets, &data, input.models)?; - let StagePipelineRun { - physical, - selection, + let run = run_stage1_to_stage3(roots, &targets, &data, input.models)?; + let Stage1ToStage3Run { + stage2_physical_asap_optimization: physical, + stage3_plan_selection: selection, .. } = run; let selection = selection?; diff --git a/crates/asap-aware-mapping/src/stage2_physical_asap_optimization.rs b/crates/asap-aware-mapping/src/stage2_physical_asap_optimization.rs index 30dee2a3f..67c21dd9d 100644 --- a/crates/asap-aware-mapping/src/stage2_physical_asap_optimization.rs +++ b/crates/asap-aware-mapping/src/stage2_physical_asap_optimization.rs @@ -49,7 +49,9 @@ use thiserror::Error; pub struct PhysicalCandidate { pub id: String, pub from_logical: String, - pub label: String, + /// Human-readable name of the candidate, used only for display (see + /// `pass::stage1_to_stage3`). + pub display_name: String, pub roots: Vec, pub dag: PhysicalASAPDAG, } @@ -66,7 +68,7 @@ pub enum PhysicalASAPOptimizationError { /// Stage 2 for one CandidateLogicalASAPDAG: implement every operator /// (physical operator implementation) and time it with every -/// sub-DAG `NotMaterialized`. `id` and `label` are left empty for the caller +/// sub-DAG `NotMaterialized`. `id` and `display_name` are left empty for the caller /// to name. Sharing between `roots` is preserved: one memo serves the whole /// workload. pub fn optimize_physical_asap( @@ -88,7 +90,7 @@ pub fn optimize_physical_asap( Ok(PhysicalCandidate { id: String::new(), from_logical: from_logical.to_string(), - label: String::new(), + display_name: String::new(), roots: timed, dag, }) diff --git a/crates/asap-physical-operators/src/physical_planner/candidates.rs b/crates/asap-physical-operators/src/physical_planner/candidates.rs index 1657e9f0a..36075d825 100644 --- a/crates/asap-physical-operators/src/physical_planner/candidates.rs +++ b/crates/asap-physical-operators/src/physical_planner/candidates.rs @@ -376,7 +376,7 @@ mod tests { let root = promql_rows::with_series_identity(&root).unwrap(); // The logical plan the planner pipeline selects, materialized at // ingestion time. - let run = asap_aware_mapping::plan_stages( + let run = asap_aware_mapping::run_stage1_to_stage3( vec![(0, planner_types::ir::QueryRoot::Operator(root))], &[Some(planner_types::types::AccuracyTarget::Exact)], &workload.data_workload.unwrap(), diff --git a/crates/asap-physical-operators/tests/common/mod.rs b/crates/asap-physical-operators/tests/common/mod.rs index e815c5e75..a3b900287 100644 --- a/crates/asap-physical-operators/tests/common/mod.rs +++ b/crates/asap-physical-operators/tests/common/mod.rs @@ -16,14 +16,14 @@ pub fn compile_physical_asap_dag( Ok(planner_types::ir::physical_export::compile_physical_asap_dag(&root)?) } -/// The logical DAG the planner pipeline (`plan_stages`) selects for the +/// The logical DAG the planner pipeline (`run_stage1_to_stage3`) selects for the /// single query `root` at `accuracy`, over a continuously ingested source, /// with the built-in models. pub fn selected_dag( root: Rc, accuracy: planner_types::types::AccuracyTarget, ) -> Rc { - let run = plan_stages(root, accuracy); + let run = run_stage1_to_stage3(root, accuracy); let [QueryRoot::Operator(selected)] = run.selected_logical().expect("a plan is selected") else { panic!("one operator root"); @@ -38,8 +38,8 @@ pub fn logical_candidates( root: Rc, accuracy: planner_types::types::AccuracyTarget, ) -> Vec> { - plan_stages(root, accuracy) - .logical + run_stage1_to_stage3(root, accuracy) + .stage1_logical_asap_optimization .into_iter() .filter_map(|candidate| match candidate.roots.ok()?.remove(0) { QueryRoot::Operator(node) => Some(node), @@ -48,10 +48,10 @@ pub fn logical_candidates( .collect() } -fn plan_stages( +fn run_stage1_to_stage3( root: Rc, accuracy: planner_types::types::AccuracyTarget, -) -> asap_aware_mapping::StagePipelineRun { +) -> asap_aware_mapping::Stage1ToStage3Run { use planner_types::workload::{DataArrival, DataWorkload, Evidence, EvidenceSource, Rate}; let data = DataWorkload { arrival: DataArrival::ContinuouslyIngesting, @@ -62,7 +62,7 @@ fn plan_stages( }, ..Default::default() }; - asap_aware_mapping::plan_stages( + asap_aware_mapping::run_stage1_to_stage3( vec![(0, QueryRoot::Operator(root))], &[Some(accuracy)], &data, diff --git a/crates/devtools/src/bin/stage_pipeline.rs b/crates/devtools/src/bin/stage_pipeline.rs index bf3a495e4..49fba9183 100644 --- a/crates/devtools/src/bin/stage_pipeline.rs +++ b/crates/devtools/src/bin/stage_pipeline.rs @@ -140,21 +140,25 @@ fn stage_pipeline(workload: &PlanningWorkload, max_displayed: usize) -> Result json!({ "id": c.id, "label": c.label, "dag": export(roots) }), - Err(error) => json!({ "id": c.id, "label": c.label, "error": error.to_string() }), + Ok(roots) => { + json!({ "id": c.id, "display_name": c.display_name, "dag": export(roots) }) + } + Err(error) => { + json!({ "id": c.id, "display_name": c.display_name, "error": error.to_string() }) + } }) .collect(); let physical: Vec<_> = run .stage2_physical_asap_optimization .iter() .filter(|p| shown_physical.contains(p.id.as_str())) - .map(|p| json!({ "id": p.id, "from_logical": p.from_logical, "label": p.label, "dag": p.dag })) + .map(|p| json!({ "id": p.id, "from_logical": p.from_logical, "display_name": p.display_name, "dag": p.dag })) .collect(); let pruned: Vec<_> = run .stage2_pruned .iter() .filter(|p| shown.contains(p.from_logical.as_str())) - .map(|p| json!({ "from_logical": p.from_logical, "label": p.label, "reason": p.reason })) + .map(|p| json!({ "from_logical": p.from_logical, "display_name": p.display_name, "reason": p.reason })) .collect(); Ok(json!({ "format": "asap-stage-pipeline/v1", diff --git a/crates/devtools/src/planning_stages.rs b/crates/devtools/src/planning_stages.rs index 63e9bcaa1..7d673eabd 100644 --- a/crates/devtools/src/planning_stages.rs +++ b/crates/devtools/src/planning_stages.rs @@ -2,7 +2,7 @@ //! one pipeline behind `stage_pipeline` and the planner-layering acceptance //! tests, so both see the same candidates, ids and selection. -use asap_aware_mapping::pass::plan_stages; +use asap_aware_mapping::pass::run_stage1_to_stage3; pub use asap_aware_mapping::pass::{LogicalCandidate, PrunedCandidate}; use asap_aware_mapping::stage2_physical_asap_optimization::PhysicalCandidate; use asap_aware_mapping::stage3_plan_selection::{Selection, SelectionError}; @@ -40,7 +40,7 @@ pub fn lower_promql_roots(workload: &PlanningWorkload) -> Result, /// Logical ASAP-aware optimization (Pass 1 and composition; no Pass 2), /// physical ASAP-aware optimization and plan selection over `roots`, the /// lowered entries of `workload`: the library's planner pipeline -/// ([`plan_stages`]) with the built-in models. Logical candidates are every +/// ([`run_stage1_to_stage3`]) with the built-in models. Logical candidates are every /// choice of a per-computation candidate for every target, the last target /// varying fastest. Every candidate is enumerated and priced, so the /// selection is the cheapest of all; a caller that shows fewer cuts only its @@ -58,24 +58,24 @@ pub fn run_planning_stages( .map(|entry| Some(entry.requirements.accuracy.target())) .collect(); let data = workload.data_workload.clone().unwrap_or_default(); - let run = plan_stages( + let run = run_stage1_to_stage3( roots.iter().cloned().enumerate().collect(), &targets, &data, PlanningModels::builtin(), ) .map_err(|e| format!("Pass 1: {e}"))?; - for candidate in &run.logical { + for candidate in &run.stage1_logical_asap_optimization { if let Ok(roots) = &candidate.roots { validate(roots)?; } } Ok(PlanningStagesRun { stage0_query_language_frontends: roots, - stage1_logical_asap_optimization: run.logical, - stage2_physical_asap_optimization: run.physical, - stage2_pruned: run.pruned, - stage3_plan_selection: run.selection, + stage1_logical_asap_optimization: run.stage1_logical_asap_optimization, + stage2_physical_asap_optimization: run.stage2_physical_asap_optimization, + stage2_pruned: run.stage2_pruned, + stage3_plan_selection: run.stage3_plan_selection, }) } diff --git a/crates/frontend-promql/tests/support.rs b/crates/frontend-promql/tests/support.rs index 3a3e94278..0d8f79927 100644 --- a/crates/frontend-promql/tests/support.rs +++ b/crates/frontend-promql/tests/support.rs @@ -113,11 +113,11 @@ pub fn sample_expression(node: &OperatorNode) -> &ScalarExpr { /// Every planning stage's output for the single query `root` at `accuracy`, /// with the built-in models. #[allow(dead_code)] -pub fn plan_stages( +pub fn run_stage1_to_stage3( root: Rc, accuracy: AccuracyTarget, -) -> asap_aware_mapping::StagePipelineRun { - asap_aware_mapping::plan_stages( +) -> asap_aware_mapping::Stage1ToStage3Run { + asap_aware_mapping::run_stage1_to_stage3( vec![(0, asap_types::ir::QueryRoot::Operator(root))], &[Some(accuracy)], &DataWorkload::default(), @@ -130,7 +130,7 @@ pub fn plan_stages( /// at `accuracy`. #[allow(dead_code)] pub fn selected_dag(root: Rc, accuracy: AccuracyTarget) -> Rc { - let run = plan_stages(root, accuracy); + let run = run_stage1_to_stage3(root, accuracy); match run.selected_logical().expect("a plan is selected") { [asap_types::ir::QueryRoot::Operator(node)] => Rc::clone(node), other => panic!("one operator root: {other:?}"), @@ -145,8 +145,8 @@ pub fn logical_candidates( root: Rc, accuracy: AccuracyTarget, ) -> Vec> { - plan_stages(root, accuracy) - .logical + run_stage1_to_stage3(root, accuracy) + .stage1_logical_asap_optimization .into_iter() .filter_map(|candidate| match candidate.roots.ok()?.remove(0) { asap_types::ir::QueryRoot::Operator(node) => Some(node), @@ -162,7 +162,7 @@ pub fn logical_candidates( /// compose (a candidate its input cannot feed), and selection skips it. #[allow(dead_code)] pub fn logical_plan(root: &Rc) -> Result, String> { - let run = asap_aware_mapping::plan_stages( + let run = asap_aware_mapping::run_stage1_to_stage3( vec![(0, asap_types::ir::QueryRoot::Operator(Rc::clone(root)))], &[None], &DataWorkload::default(), @@ -174,13 +174,13 @@ pub fn logical_plan(root: &Rc) -> Result, String> asap_types::ir::QueryRoot::Scalar(_) => unreachable!("an operator root stays one"), }; let summarized = run - .logical + .stage1_logical_asap_optimization .iter() .filter_map(|candidate| candidate.roots.as_deref().ok().map(operator)) .find(|node| node.contains_asap()); match summarized { Some(node) => Ok(node), - None => run.logical[0] + None => run.stage1_logical_asap_optimization[0] .roots .as_deref() .map(operator) diff --git a/crates/frontend-promql/tests/univmon_candidates.rs b/crates/frontend-promql/tests/univmon_candidates.rs index 142c2345f..4552cd083 100644 --- a/crates/frontend-promql/tests/univmon_candidates.rs +++ b/crates/frontend-promql/tests/univmon_candidates.rs @@ -6,7 +6,7 @@ use asap_types::ir::cse::share_common_sub_dags; use asap_types::ir::{ASAPOp, Operator, OperatorNode}; use asap_types::post_asap::{FieldDataType, SketchAlgorithm, SketchStatistic, SummaryInputExpr}; use asap_types::types::AccuracyTarget; -use support::{logical_candidates, lower_promql, plan_stages, post_asap_dag}; +use support::{logical_candidates, lower_promql, post_asap_dag, run_stage1_to_stage3}; /// The summary evaluation over a UnivMon state that `root` reads, if any. fn univmon_estimate(root: &Rc) -> Option> { @@ -103,9 +103,9 @@ fn uncalibrated_frequency_evaluations_do_not_bypass_accuracy_targets() { }, ] { let root = lower_promql(query, target.clone()).unwrap(); - let run = plan_stages(root, target.clone()); + let run = run_stage1_to_stage3(root, target.clone()); let unknown: Vec<_> = run - .physical + .stage2_physical_asap_optimization .iter() .filter(|candidate| { let asap_types::ir::QueryRoot::Operator(node) = &candidate.roots[0] else { @@ -118,7 +118,7 @@ fn uncalibrated_frequency_evaluations_do_not_bypass_accuracy_targets() { assert!(unknown.is_empty(), "{query}"); } else { assert!(!unknown.is_empty(), "{query}"); - let selected = &run.selection.as_ref().unwrap().selected; + let selected = &run.stage3_plan_selection.as_ref().unwrap().selected; assert!( unknown.iter().all(|candidate| &candidate.id != selected), "{query}" diff --git a/crates/frontend-sql/tests/pearson_corr.rs b/crates/frontend-sql/tests/pearson_corr.rs index 07dfdcbf6..2f19daf9c 100644 --- a/crates/frontend-sql/tests/pearson_corr.rs +++ b/crates/frontend-sql/tests/pearson_corr.rs @@ -148,7 +148,7 @@ async fn corr_filter_is_a_measure_filter() { #[tokio::test] async fn corr_survives_exact_plan_compilation() { let query = lower("SELECT corr(x, y) AS r FROM a").await; - let run = asap_aware_mapping::plan_stages( + let run = asap_aware_mapping::run_stage1_to_stage3( vec![(0, asap_types::ir::QueryRoot::Operator(query.clone()))], &[Some(asap_types::types::AccuracyTarget::Exact)], &Default::default(), diff --git a/crates/integration-tests/src/lib.rs b/crates/integration-tests/src/lib.rs index ba376878b..28819a282 100644 --- a/crates/integration-tests/src/lib.rs +++ b/crates/integration-tests/src/lib.rs @@ -142,7 +142,7 @@ pub mod post_asap { } } -/// Plans and candidates from the planner pipeline (`plan_stages`) for one +/// Plans and candidates from the planner pipeline (`run_stage1_to_stage3`) for one /// query, with the built-in models. pub mod planning { use asap_types::ir::{OperatorNode, QueryRoot}; @@ -151,11 +151,11 @@ pub mod planning { use std::rc::Rc; /// Every planning stage's output for the single query `root`. - pub fn plan_stages( + pub fn run_stage1_to_stage3( root: Rc, accuracy: AccuracyTarget, - ) -> asap_aware_mapping::StagePipelineRun { - asap_aware_mapping::plan_stages( + ) -> asap_aware_mapping::Stage1ToStage3Run { + asap_aware_mapping::run_stage1_to_stage3( vec![(0, QueryRoot::Operator(root))], &[Some(accuracy)], &DataWorkload::default(), @@ -167,7 +167,7 @@ pub mod planning { /// The logical DAG the planner pipeline selects for the single query /// `root` at `accuracy`. pub fn selected_dag(root: Rc, accuracy: AccuracyTarget) -> Rc { - match plan_stages(root, accuracy).selected_logical() { + match run_stage1_to_stage3(root, accuracy).selected_logical() { Some([QueryRoot::Operator(node)]) => Rc::clone(node), other => panic!("one selected operator root: {other:?}"), } @@ -180,8 +180,8 @@ pub mod planning { root: Rc, accuracy: AccuracyTarget, ) -> Vec> { - plan_stages(root, accuracy) - .logical + run_stage1_to_stage3(root, accuracy) + .stage1_logical_asap_optimization .into_iter() .filter_map(|candidate| match candidate.roots.ok()?.remove(0) { QueryRoot::Operator(node) => Some(node), diff --git a/crates/integration-tests/tests/operator_sharing.rs b/crates/integration-tests/tests/operator_sharing.rs index d1906f339..22b68d2e2 100644 --- a/crates/integration-tests/tests/operator_sharing.rs +++ b/crates/integration-tests/tests/operator_sharing.rs @@ -4,7 +4,7 @@ //! summaries below set operators. //! //! Each test drives SQL text through `lower_sql` and the plan the planner -//! pipeline selects (`plan_stages`). +//! pipeline selects (`run_stage1_to_stage3`). use std::rc::Rc; diff --git a/crates/integration-tests/tests/planner_layering_common/mod.rs b/crates/integration-tests/tests/planner_layering_common/mod.rs index c9fdc5198..584098efa 100644 --- a/crates/integration-tests/tests/planner_layering_common/mod.rs +++ b/crates/integration-tests/tests/planner_layering_common/mod.rs @@ -78,7 +78,7 @@ pub fn operation(payload: &Payload) -> String { pub struct Logical { pub id: String, /// The choice of a per-computation candidate for every Pass 1 target. - pub label: String, + pub display_name: String, pub dag: FlatDag, pub query_roots: Vec, pub roots: Vec, @@ -144,7 +144,7 @@ impl Run { } } -fn logical(id: String, label: String, roots: Vec) -> Logical { +fn logical(id: String, display_name: String, roots: Vec) -> Logical { let dag = flatten(&roots).0; let query_roots = dag .roots @@ -156,7 +156,7 @@ fn logical(id: String, label: String, roots: Vec) -> Logical { .collect(); Logical { id, - label, + display_name, dag, query_roots, roots, @@ -175,8 +175,8 @@ pub fn run_stages(workload: &PlanningWorkload, roots: Vec) -> Run { .map(|c| { let roots = c .roots - .unwrap_or_else(|e| panic!("{} ({}) composes: {e}", c.id, c.label)); - logical(c.id, c.label, roots) + .unwrap_or_else(|e| panic!("{} ({}) composes: {e}", c.id, c.display_name)); + logical(c.id, c.display_name, roots) }) .collect(); let physical = run diff --git a/crates/integration-tests/tests/planner_layering_example1.rs b/crates/integration-tests/tests/planner_layering_example1.rs index aa195a897..eb36de598 100644 --- a/crates/integration-tests/tests/planner_layering_example1.rs +++ b/crates/integration-tests/tests/planner_layering_example1.rs @@ -436,7 +436,7 @@ fn compile_in_runtime(p: &Physical) -> Result<(), String> { let roots: Vec = dag.roots.iter().map(|&id| id as u64).collect(); compile(dag, inputs, &roots) .map(|_| ()) - .map_err(|e| format!("{} ({}): {e}", p.id, p.candidate.label)) + .map_err(|e| format!("{} ({}): {e}", p.id, p.candidate.display_name)) } /// Runtime capability check (added by the implementer, not part of the diff --git a/crates/planner/src/lib.rs b/crates/planner/src/lib.rs index 28aff12d3..9b95c5d31 100644 --- a/crates/planner/src/lib.rs +++ b/crates/planner/src/lib.rs @@ -6,9 +6,9 @@ //! (this crate) (asap-aware-mapping) //! ``` //! -//! The default pass, [`StagePipeline`], runs the planning stages in order: -//! logical ASAP-aware optimization, physical ASAP-aware optimization and plan -//! selection ([`plan_stages`]). +//! The default pass, [`Stage1ToStage3Pass`], runs Stage 1 (logical ASAP-aware +//! optimization), Stage 2 (physical ASAP-aware optimization) and Stage 3 (plan +//! selection) in order ([`run_stage1_to_stage3`]). //! //! [`e2e_plan`] runs both steps. A caller who already holds pre-ASAP IR — a //! new frontend, a deserialized plan, a test that does not want to build SQL @@ -29,8 +29,8 @@ use asap_frontend_sql::{lower_sql_dialect, SqlCatalog, SqlError}; // configures the same models and reads the same output whether it goes through // `e2e_plan` or straight to `optimize`. pub use asap_aware_mapping::pass::{ - optimize, plan_stages, OptimizationInput, OptimizationPass, OptimizeError, PassRegistry, - PlanOutput, PlanningModels, QueryPlan, StagePipeline, StagePipelineRun, + optimize, run_stage1_to_stage3, OptimizationInput, OptimizationPass, OptimizeError, + PassRegistry, PlanOutput, PlanningModels, QueryPlan, Stage1ToStage3Pass, Stage1ToStage3Run, }; // ── Input ──────────────────────────────────────────────────────────────── @@ -61,7 +61,7 @@ pub struct UserInput<'a> { pub workload: &'a PlanningWorkload, pub frontend_specific: FrontendInput<'a>, pub models: PlanningModels<'a>, - /// `None` uses [`StagePipeline`]. A black-box caller never sets this. + /// `None` uses [`Stage1ToStage3Pass`]. A black-box caller never sets this. pub pass: Option<&'a dyn OptimizationPass>, } @@ -166,7 +166,7 @@ pub enum PlanError { // ── Entry point ────────────────────────────────────────────────────────── /// Lower every query (the query-language frontends), then run the -/// optimization pass over the result: by default [`StagePipeline`], the +/// optimization pass over the result: by default [`Stage1ToStage3Pass`], the /// planning stages in order. /// /// Async because the SQL frontend plans through DataFusion. One failed query @@ -178,7 +178,7 @@ pub async fn e2e_plan(input: UserInput<'_>) -> Result { let exprs = lower(&input).await?; let parsed = ParsedWorkload::from_roots(input.workload.clone(), exprs)?; - let default_pass = StagePipeline; + let default_pass = Stage1ToStage3Pass; let pass: &dyn OptimizationPass = input.pass.unwrap_or(&default_pass); let optimization = OptimizationInput::new(&parsed, input.models); diff --git a/crates/planner/tests/e2e_plan.rs b/crates/planner/tests/e2e_plan.rs index 920a495c7..0ffa94fcc 100644 --- a/crates/planner/tests/e2e_plan.rs +++ b/crates/planner/tests/e2e_plan.rs @@ -4,8 +4,8 @@ use std::rc::Rc; use asap_aware_mapping::pass::{ - plan_stages, OptimizationInput, OptimizationPass, OptimizeError, PlanOutput, PlanningModels, - StagePipeline, + run_stage1_to_stage3, OptimizationInput, OptimizationPass, OptimizeError, PlanOutput, + PlanningModels, Stage1ToStage3Pass, }; use asap_frontend_sql::{lower_sql_dialect, SqlCatalog}; use asap_planner::{e2e_plan, FrontendInput, PlanError, UserInput, UserInputError}; @@ -89,9 +89,9 @@ async fn plans_every_query_in_entry_order() { } /// The facade selects what the planning stages select over the frontends' -/// roots ([`plan_stages`]), and reports that selection. +/// roots ([`run_stage1_to_stage3`]), and reports that selection. #[tokio::test] -async fn facade_plans_match_plan_stages() { +async fn facade_plans_match_run_stage1_to_stage3() { let workload = sql_workload( vec![ batch("SELECT COUNT(DISTINCT l_orderkey) FROM lineitem"), @@ -128,11 +128,17 @@ async fn facade_plans_match_plan_stages() { targets.push(Some(accuracy)); } let data = workload.data_workload.clone().unwrap(); - let run = plan_stages(roots, &targets, &data, models).unwrap(); - let selection = run.selection.as_ref().expect("a candidate is selected"); + let run = run_stage1_to_stage3(roots, &targets, &data, models).unwrap(); + let selection = run + .stage3_plan_selection + .as_ref() + .expect("a candidate is selected"); let selected = run.selected().expect("the selected candidate is listed"); - assert!(run.physical.len() > 1, "the stages chose among candidates"); + assert!( + run.stage2_physical_asap_optimization.len() > 1, + "the stages chose among candidates" + ); assert_eq!(output.selection.as_ref(), Some(selection)); assert_eq!(output.roots(), selected.roots); } @@ -222,7 +228,7 @@ async fn harness_rejects_a_pass_that_mislabels_entry_indices() { "mangling" } fn optimize(&self, input: OptimizationInput<'_>) -> Result { - let mut output = StagePipeline.optimize(input)?; + let mut output = Stage1ToStage3Pass.optimize(input)?; for plan in output.plans.iter_mut() { plan.entry_index += 1; } diff --git a/crates/planner/tests/plan_stages.rs b/crates/planner/tests/plan_stages.rs index f2877a9fe..5e1d8e14a 100644 --- a/crates/planner/tests/plan_stages.rs +++ b/crates/planner/tests/plan_stages.rs @@ -1,7 +1,7 @@ -//! The planner pipeline ([`plan_stages`]): the planning stages in order over +//! The planner pipeline ([`run_stage1_to_stage3`]): the planning stages in order over //! the frontends' roots, enumerating every whole-workload candidate. -use asap_aware_mapping::pass::{plan_stages, PlanningModels, StagePipelineRun}; +use asap_aware_mapping::pass::{run_stage1_to_stage3, PlanningModels, Stage1ToStage3Run}; use asap_frontend_promql::lower_promql_query_workload; use asap_types::types::AccuracyTarget; use asap_types::workload::{ @@ -11,7 +11,7 @@ use asap_types::workload::{ const NOW_MS: u64 = 1_700_000_000_000; -fn promql_stages(query: &str, epsilon: f64) -> StagePipelineRun { +fn promql_stages(query: &str, epsilon: f64) -> Stage1ToStage3Run { let target = AccuracyTarget::Epsilon(epsilon); let workload = PlanningWorkload { query_workload: QueryWorkload { @@ -38,7 +38,7 @@ fn promql_stages(query: &str, epsilon: f64) -> StagePipelineRun { }), }; let roots = lower_promql_query_workload(&workload, NOW_MS).expect("lowers"); - plan_stages( + run_stage1_to_stage3( roots.into_iter().enumerate().collect(), &[Some(target)], &workload.data_workload.unwrap(), @@ -54,23 +54,32 @@ fn promql_stages(query: &str, epsilon: f64) -> StagePipelineRun { fn enumerates_every_combination_and_selects_one() { let run = promql_stages("count(a) + count(b)", 0.01); let sizes: Vec<_> = run - .per_computation + .stage1_pass1_per_computation .targets .iter() .map(|target| target.choices.candidates.len()) .collect(); assert_eq!(sizes.len(), 2); assert_eq!(run.combinations, sizes[0] * sizes[1]); - assert_eq!(run.logical.len(), run.combinations); - assert_eq!(run.logical[0].choice, vec![0, 0]); - assert_eq!(run.logical[1].choice, vec![0, 1]); - assert_eq!(run.logical[sizes[1]].choice, vec![1, 0]); + assert_eq!(run.stage1_logical_asap_optimization.len(), run.combinations); + assert_eq!(run.stage1_logical_asap_optimization[0].choice, vec![0, 0]); + assert_eq!(run.stage1_logical_asap_optimization[1].choice, vec![0, 1]); assert_eq!( - run.physical.len() + run.pruned.len(), - run.logical.iter().filter(|l| l.roots.is_ok()).count() + run.stage1_logical_asap_optimization[sizes[1]].choice, + vec![1, 0] ); - let selection = run.selection.as_ref().unwrap(); + assert_eq!( + run.stage2_physical_asap_optimization.len() + run.stage2_pruned.len(), + run.stage1_logical_asap_optimization + .iter() + .filter(|l| l.roots.is_ok()) + .count() + ); + let selection = run.stage3_plan_selection.as_ref().unwrap(); let selected = run.selected().expect("the selected candidate is listed"); assert_eq!(selected.id, selection.selected); - assert_eq!(selection.rejected.len(), run.physical.len() - 1); + assert_eq!( + selection.rejected.len(), + run.stage2_physical_asap_optimization.len() - 1 + ); } diff --git a/crates/planner/tests/summary_sharing.rs b/crates/planner/tests/summary_sharing.rs index 544786ec7..3fb9c19cc 100644 --- a/crates/planner/tests/summary_sharing.rs +++ b/crates/planner/tests/summary_sharing.rs @@ -8,7 +8,7 @@ use asap_types::ir::{ASAPOp, OperatorNode, QueryRoot}; use std::rc::Rc; use asap_aware_mapping::pass::{ - plan_stages, PlanOutput, PlanningModels, QueryPlan, StagePipelineRun, + run_stage1_to_stage3, PlanOutput, PlanningModels, QueryPlan, Stage1ToStage3Run, }; use asap_aware_mapping::replacement::{default_size_params, DEFAULT_DELTA}; use asap_frontend_promql::lower_promql_query_workload; @@ -85,14 +85,14 @@ async fn plan_promql(queries: &[(&str, f64)]) -> PlanOutput { } /// Every planning stage's output for `queries`, run over the frontend's roots. -fn promql_stages(queries: &[(&str, f64)]) -> StagePipelineRun { +fn promql_stages(queries: &[(&str, f64)]) -> Stage1ToStage3Run { let workload = promql_workload(queries); let roots = lower_promql_query_workload(&workload, NOW_MS).expect("lowers"); let targets: Vec<_> = queries .iter() .map(|(_, epsilon)| Some(AccuracyTarget::Epsilon(*epsilon))) .collect(); - plan_stages( + run_stage1_to_stage3( roots.into_iter().enumerate().collect(), &targets, &workload.data_workload.clone().unwrap(), @@ -434,14 +434,14 @@ fn combined_scalar_references_keep_summary_candidates() { 0.01, )]); let both = run - .physical + .stage2_physical_asap_optimization .iter() .find(|candidate| { let refs = root_operators(&candidate.roots[0]); refs.len() == 2 && refs.into_iter().all(reads_sketch) }) .expect("a candidate reads a sketch in both references"); - let selection = run.selection.as_ref().unwrap(); + let selection = run.stage3_plan_selection.as_ref().unwrap(); let rejection = selection .rejected .iter() @@ -467,7 +467,7 @@ async fn combined_vector_quantiles_keep_unselected_summary_candidates() { ] { let run = promql_stages(&[(query, 0.01)]); let summary = run - .physical + .stage2_physical_asap_optimization .iter() .find(|candidate| { root_operators(&candidate.roots[0]) @@ -480,7 +480,7 @@ async fn combined_vector_quantiles_keep_unselected_summary_candidates() { }; assert!(root.guarantee.is_none(), "{query}"); assert_ne!( - run.selection.as_ref().unwrap().selected, + run.stage3_plan_selection.as_ref().unwrap().selected, summary.id, "{query}" ); diff --git a/docs/design_docs/architecture/updated_interface_with_pluggable_optimization.md b/docs/design_docs/architecture/updated_interface_with_pluggable_optimization.md index 2a04dee3b..a283fb0be 100644 --- a/docs/design_docs/architecture/updated_interface_with_pluggable_optimization.md +++ b/docs/design_docs/architecture/updated_interface_with_pluggable_optimization.md @@ -43,7 +43,7 @@ flowchart TD direction TB L["lowering"] O["OptimizationInput"] - PASS["OptimizationPass: StagePipeline, or another implementation"] + PASS["OptimizationPass: Stage1ToStage3Pass, or another implementation"] L --> O --> PASS end @@ -62,8 +62,8 @@ Details of these types are provided below. |---|---| | `workload` | `&PlanningWorkload` | | `frontend_specific` | `Sql { catalog }` / `Promql { now_ms, histograms }` / `Metricsql`; fixed by `query_workload.language` | -| `models` | Cost model, accuracy model, evidence provider; `PlanningModels::builtin()` for the defaults. `StagePipeline` reads the accuracy model and the evidence provider, not the cost model | -| `pass` | `None` uses `StagePipeline` | +| `models` | Cost model, accuracy model, evidence provider; `PlanningModels::builtin()` for the defaults. `Stage1ToStage3Pass` reads the accuracy model and the evidence provider, not the cost model | +| `pass` | `None` uses `Stage1ToStage3Pass` | ### `OptimizationInput` @@ -92,7 +92,7 @@ pub struct QueryPlan { } ``` -`StagePipeline` returns the selected physical candidate's roots, with every +`Stage1ToStage3Pass` returns the selected physical candidate's roots, with every sub-DAG `NotMaterialized`; `PlanOutput::execution_timed_dag()` exports them with that same assignment. Physical ASAP-aware optimization decides per sub-DAG whether to materialize, and whether at ingestion or query time, from @@ -103,26 +103,27 @@ sub-DAG whether to materialize, and whether at ingestion or query time, from ## 3. The pluggable optimization pass The optimization pass is fully pluggable, as long as the end-to-end behavior is satisfied. -The `StagePipeline` described below is used by default. +The `Stage1ToStage3Pass` described below is used by default. -### 3.1 `StagePipeline` — the planning stages +### 3.1 `Stage1ToStage3Pass` — Stage 1 to Stage 3 -`StagePipeline` runs the planning stages of +`Stage1ToStage3Pass` runs Stage 1 (logical ASAP-aware optimization), Stage 2 +(physical ASAP-aware optimization) and Stage 3 (plan selection) of [planner-layering.md](../proposals/planner-layering.md) in order and is -registered under the name `stage-pipeline`. It replaced `MajorPass`, the +registered under the name `stage1-to-stage3`. It replaced `MajorPass`, the legacy replacement search; the coverage this gives up until Pass 2 is tracked in #580. | Step | Call | |---|---| | Prepare roots | Entries in `entries()` order, the accuracy target from each entry's `requirements`; PromQL operator roots carry the series identity (`with_promql_series_identity`) where it has a realization | -| Logical ASAP-aware optimization | `generate_per_computation_candidates` (Pass 1), then `compose_logical_candidate` for every combination of per-computation candidates | -| Physical ASAP-aware optimization | `optimize_physical_asap` for every composed candidate | -| Plan selection | `select_plan`: reject candidates that miss an accuracy target, price the rest, select the cheapest | +| Stage 1: logical ASAP-aware optimization | `generate_per_computation_candidates` (Pass 1), then `compose_logical_candidate` for every combination of per-computation candidates | +| Stage 2: physical ASAP-aware optimization | `optimize_physical_asap` for every composed candidate | +| Stage 3: plan selection | `select_plan`: reject candidates that miss an accuracy target, price the rest, select the cheapest | -`plan_stages` runs these steps over the frontends' roots and returns every -stage's candidates as well as the selection; the `stage_pipeline` devtool -shows them. Every combination is enumerated and plan selection prices +`run_stage1_to_stage3` runs these steps over the query-language frontends' +roots (Stage 0) and returns every stage's candidates as well as the +selection; the `stage_pipeline` devtool shows them. Every combination is enumerated and plan selection prices every candidate; nothing is capped. A search that avoids building every candidate is future work (#671). @@ -157,7 +158,7 @@ let output = e2e_plan(user_input.with_pass(&my_pass)).await?; // the whole pipe Or through an optimization pass registry: ```rust -let mut registry = PassRegistry::with_builtin(); // holds "stage-pipeline" +let mut registry = PassRegistry::with_builtin(); // holds "stage1-to-stage3" registry.register(Box::new(my_pass))?; for name in registry.names() { optimize(registry.get(name).unwrap(), optimization_input)?; @@ -227,7 +228,7 @@ pass no longer runs these steps; another pass need not run any of them. | Crate | What it holds | |---|---| | `asap-types` | `ParsedWorkload` | -| `asap-aware-mapping` | `OptimizationPass`, `OptimizationInput`, `PlanOutput`, `PlanningModels`, `optimize`, `PassRegistry`, `StagePipeline`, `plan_stages` | +| `asap-aware-mapping` | `OptimizationPass`, `OptimizationInput`, `PlanOutput`, `PlanningModels`, `optimize`, `PassRegistry`, `Stage1ToStage3Pass`, `run_stage1_to_stage3` | | `asap-planner` *(new)* | `e2e_plan`, `UserInput`, `FrontendInput`, lowering dispatch | ```text diff --git a/tools/dag-viewer/examples/planner-layering-example1.json b/tools/dag-viewer/examples/planner-layering-example1.json index 4b365ab26..3ee848f6a 100644 --- a/tools/dag-viewer/examples/planner-layering-example1.json +++ b/tools/dag-viewer/examples/planner-layering-example1.json @@ -1411,8 +1411,8 @@ } ] }, - "id": "L1", - "label": "Q1 exact (Sum acc, Rate acc) · Q2 exact (Sum acc)" + "display_name": "Q1 exact (Sum acc, Rate acc) · Q2 exact (Sum acc)", + "id": "L1" }, { "dag": { @@ -2375,8 +2375,8 @@ } ] }, - "id": "L2", - "label": "Q1 exact (Sum acc, Rate acc) · Q2 CMS+heap (Sum acc)" + "display_name": "Q1 exact (Sum acc, Rate acc) · Q2 CMS+heap (Sum acc)", + "id": "L2" }, { "dag": { @@ -3339,8 +3339,8 @@ } ] }, - "id": "L3", - "label": "Q1 exact (Sum acc, Rate acc) · Q2 CountSketch+heap (Sum acc)" + "display_name": "Q1 exact (Sum acc, Rate acc) · Q2 CountSketch+heap (Sum acc)", + "id": "L3" } ], "capped": false, @@ -4919,9 +4919,9 @@ ], "scalar_roots": [] }, + "display_name": "Q1 exact (Sum acc, Rate acc) · Q2 exact (Sum acc)", "from_logical": "L1", - "id": "P1", - "label": "Q1 exact (Sum acc, Rate acc) · Q2 exact (Sum acc)" + "id": "P1" }, { "dag": { @@ -6458,9 +6458,9 @@ ], "scalar_roots": [] }, + "display_name": "Q1 exact (Sum acc, Rate acc) · Q2 CMS+heap (Sum acc)", "from_logical": "L2", - "id": "P2", - "label": "Q1 exact (Sum acc, Rate acc) · Q2 CMS+heap (Sum acc)" + "id": "P2" }, { "dag": { @@ -7997,9 +7997,9 @@ ], "scalar_roots": [] }, + "display_name": "Q1 exact (Sum acc, Rate acc) · Q2 CountSketch+heap (Sum acc)", "from_logical": "L3", - "id": "P3", - "label": "Q1 exact (Sum acc, Rate acc) · Q2 CountSketch+heap (Sum acc)" + "id": "P3" } ], "pruned": []