From 4150cbb71a5c391a97185adb42599c27a79ea25c Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Mon, 2 Nov 2026 16:59:47 +0000 Subject: [PATCH 1/2] feat(devtools): print_planning_stages prints every planner stage's output as text Adds asap_devtools::planning_stages_text and the print_planning_stages bin: the frontends' roots, the memo groups, Pass 1's PerComputationLogicalMemoCandidates, the memo after Pass 2's local window composition, LogicalASAPMemoCandidates, PhysicalASAPMemoCandidates and Stage 3's Selection for one PromQL workload. node_label moves from stage_pipeline to candidate_labels so both share it, and now names a pane chain's layout (pane_width_ms, pane_count). Co-Authored-By: Claude Opus 5.5 --- .../devtools/src/bin/print_planning_stages.rs | 55 +++ crates/devtools/src/bin/stage_pipeline.rs | 106 +---- crates/devtools/src/candidate_labels.rs | 112 +++++- crates/devtools/src/lib.rs | 1 + crates/devtools/src/planning_stages_text.rs | 364 ++++++++++++++++++ 5 files changed, 532 insertions(+), 106 deletions(-) create mode 100644 crates/devtools/src/bin/print_planning_stages.rs create mode 100644 crates/devtools/src/planning_stages_text.rs diff --git a/crates/devtools/src/bin/print_planning_stages.rs b/crates/devtools/src/bin/print_planning_stages.rs new file mode 100644 index 000000000..fd77bceab --- /dev/null +++ b/crates/devtools/src/bin/print_planning_stages.rs @@ -0,0 +1,55 @@ +// cargo run -p asap-devtools --bin print_planning_stages +// cargo run -p asap-devtools --bin print_planning_stages -- \ +// 'sum by (job) (rate(http_requests_total[1m]))' \ +// 'sum by (job) (sum_over_time(http_requests_total[5m]))' +// +// Prints every planner stage's output as text +// (`asap_devtools::planning_stages_text`): the frontends' roots, the memo +// groups, Pass 1's `PerComputationLogicalMemoCandidates`, the memo after +// Pass 2's local window composition, Pass 2's `LogicalASAPMemoCandidates`, +// Stage 2's `PhysicalASAPMemoCandidates` and Stage 3's `Selection`. +// +// Without arguments it plans #509 Example 1. Each argument is a PromQL +// query that replaces Example 1's queries, keeping its data facts +// (`http_requests_total`, a counter) and its panels' requirements: exact, +// every 10 s. + +use asap_devtools::planner_layering_examples::example1; +use asap_devtools::planning_stages_text::{plan_promql_with_stage_outputs, planning_stages_text}; +use asap_types::workload::{Query, RepeatingEntry}; + +fn main() { + let queries: Vec = std::env::args().skip(1).collect(); + // Example 1's workload; its queries replaced when given. + let mut workload = example1(); + if !queries.is_empty() { + let repeating = workload + .query_workload + .repeating_queries + .as_mut() + .expect("Example 1 has repeating queries"); + // Example 1's first panel: exact, every 10 s. + let panel = repeating[0].clone(); + *repeating = queries + .iter() + .map(|query| RepeatingEntry { + query: Query(query.clone()), + ..panel.clone() + }) + .collect(); + } + println!("# Workload"); + for (index, entry) in workload.query_workload.entries().enumerate() { + println!( + " q{index}: {} accuracy={:?}", + entry.query.0, entry.requirements.accuracy + ); + } + match plan_promql_with_stage_outputs(&workload) { + Ok(stages) => print!("{}", planning_stages_text(&stages)), + Err(error) => { + eprintln!("print_planning_stages: {error}"); + std::process::exit(1); + } + } +} diff --git a/crates/devtools/src/bin/stage_pipeline.rs b/crates/devtools/src/bin/stage_pipeline.rs index 0e019c74b..17865b922 100644 --- a/crates/devtools/src/bin/stage_pipeline.rs +++ b/crates/devtools/src/bin/stage_pipeline.rs @@ -264,7 +264,7 @@ use asap_cost::{ ClusterLimits, ClusterObjective, CostInputs, CostModel, MeasuredOperationCosts, ResourceUnitCost, ResourceUsage, }; -use asap_devtools::candidate_labels::{candidate_labels, computation_label}; +use asap_devtools::candidate_labels::{candidate_labels, computation_label, node_label}; 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, @@ -1896,110 +1896,6 @@ fn candidate_node_label( labels.join(" ← ") } -/// One node's operator kind with its sketch family and parameters, its -/// summary query and its time range, when it has them: e.g. -/// `SummaryAgg Kll(k=200)`, `SummaryEstimate Quantile(q=0.5)`. -fn node_label(node: &asap_types::ir::flat::FlatNode) -> String { - let operator = serde_json::to_value(&node.operator).unwrap_or_default(); - // `{"ASAP": {"SummaryAgg": {...}}}` (or a unit variant's string). - let (kind, fields) = match operator.as_object().and_then(|o| o.values().next()) { - Some(Value::Object(inner)) => match inner.iter().next() { - Some((kind, fields)) => (kind.clone(), fields.clone()), - None => (String::new(), Value::Null), - }, - Some(Value::String(kind)) => (kind.clone(), Value::Null), - _ => (operator.to_string(), Value::Null), - }; - let mut details = Vec::new(); - find_details(&fields, &mut details, 0); - let mut label = kind; - for detail in details { - label.push(' '); - label.push_str(&detail); - } - label -} - -/// Sketch algorithms with parameters, summary queries and time ranges found -/// in an operator's fields (not inside its schemas). -fn find_details(value: &Value, details: &mut Vec, depth: usize) { - if depth > 6 || details.len() >= 3 { - return; - } - let compact = |value: &Value| -> String { - match value { - Value::Object(fields) => fields - .iter() - .map(|(key, value)| match value { - Value::Object(inner) if inner.len() == 1 => { - let (k, v) = inner.iter().next().unwrap(); - format!("{key}({k}={v})") - } - Value::Object(inner) => format!( - "{key}({})", - inner - .iter() - .filter(|(_, v)| !v.is_object() && !v.is_array()) - .map(|(k, v)| format!("{k}={v}")) - .collect::>() - .join(", ") - ), - other => format!("{key}={other}"), - }) - .collect::>() - .join(" "), - other => other.to_string(), - } - }; - match value { - Value::Object(fields) => { - if let (Some(Value::String(algorithm)), Some(params)) = - (fields.get("algorithm"), fields.get("params")) - { - let params = params - .as_object() - .and_then(|p| p.values().next()) - .map(compact) - .unwrap_or_default(); - details.push(format!("{algorithm}({params})")); - return; - } - for (key, field) in fields { - match key.as_str() { - "schema" | "input_schema" | "output_schema" | "predicates" | "cols" => {} - "query" => details.push(compact(field)), - // An aggregate's measures, e.g. `quantile(q=0.5)`. - "measures" => { - for measure in field.as_array().into_iter().flatten() { - let Some(kind) = measure.get("kind").and_then(Value::as_str) else { - continue; - }; - let params: Vec<_> = measure - .as_object() - .into_iter() - .flatten() - .filter(|(key, value)| { - *key != "kind" && (value.is_number() || value.is_string()) - }) - .map(|(key, value)| format!("{key}={value}")) - .collect(); - details.push(format!("{kind}({})", params.join(", "))); - } - } - "range" | "window" => details.push(format!("{key}={field}")), - _ => find_details(field, details, depth + 1), - } - } - } - Value::Array(items) => { - for item in items { - find_details(item, details, depth + 1); - } - } - _ => {} - } -} - #[cfg(test)] mod tests { use super::*; diff --git a/crates/devtools/src/candidate_labels.rs b/crates/devtools/src/candidate_labels.rs index c80932437..766040fe5 100644 --- a/crates/devtools/src/candidate_labels.rs +++ b/crates/devtools/src/candidate_labels.rs @@ -1,11 +1,14 @@ //! Display labels for the planning stages' candidates, for the -//! `stage_pipeline` viewer document only. Planning never reads them: they +//! `stage_pipeline` viewer document and the `print_planning_stages` text +//! only. Planning never reads them: they //! are pure functions of the candidates' planning data (the logical roots //! and the materialization assignment). use std::collections::BTreeMap; use std::rc::Rc; +use serde_json::Value; + use asap_types::ir::physical_export::{PhysicalASAPDAG, PhysicalASAPNodeId}; use asap_types::ir::schema::{FieldDataType, SketchAlgorithm}; use asap_types::ir::NonASAPOp; @@ -263,6 +266,113 @@ fn root_operators(root: &QueryRoot) -> Vec<&Rc> { } } +/// One node's operator kind with its sketch family and parameters, its +/// summary query, its time range and its pane layout, when it has them: e.g. +/// `SummaryAgg Kll(k=200)`, `SummaryEstimate Quantile(q=0.5)`. +pub fn node_label(node: &asap_types::ir::flat::FlatNode) -> String { + let operator = serde_json::to_value(&node.operator).unwrap_or_default(); + // `{"ASAP": {"SummaryAgg": {...}}}` (or a unit variant's string). + let (kind, fields) = match operator.as_object().and_then(|o| o.values().next()) { + Some(Value::Object(inner)) => match inner.iter().next() { + Some((kind, fields)) => (kind.clone(), fields.clone()), + None => (String::new(), Value::Null), + }, + Some(Value::String(kind)) => (kind.clone(), Value::Null), + _ => (operator.to_string(), Value::Null), + }; + let mut details = Vec::new(); + find_details(&fields, &mut details, 0); + let mut label = kind; + for detail in details { + label.push(' '); + label.push_str(&detail); + } + label +} + +/// Sketch algorithms with parameters, summary queries and time ranges found +/// in an operator's fields (not inside its schemas). +fn find_details(value: &Value, details: &mut Vec, depth: usize) { + if depth > 6 || details.len() >= 4 { + return; + } + let compact = |value: &Value| -> String { + match value { + Value::Object(fields) => fields + .iter() + .map(|(key, value)| match value { + Value::Object(inner) if inner.len() == 1 => { + let (k, v) = inner.iter().next().unwrap(); + format!("{key}({k}={v})") + } + Value::Object(inner) => format!( + "{key}({})", + inner + .iter() + .filter(|(_, v)| !v.is_object() && !v.is_array()) + .map(|(k, v)| format!("{k}={v}")) + .collect::>() + .join(", ") + ), + other => format!("{key}={other}"), + }) + .collect::>() + .join(" "), + other => other.to_string(), + } + }; + match value { + Value::Object(fields) => { + if let (Some(Value::String(algorithm)), Some(params)) = + (fields.get("algorithm"), fields.get("params")) + { + let params = params + .as_object() + .and_then(|p| p.values().next()) + .map(compact) + .unwrap_or_default(); + details.push(format!("{algorithm}({params})")); + return; + } + for (key, field) in fields { + match key.as_str() { + "schema" | "input_schema" | "output_schema" | "predicates" | "cols" => {} + "query" => details.push(compact(field)), + // An aggregate's measures, e.g. `quantile(q=0.5)`. + "measures" => { + for measure in field.as_array().into_iter().flatten() { + let Some(kind) = measure.get("kind").and_then(Value::as_str) else { + continue; + }; + let params: Vec<_> = measure + .as_object() + .into_iter() + .flatten() + .filter(|(key, value)| { + *key != "kind" && (value.is_number() || value.is_string()) + }) + .map(|(key, value)| format!("{key}={value}")) + .collect(); + details.push(format!("{kind}({})", params.join(", "))); + } + } + // A window's range, and a pane chain's layout. + "range" | "window" | "pane_width_ms" | "pane_count" => { + details.push(format!("{key}={field}")) + } + _ => find_details(field, details, depth + 1), + } + } + } + Value::Array(items) => { + for item in items { + find_details(item, details, depth + 1); + } + } + _ => {} + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/devtools/src/lib.rs b/crates/devtools/src/lib.rs index ba88ae702..6a88072e6 100644 --- a/crates/devtools/src/lib.rs +++ b/crates/devtools/src/lib.rs @@ -27,6 +27,7 @@ pub mod memo_pricing; pub mod planner_fuzz; pub mod planner_layering_examples; pub mod planning_stages; +pub mod planning_stages_text; pub mod rqe_atomic_costs; pub use asap_frontend_promql::{ diff --git a/crates/devtools/src/planning_stages_text.rs b/crates/devtools/src/planning_stages_text.rs new file mode 100644 index 000000000..2a35856a5 --- /dev/null +++ b/crates/devtools/src/planning_stages_text.rs @@ -0,0 +1,364 @@ +//! Every planner stage's output as plain text, for reading one workload's +//! planning end to end (`print_planning_stages`): the frontends' roots +//! (Stage 0), the memo groups (Stage 0 → Stage 1), Pass 1's +//! `PerComputationLogicalMemoCandidates`, the memo after Pass 2's local +//! window composition, Pass 2's `LogicalASAPMemoCandidates` (Stage 2's +//! input), Stage 2's `PhysicalASAPMemoCandidates` (Stage 3's input) and +//! Stage 3's `Selection`. Planning never reads it. +//! +//! Notation: `G` is memo group `n`; `G[]` its candidate `i`; +//! `O` Stage 2's operator node `n`; `C` physical candidate `n`; a +//! candidate's chain lists its own nodes root first, down to its input +//! groups. + +use std::collections::HashSet; +use std::fmt::Write; +use std::rc::Rc; + +use asap_planner::{PlanSearchSettings, PlanningModels}; +use asap_types::ir::flat::flatten; +use asap_types::ir::{OperatorNode, QueryRoot}; +use asap_types::stages::{MemoCandidates, MemoGroupId, PhysicalASAPMemoCandidates}; +use asap_types::workload::PlanningWorkload; + +use crate::candidate_labels::node_label; +use crate::planning_stages::{ + lower_promql_roots, run_search_over, PlanningStagesError, PlanningStagesRun, +}; + +/// Plans the PromQL `workload` as the facade does, keeping every stage's +/// memo (`PlanSearchSettings::inspect`), without the reference enumeration. +pub fn plan_promql_with_stage_outputs( + workload: &PlanningWorkload, +) -> Result { + let roots = lower_promql_roots(workload)?; + let entries: Vec<_> = workload.query_workload.entries().collect(); + let data = workload.data_workload.clone().unwrap_or_default(); + run_search_over( + roots, + &entries, + &data, + PlanningModels::builtin(), + PlanSearchSettings { + inspect: true, + ..PlanSearchSettings::default() + }, + None, + ) + .map_err(|error: PlanningStagesError| error.to_string()) +} + +/// Every stage's output of `stages` (planned with +/// `PlanSearchSettings::inspect`) as text, one section per stage boundary. +/// A stage that did not run is reported as missing. +pub fn planning_stages_text(stages: &PlanningStagesRun) -> String { + let mut out = String::new(); + let run = &stages.run; + + section( + &mut out, + "Stage 0 output: the frontends' roots (Vec)", + ); + let (flat, _) = flatten(&stages.stage0_query_language_frontends); + for (id, node) in flat.nodes.iter().enumerate() { + let children: Vec = node + .operator + .children() + .into_iter() + .map(|child| format!("n{child}")) + .collect(); + let _ = writeln!(out, " n{id}: {} <- {children:?}", node_label(node)); + } + let _ = writeln!(out, " roots: {:?}", flat.roots); + + let Some(pass1) = &run.per_computation_candidates else { + let _ = writeln!(out, "\n(no Pass 1 output)"); + return out; + }; + section( + &mut out, + "Stage 0 → Stage 1: DagOfLogicalMemoGroups (the roots hash-consed)", + ); + write_memo(&mut out, &pass1.memo, false); + + section( + &mut out, + "Stage 1 Pass 1 output: PerComputationLogicalMemoCandidates", + ); + write_memo(&mut out, &pass1.memo, true); + let _ = writeln!(out, " computation_targets:"); + for target in &pass1.computation_targets { + let _ = writeln!( + out, + " G{} measure {}: ways_to_compute={:?} fused={:?}", + target.aggregate_group.0, + target.measure_index, + target.ways_to_compute, + target + .fused_computation + .as_ref() + .map(|fused| &fused.ways_to_compute) + ); + } + let _ = writeln!(out, " pruned_candidates:"); + for pruned in &pass1.pruned_candidates { + let _ = writeln!( + out, + " G{}: {:?} reason={:?}", + pruned.group.0, pruned.candidate, pruned.reason + ); + } + + if let Some(steps) = &run.memo_steps { + section( + &mut out, + "Stage 1 Pass 2, after local window composition (pane candidates)", + ); + write_memo(&mut out, &steps.pass2, true); + } + match &run.logical_asap_candidates { + Some(logical) => { + section( + &mut out, + "Stage 1 Pass 2 output = Stage 2 input: LogicalASAPMemoCandidates", + ); + write_memo(&mut out, &logical.memo, true); + } + None => { + let _ = writeln!(out, "\n(no Pass 2 output)"); + } + } + match &run.physical_asap_candidates { + Some(physical) => { + section( + &mut out, + "Stage 2 output = Stage 3 input: PhysicalASAPMemoCandidates", + ); + write_physical(&mut out, physical); + } + None => { + let _ = writeln!(out, "\n(no Stage 2 output)"); + } + } + + section(&mut out, "Stage 3 output: Selection"); + match &run.selection { + Ok(selection) => { + let dag = &selection.selected_dag; + for (node, candidate) in dag.candidate_of_node.iter().enumerate() { + let inputs: Vec = dag + .edges + .iter() + .filter(|edge| edge.consumer.0 == node) + .map(|edge| format!("input {} <- node{}", edge.input, edge.producer.0)) + .collect(); + let _ = writeln!(out, " node{node}: runs C{} {inputs:?}", candidate.0); + } + let outputs: Vec = dag.node_of_output_reference.iter().map(|n| n.0).collect(); + let _ = writeln!(out, " node_of_output_reference: {outputs:?}"); + let _ = writeln!( + out, + " cost_per_second: {:?}", + selection.cost.cost_per_second + ); + } + Err(error) => { + let _ = writeln!(out, " error: {error}"); + } + } + out +} + +fn section(out: &mut String, title: &str) { + let _ = writeln!(out, "\n==================== {title} ===================="); +} + +fn group_names(groups: &[MemoGroupId]) -> Vec { + groups.iter().map(|group| format!("G{}", group.0)).collect() +} + +/// `memo`'s groups, and with `with_candidates` each group's candidates. +fn write_memo(out: &mut String, memo: &MemoCandidates, with_candidates: bool) { + for (id, group) in memo.groups.iter().enumerate() { + let _ = writeln!( + out, + " G{id}: {} children={:?} parents={:?} queries={:?}", + root_label(&group.logical_node), + group_names(&group.child_groups), + group_names(group.parent_groups()), + group.reached_by_queries() + ); + if !with_candidates { + continue; + } + for (index, candidate) in memo.candidates[id].iter().enumerate() { + let fuses = if candidate.fused_groups.is_empty() { + String::new() + } else { + format!(" fuses={:?}", group_names(&candidate.fused_groups)) + }; + let _ = writeln!( + out, + " [{index}] {:<12} {} inputs={:?}{fuses} accuracy_budget_parts={}", + format!("{:?}", candidate.origin), + chain_label(memo, &candidate.root_node, &candidate.input_groups), + group_names(&candidate.input_groups), + candidate.accuracy_budget_parts + ); + } + } + let roots: Vec = memo + .query_roots + .iter() + .map(|root| match root { + QueryRoot::Operator(group) => format!("G{}", group.0), + scalar => format!("{scalar:?}"), + }) + .collect(); + let _ = writeln!(out, " query_roots: {roots:?}"); +} + +fn write_physical(out: &mut String, physical: &PhysicalASAPMemoCandidates) { + let nodes = &physical.operator_nodes_of_candidates_for_memo_groups; + let _ = writeln!( + out, + " operator_nodes_of_candidates_for_memo_groups ({}):", + nodes.len() + ); + for (id, node) in nodes.iter().enumerate() { + let implements: Vec = node + .candidates_for_memo_group + .iter() + .map(|candidate| format!("G{}[{}]", candidate.group.0, candidate.index)) + .collect(); + let _ = writeln!( + out, + " O{id}: {} implements={implements:?} is_implementation_root={} accuracy_budget_parts={}", + root_label(&node.physical_node), + node.is_implementation_root, + node.accuracy_budget_parts + ); + } + let _ = writeln!(out, " candidates ({}):", physical.candidates.len()); + for (id, candidate) in physical.candidates.iter().enumerate() { + let producers: Vec = candidate + .producer_candidates_per_input + .iter() + .map(|producers| { + let ids: Vec = producers.iter().map(|p| format!("C{}", p.0)).collect(); + format!("{{{}}}", ids.join(",")) + }) + .collect(); + let _ = writeln!( + out, + " C{id}: O{} {:?} {:?} producers per input: [{}]{}", + candidate.operator_node.0, + candidate.timing, + candidate.materialization, + producers.join(" "), + if candidate.may_produce_query_output { + " (may produce a query output)" + } else { + "" + } + ); + } + let _ = writeln!(out, " query_roots: {:?}", physical.query_roots); + let _ = writeln!( + out, + " reader_constraints ({}):", + physical.reader_constraints.len() + ); + for constraint in &physical.reader_constraints { + let _ = writeln!(out, " {constraint:?}"); + } + let _ = writeln!( + out, + " pruned_candidates ({}):", + physical.pruned_candidates.len() + ); + for pruned in &physical.pruned_candidates { + let key = &pruned.candidate; + let _ = writeln!( + out, + " G{} O{} {:?} {:?} reason={:?}", + pruned.group.0, key.operator_node.0, key.timing, key.materialization, pruned.reason + ); + } +} + +/// The label of `node`'s own operator, without its children. +fn root_label(node: &Rc) -> String { + let (flat, _) = flatten(&[QueryRoot::Operator(node.clone())]); + flat.nodes.last().map(node_label).unwrap_or_default() +} + +/// A candidate's own nodes, from `root` down to its `input_groups`' nodes +/// (excluded), root first, joined by " ← ". +fn chain_label( + memo: &MemoCandidates, + root: &Rc, + input_groups: &[MemoGroupId], +) -> String { + let (flat, originals) = flatten(&[QueryRoot::Operator(root.clone())]); + let is_input = |id: usize| { + input_groups + .iter() + .any(|group| Rc::ptr_eq(&memo.groups[group.0 as usize].logical_node, &originals[id])) + }; + let mut labels = Vec::new(); + let mut pending = vec![flat.nodes.len() - 1]; + let mut seen = HashSet::new(); + while let Some(id) = pending.pop() { + if is_input(id) || !seen.insert(id) { + continue; + } + labels.push(node_label(&flat.nodes[id])); + pending.extend( + flat.nodes[id] + .operator + .children() + .into_iter() + .rev() + .copied(), + ); + } + labels.join(" ← ") +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::planner_layering_examples::example1; + use asap_types::workload::Query; + + /// One repeating `sum(rate(...[1m]))` query: every section is printed, + /// Pass 2 adds pane candidates with their layout, and Stage 2 offers the + /// pane chain at ingestion time. + #[test] + fn prints_every_stage_of_a_repeating_rate_query() { + let mut workload = example1(); + let queries = workload.query_workload.repeating_queries.as_mut().unwrap(); + queries.truncate(1); + queries[0].query = Query("sum by (job) (rate(http_requests_total[1m]))".into()); + + let text = planning_stages_text(&plan_promql_with_stage_outputs(&workload).unwrap()); + + for heading in [ + "Stage 0 output", + "Stage 0 → Stage 1: DagOfLogicalMemoGroups", + "Stage 1 Pass 1 output: PerComputationLogicalMemoCandidates", + "Stage 1 Pass 2, after local window composition", + "Stage 2 input: LogicalASAPMemoCandidates", + "Stage 3 input: PhysicalASAPMemoCandidates", + "Stage 3 output: Selection", + ] { + assert!(text.contains(heading), "missing {heading}:\n{text}"); + } + assert!(text.contains( + "Panes FinalizeExactAccumulator ← SummaryMerge ← PaneChain pane_count=6 pane_width_ms=10000" + )); + assert!(text.contains("IngestionTime MatAtIngestion(Memory)")); + assert!(text.contains("cost_per_second: Some(")); + } +} From 0e8c92812d6fd720f0c66c0369dab24810b429df Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Mon, 2 Nov 2026 17:22:42 +0000 Subject: [PATCH 2/2] feat(devtools): stage_pipeline --text replaces the separate print bin One binary plans the workload; --text writes the run as text instead of the viewer document, so both outputs come from the same inputs and flags. The text output, now in the aligned format (G1: TimeRange 60s <- G0), also covers what tools/dag-viewer/print_stages.py (#1158) prints: candidates new at each memo step, the selected memo choice, Stage 3's search outcome, cost per plan node and accepted accuracy per query, and with --enumerated the memo choices with their DAGs. --out - writes to standard output. node_label prints a time range as 60s. Co-Authored-By: Claude Opus 5.5 --- .../devtools/src/bin/print_planning_stages.rs | 55 -- crates/devtools/src/bin/stage_pipeline.rs | 36 +- crates/devtools/src/candidate_labels.rs | 19 +- crates/devtools/src/planning_stages_text.rs | 685 +++++++++++++----- crates/devtools/tests/stage_pipeline.rs | 50 ++ 5 files changed, 604 insertions(+), 241 deletions(-) delete mode 100644 crates/devtools/src/bin/print_planning_stages.rs diff --git a/crates/devtools/src/bin/print_planning_stages.rs b/crates/devtools/src/bin/print_planning_stages.rs deleted file mode 100644 index fd77bceab..000000000 --- a/crates/devtools/src/bin/print_planning_stages.rs +++ /dev/null @@ -1,55 +0,0 @@ -// cargo run -p asap-devtools --bin print_planning_stages -// cargo run -p asap-devtools --bin print_planning_stages -- \ -// 'sum by (job) (rate(http_requests_total[1m]))' \ -// 'sum by (job) (sum_over_time(http_requests_total[5m]))' -// -// Prints every planner stage's output as text -// (`asap_devtools::planning_stages_text`): the frontends' roots, the memo -// groups, Pass 1's `PerComputationLogicalMemoCandidates`, the memo after -// Pass 2's local window composition, Pass 2's `LogicalASAPMemoCandidates`, -// Stage 2's `PhysicalASAPMemoCandidates` and Stage 3's `Selection`. -// -// Without arguments it plans #509 Example 1. Each argument is a PromQL -// query that replaces Example 1's queries, keeping its data facts -// (`http_requests_total`, a counter) and its panels' requirements: exact, -// every 10 s. - -use asap_devtools::planner_layering_examples::example1; -use asap_devtools::planning_stages_text::{plan_promql_with_stage_outputs, planning_stages_text}; -use asap_types::workload::{Query, RepeatingEntry}; - -fn main() { - let queries: Vec = std::env::args().skip(1).collect(); - // Example 1's workload; its queries replaced when given. - let mut workload = example1(); - if !queries.is_empty() { - let repeating = workload - .query_workload - .repeating_queries - .as_mut() - .expect("Example 1 has repeating queries"); - // Example 1's first panel: exact, every 10 s. - let panel = repeating[0].clone(); - *repeating = queries - .iter() - .map(|query| RepeatingEntry { - query: Query(query.clone()), - ..panel.clone() - }) - .collect(); - } - println!("# Workload"); - for (index, entry) in workload.query_workload.entries().enumerate() { - println!( - " q{index}: {} accuracy={:?}", - entry.query.0, entry.requirements.accuracy - ); - } - match plan_promql_with_stage_outputs(&workload) { - Ok(stages) => print!("{}", planning_stages_text(&stages)), - Err(error) => { - eprintln!("print_planning_stages: {error}"); - std::process::exit(1); - } - } -} diff --git a/crates/devtools/src/bin/stage_pipeline.rs b/crates/devtools/src/bin/stage_pipeline.rs index 17865b922..65f4242ec 100644 --- a/crates/devtools/src/bin/stage_pipeline.rs +++ b/crates/devtools/src/bin/stage_pipeline.rs @@ -23,6 +23,12 @@ // // Writes an `asap-stage-pipeline/v1` document (tools/dag-viewer) with the // four planner stages as the facade runs them (`asap_planner::run_planning`). +// `--text` writes the same run as text instead, for reading and debugging +// (`asap_devtools::planning_stages_text`): every stage boundary, including +// Stage 2's whole `PhysicalASAPMemoCandidates`, the selected plan's cost per +// node and each query's accepted accuracy; `--text --enumerated` adds the +// first `--max-enumerated` memo choices with their DAGs. `--out -` writes +// to standard output. // The default document keeps the selected plan; `--exhaustive` adds the // reference enumeration: the selected round's memo choices, each implemented // by Stage 2 with every materialization assignment, all priced by the @@ -273,6 +279,7 @@ use asap_devtools::planning_stages::{ lower_promql_roots, lower_sql_roots, run_search_over, sql_catalog, PlanningStagesError, PlanningStagesRun, }; +use asap_devtools::planning_stages_text::planning_stages_text; use asap_devtools::SqlCatalog; use asap_frontend_promql::{counter_function_over_gauge_warnings, promql_metric_names}; use asap_logical_asap_optimization::DeploymentCapabilities; @@ -306,7 +313,8 @@ 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]] \ [--cost-inputs ] [--enforce-accuracy-guarantees] [--deployment-capabilities ] [--exhaustive [--max-combinations ]] [--milp-time-limit-s ] [--max-displayed ] \ -[--no-memo] [--max-memo-candidates ] [--memo-enumeration-limit ] [--max-enumerated ] [--price-enumerated ] --out "; +[--no-memo] [--max-memo-candidates ] [--memo-enumeration-limit ] [--max-enumerated ] [--price-enumerated ] \ +[--text [--enumerated]] --out "; /// Default evaluator-assignment budget for an exhaustive archive run. const MAX_COMBINATIONS: usize = 10_000; @@ -398,6 +406,8 @@ fn run(args: Vec) -> Result<(), Failure> { let mut milp_time_limit_ms: Option = None; let mut max_combinations_given = false; let mut write_memo = true; + let mut text = false; + let mut text_enumerated = false; let mut memo_caps = MemoCaps { candidates_per_group: 64, enumeration_limit: 1000, @@ -466,6 +476,8 @@ fn run(args: Vec) -> Result<(), Failure> { "--enforce-accuracy-guarantees" => enforce_accuracy_guarantees = true, "--exhaustive" => exhaustive = true, "--no-memo" => write_memo = false, + "--text" => text = true, + "--enumerated" => text_enumerated = true, "--max-memo-candidates" => { memo_caps.candidates_per_group = value()?.parse().map_err(|e| format!("{e}"))? } @@ -503,6 +515,9 @@ fn run(args: Vec) -> Result<(), Failure> { .into(), ); } + if text_enumerated && !text { + return Err("--enumerated lists memo choices in the --text output: it needs --text".into()); + } if max_combinations_given && !exhaustive { return Err("--max-combinations limits the exhaustive search: it needs --exhaustive (bounds evaluated assignments)".into()); } @@ -757,6 +772,14 @@ fn run(args: Vec) -> Result<(), Failure> { ) }), ); + if text { + let run = result.map_err(|e| e.to_string())?; + let mut text = planning_stages_text(&run, text_enumerated.then_some(memo_caps.enumerated)); + for warning in &warnings { + text.push_str(&format!("warning: {warning}\n")); + } + return write_output(&out, &text); + } let memo = result .as_ref() .ok() @@ -802,7 +825,16 @@ fn run(args: Vec) -> Result<(), Failure> { document["deployment_capabilities"] = serde_json::to_value(&deployment_capabilities).map_err(|e| e.to_string())?; 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}").into()) + write_output(&out, &text) +} + +/// Writes `text` to the file `out`, or to standard output when `out` is `-`. +fn write_output(out: &str, text: &str) -> Result<(), Failure> { + if out == "-" { + print!("{text}"); + return Ok(()); + } + std::fs::write(out, text).map_err(|e| format!("{out}: {e}").into()) } /// One warning per PromQL query of `described` (the document's queries) diff --git a/crates/devtools/src/candidate_labels.rs b/crates/devtools/src/candidate_labels.rs index 766040fe5..61e44f53e 100644 --- a/crates/devtools/src/candidate_labels.rs +++ b/crates/devtools/src/candidate_labels.rs @@ -356,8 +356,10 @@ fn find_details(value: &Value, details: &mut Vec, depth: usize) { details.push(format!("{kind}({})", params.join(", "))); } } - // A window's range, and a pane chain's layout. - "range" | "window" | "pane_width_ms" | "pane_count" => { + // A time range's length, e.g. `60s`. + "range" => details.push(duration_text(field)), + // A window, and a pane chain's layout. + "window" | "pane_width_ms" | "pane_count" => { details.push(format!("{key}={field}")) } _ => find_details(field, details, depth + 1), @@ -373,6 +375,19 @@ fn find_details(value: &Value, details: &mut Vec, depth: usize) { } } +/// A serialized `Duration` (`{"secs": 60, "nanos": 0}`) as `60s`, or +/// `1500ms` when it is not whole seconds; any other value as its JSON. +fn duration_text(value: &Value) -> String { + match ( + value.get("secs").and_then(Value::as_u64), + value.get("nanos").and_then(Value::as_u64), + ) { + (Some(secs), Some(0)) => format!("{secs}s"), + (Some(secs), Some(nanos)) => format!("{}ms", secs * 1000 + nanos / 1_000_000), + _ => value.to_string(), + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/devtools/src/planning_stages_text.rs b/crates/devtools/src/planning_stages_text.rs index 2a35856a5..c4cf60ce2 100644 --- a/crates/devtools/src/planning_stages_text.rs +++ b/crates/devtools/src/planning_stages_text.rs @@ -1,57 +1,43 @@ -//! Every planner stage's output as plain text, for reading one workload's -//! planning end to end (`print_planning_stages`): the frontends' roots -//! (Stage 0), the memo groups (Stage 0 → Stage 1), Pass 1's -//! `PerComputationLogicalMemoCandidates`, the memo after Pass 2's local -//! window composition, Pass 2's `LogicalASAPMemoCandidates` (Stage 2's -//! input), Stage 2's `PhysicalASAPMemoCandidates` (Stage 3's input) and -//! Stage 3's `Selection`. Planning never reads it. +//! Every planner stage's output as plain text: `stage_pipeline --text`. +//! It prints the frontends' roots (Stage 0), the memo after each step (its +//! groups, then Pass 1's `PerComputationLogicalMemoCandidates`, the memo +//! after Pass 2's local window composition, and Pass 2's +//! `LogicalASAPMemoCandidates`, Stage 2's input), the selected memo choice, +//! Stage 2's `PhysicalASAPMemoCandidates` (Stage 3's input), Stage 3's +//! `Selection` with each node's cost and each query's accepted accuracy, +//! and optionally the enumerated memo choices. Planning never reads it. //! -//! Notation: `G` is memo group `n`; `G[]` its candidate `i`; -//! `O` Stage 2's operator node `n`; `C` physical candidate `n`; a -//! candidate's chain lists its own nodes root first, down to its input -//! groups. +//! Notation: `G` is memo group `n` and `G[]` its candidate `i`; +//! `O` is Stage 2's operator node `n` and `C` physical candidate `n`; +//! `<- ...` lists what a line reads; `new` marks a candidate the step +//! added. A candidate's chain lists its own nodes root first, down to its +//! input groups: +//! +//! ```text +//! G1: TimeRange 60s <- G0 +//! G2: Aggregate rate() <- G1 +//! [0] Pass1 FinalizeExactAccumulator ← SummaryAgg inputs=[G1] +//! [1] Panes new FinalizeExactAccumulator ← SummaryMerge ← PaneChain pane_count=6 pane_width_ms=10000 inputs=[G0] +//! ``` use std::collections::HashSet; use std::fmt::Write; use std::rc::Rc; -use asap_planner::{PlanSearchSettings, PlanningModels}; +use asap_logical_asap_optimization::{build_logical_asap_dags_from_chosen_candidates, enumerate}; use asap_types::ir::flat::flatten; use asap_types::ir::{OperatorNode, QueryRoot}; use asap_types::stages::{MemoCandidates, MemoGroupId, PhysicalASAPMemoCandidates}; -use asap_types::workload::PlanningWorkload; use crate::candidate_labels::node_label; -use crate::planning_stages::{ - lower_promql_roots, run_search_over, PlanningStagesError, PlanningStagesRun, -}; - -/// Plans the PromQL `workload` as the facade does, keeping every stage's -/// memo (`PlanSearchSettings::inspect`), without the reference enumeration. -pub fn plan_promql_with_stage_outputs( - workload: &PlanningWorkload, -) -> Result { - let roots = lower_promql_roots(workload)?; - let entries: Vec<_> = workload.query_workload.entries().collect(); - let data = workload.data_workload.clone().unwrap_or_default(); - run_search_over( - roots, - &entries, - &data, - PlanningModels::builtin(), - PlanSearchSettings { - inspect: true, - ..PlanSearchSettings::default() - }, - None, - ) - .map_err(|error: PlanningStagesError| error.to_string()) -} +use crate::planning_stages::PlanningStagesRun; /// Every stage's output of `stages` (planned with -/// `PlanSearchSettings::inspect`) as text, one section per stage boundary. -/// A stage that did not run is reported as missing. -pub fn planning_stages_text(stages: &PlanningStagesRun) -> String { +/// `PlanSearchSettings::inspect`) as text, one section per stage boundary; +/// with `enumerated`, also the first that many memo choices over Stage 2's +/// input memo, each with its composed DAG. A stage that did not run is +/// reported as missing. +pub fn planning_stages_text(stages: &PlanningStagesRun, enumerated: Option) -> String { let mut out = String::new(); let run = &stages.run; @@ -59,17 +45,7 @@ pub fn planning_stages_text(stages: &PlanningStagesRun) -> String { &mut out, "Stage 0 output: the frontends' roots (Vec)", ); - let (flat, _) = flatten(&stages.stage0_query_language_frontends); - for (id, node) in flat.nodes.iter().enumerate() { - let children: Vec = node - .operator - .children() - .into_iter() - .map(|child| format!("n{child}")) - .collect(); - let _ = writeln!(out, " n{id}: {} <- {children:?}", node_label(node)); - } - let _ = writeln!(out, " roots: {:?}", flat.roots); + write_dag(&mut out, &stages.stage0_query_language_frontends); let Some(pass1) = &run.per_computation_candidates else { let _ = writeln!(out, "\n(no Pass 1 output)"); @@ -79,42 +55,61 @@ pub fn planning_stages_text(stages: &PlanningStagesRun) -> String { &mut out, "Stage 0 → Stage 1: DagOfLogicalMemoGroups (the roots hash-consed)", ); - write_memo(&mut out, &pass1.memo, false); + write_memo(&mut out, &pass1.memo, None, false); section( &mut out, "Stage 1 Pass 1 output: PerComputationLogicalMemoCandidates", ); - write_memo(&mut out, &pass1.memo, true); - let _ = writeln!(out, " computation_targets:"); - for target in &pass1.computation_targets { - let _ = writeln!( - out, - " G{} measure {}: ways_to_compute={:?} fused={:?}", - target.aggregate_group.0, - target.measure_index, - target.ways_to_compute, - target + write_memo(&mut out, &pass1.memo, None, true); + let _ = writeln!(out, "computation_targets:"); + let rows = pass1 + .computation_targets + .iter() + .map(|target| { + let fused = target .fused_computation .as_ref() - .map(|fused| &fused.ways_to_compute) - ); - } - let _ = writeln!(out, " pruned_candidates:"); - for pruned in &pass1.pruned_candidates { - let _ = writeln!( - out, - " G{}: {:?} reason={:?}", - pruned.group.0, pruned.candidate, pruned.reason - ); - } + .map_or(String::new(), |fused| { + format!("fused: {:?}", fused.ways_to_compute) + }); + vec![ + format!( + "G{} measure {}:", + target.aggregate_group.0, target.measure_index + ), + format!("{:?}", target.ways_to_compute), + fused, + ] + }) + .collect(); + write_rows(&mut out, " ", rows); + let _ = writeln!( + out, + "pruned_candidates ({}):", + pass1.pruned_candidates.len() + ); + let rows = pass1 + .pruned_candidates + .iter() + .map(|pruned| { + vec![ + format!("G{}:", pruned.group.0), + format!("{:?}", pruned.candidate), + format!("{:?}", pruned.reason), + ] + }) + .collect(); + write_rows(&mut out, " ", rows); + let mut previous = &pass1.memo; if let Some(steps) = &run.memo_steps { section( &mut out, "Stage 1 Pass 2, after local window composition (pane candidates)", ); - write_memo(&mut out, &steps.pass2, true); + write_memo(&mut out, &steps.pass2, Some(previous), true); + previous = &steps.pass2; } match &run.logical_asap_candidates { Some(logical) => { @@ -122,12 +117,23 @@ pub fn planning_stages_text(stages: &PlanningStagesRun) -> String { &mut out, "Stage 1 Pass 2 output = Stage 2 input: LogicalASAPMemoCandidates", ); - write_memo(&mut out, &logical.memo, true); + write_memo(&mut out, &logical.memo, Some(previous), true); } None => { let _ = writeln!(out, "\n(no Pass 2 output)"); } } + if let Some(steps) = &run.memo_steps { + match &steps.selected_chosen_candidates { + Ok(choice) => { + let _ = writeln!(out, "selected choice: {}", choice_text(choice)); + } + Err(reason) => { + let _ = writeln!(out, "selected choice unavailable: {reason}"); + } + } + } + match &run.physical_asap_candidates { Some(physical) => { section( @@ -142,29 +148,15 @@ pub fn planning_stages_text(stages: &PlanningStagesRun) -> String { } section(&mut out, "Stage 3 output: Selection"); - match &run.selection { - Ok(selection) => { - let dag = &selection.selected_dag; - for (node, candidate) in dag.candidate_of_node.iter().enumerate() { - let inputs: Vec = dag - .edges - .iter() - .filter(|edge| edge.consumer.0 == node) - .map(|edge| format!("input {} <- node{}", edge.input, edge.producer.0)) - .collect(); - let _ = writeln!(out, " node{node}: runs C{} {inputs:?}", candidate.0); - } - let outputs: Vec = dag.node_of_output_reference.iter().map(|n| n.0).collect(); - let _ = writeln!(out, " node_of_output_reference: {outputs:?}"); - let _ = writeln!( - out, - " cost_per_second: {:?}", - selection.cost.cost_per_second - ); - } - Err(error) => { - let _ = writeln!(out, " error: {error}"); - } + write_selection(&mut out, stages); + + if let (Some(limit), Some(steps)) = (enumerated, &run.memo_steps) { + write_enumeration( + &mut out, + &steps.concrete_states, + &steps.selected_chosen_candidates, + limit, + ); } out } @@ -173,39 +165,165 @@ fn section(out: &mut String, title: &str) { let _ = writeln!(out, "\n==================== {title} ===================="); } -fn group_names(groups: &[MemoGroupId]) -> Vec { - groups.iter().map(|group| format!("G{}", group.0)).collect() +/// One line per node of the DAG of `roots` (children before parents), then +/// the query roots. +fn write_dag(out: &mut String, roots: &[QueryRoot]) { + let (flat, _) = flatten(roots); + let rows = flat + .nodes + .iter() + .enumerate() + .map(|(id, node)| { + let children: Vec = node + .operator + .children() + .into_iter() + .map(|child| format!("n{child}")) + .collect(); + vec![format!("n{id}:"), node_label(node), reads(&children)] + }) + .collect(); + write_rows(out, "", rows); + let roots: Vec = flat + .roots + .iter() + .map(|root| match root { + QueryRoot::Operator(id) => format!("n{id}"), + scalar => format!("{scalar:?}"), + }) + .collect(); + let _ = writeln!(out, "query_roots: [{}]", roots.join(", ")); +} + +/// `rows` with each column but the last padded to its widest cell, two +/// spaces apart, each line prefixed by `indent`; trailing empty cells are +/// dropped. +fn write_rows(out: &mut String, indent: &str, rows: Vec>) { + let columns = rows.iter().map(Vec::len).max().unwrap_or(0); + let widths: Vec = (0..columns) + .map(|column| { + rows.iter() + .filter_map(|row| row.get(column)) + .map(|cell| cell.chars().count()) + .max() + .unwrap_or(0) + }) + .collect(); + for row in rows { + let last = row.iter().rposition(|cell| !cell.is_empty()).unwrap_or(0); + let mut line = indent.to_owned(); + for (column, cell) in row.iter().enumerate().take(last + 1) { + line.push_str(cell); + if column < last { + let padding = widths[column] - cell.chars().count() + 2; + line.push_str(&" ".repeat(padding)); + } + } + let _ = writeln!(out, "{line}"); + } +} + +/// `<- a, b`, or nothing when there is nothing read. +fn reads(inputs: &[String]) -> String { + if inputs.is_empty() { + String::new() + } else { + format!("<- {}", inputs.join(", ")) + } } -/// `memo`'s groups, and with `with_candidates` each group's candidates. -fn write_memo(out: &mut String, memo: &MemoCandidates, with_candidates: bool) { - for (id, group) in memo.groups.iter().enumerate() { - let _ = writeln!( - out, - " G{id}: {} children={:?} parents={:?} queries={:?}", - root_label(&group.logical_node), - group_names(&group.child_groups), - group_names(group.parent_groups()), - group.reached_by_queries() - ); +/// `G0, G1`. +fn group_list(groups: &[MemoGroupId]) -> String { + groups + .iter() + .map(|group| format!("G{}", group.0)) + .collect::>() + .join(", ") +} + +/// `G2:c0 G3:c0`: each chosen group's candidate. +fn choice_text(choice: &asap_logical_asap_optimization::ChosenCandidatePerMemoGroup) -> String { + choice + .iter() + .map(|(group, candidate)| format!("G{}:c{candidate}", group.0)) + .collect::>() + .join(" ") +} + +/// `memo`'s groups, one line each with the child groups it reads, and with +/// `with_candidates` each group's candidates below it: index, origin, +/// `new` when `previous` (the memo at the step before) has no candidate of +/// that group with the same node and inputs, chain and input groups (fused +/// groups, and accuracy budget parts when more than one). +fn write_memo( + out: &mut String, + memo: &MemoCandidates, + previous: Option<&MemoCandidates>, + with_candidates: bool, +) { + // The group lines share one set of column widths, with each group's + // candidates printed under its line. + let group_rows: Vec> = memo + .groups + .iter() + .enumerate() + .map(|(id, group)| { + let children: Vec = group + .child_groups + .iter() + .map(|child| format!("G{}", child.0)) + .collect(); + vec![ + format!("G{id}:"), + root_label(&group.logical_node), + reads(&children), + ] + }) + .collect(); + let mut group_lines = String::new(); + write_rows(&mut group_lines, "", group_rows); + for (id, line) in group_lines.lines().enumerate() { + let _ = writeln!(out, "{line}"); if !with_candidates { continue; } - for (index, candidate) in memo.candidates[id].iter().enumerate() { - let fuses = if candidate.fused_groups.is_empty() { - String::new() - } else { - format!(" fuses={:?}", group_names(&candidate.fused_groups)) - }; - let _ = writeln!( - out, - " [{index}] {:<12} {} inputs={:?}{fuses} accuracy_budget_parts={}", - format!("{:?}", candidate.origin), - chain_label(memo, &candidate.root_node, &candidate.input_groups), - group_names(&candidate.input_groups), - candidate.accuracy_budget_parts - ); - } + let earlier = previous + .and_then(|memo| memo.candidates.get(id)) + .map_or(&[][..], Vec::as_slice); + let rows = memo.candidates[id] + .iter() + .enumerate() + .map(|(index, candidate)| { + let added = previous.is_some() + && !earlier.iter().any(|before| { + Rc::ptr_eq(&before.root_node, &candidate.root_node) + && before.input_groups == candidate.input_groups + }); + let mut inputs = format!("inputs=[{}]", group_list(&candidate.input_groups)); + if !candidate.fused_groups.is_empty() { + let _ = write!(inputs, " fuses=[{}]", group_list(&candidate.fused_groups)); + } + if candidate.accuracy_budget_parts != 1 { + let _ = write!( + inputs, + " accuracy_budget_parts={}", + candidate.accuracy_budget_parts + ); + } + vec![ + format!("[{index}]"), + format!("{:?}", candidate.origin), + if added { + "new".to_owned() + } else { + String::new() + }, + chain_label(memo, &candidate.root_node, &candidate.input_groups), + inputs, + ] + }) + .collect(); + write_rows(out, " ", rows); } let roots: Vec = memo .query_roots @@ -215,58 +333,91 @@ fn write_memo(out: &mut String, memo: &MemoCandidates, with_candidates: bool) { scalar => format!("{scalar:?}"), }) .collect(); - let _ = writeln!(out, " query_roots: {roots:?}"); + let _ = writeln!(out, "query_roots: [{}]", roots.join(", ")); } fn write_physical(out: &mut String, physical: &PhysicalASAPMemoCandidates) { let nodes = &physical.operator_nodes_of_candidates_for_memo_groups; let _ = writeln!( out, - " operator_nodes_of_candidates_for_memo_groups ({}):", + "operator_nodes_of_candidates_for_memo_groups ({}):", nodes.len() ); - for (id, node) in nodes.iter().enumerate() { - let implements: Vec = node - .candidates_for_memo_group - .iter() - .map(|candidate| format!("G{}[{}]", candidate.group.0, candidate.index)) - .collect(); - let _ = writeln!( - out, - " O{id}: {} implements={implements:?} is_implementation_root={} accuracy_budget_parts={}", - root_label(&node.physical_node), - node.is_implementation_root, - node.accuracy_budget_parts - ); - } - let _ = writeln!(out, " candidates ({}):", physical.candidates.len()); - for (id, candidate) in physical.candidates.iter().enumerate() { - let producers: Vec = candidate - .producer_candidates_per_input - .iter() - .map(|producers| { - let ids: Vec = producers.iter().map(|p| format!("C{}", p.0)).collect(); - format!("{{{}}}", ids.join(",")) - }) - .collect(); - let _ = writeln!( - out, - " C{id}: O{} {:?} {:?} producers per input: [{}]{}", - candidate.operator_node.0, - candidate.timing, - candidate.materialization, - producers.join(" "), - if candidate.may_produce_query_output { - " (may produce a query output)" + let rows = nodes + .iter() + .enumerate() + .map(|(id, node)| { + let implements: Vec = node + .candidates_for_memo_group + .iter() + .map(|candidate| format!("G{}[{}]", candidate.group.0, candidate.index)) + .collect(); + let mut role = if node.is_implementation_root { + "implementation root".to_owned() } else { - "" + "private".to_owned() + }; + if node.accuracy_budget_parts != 1 { + let _ = write!( + role, + " accuracy_budget_parts={}", + node.accuracy_budget_parts + ); } - ); - } - let _ = writeln!(out, " query_roots: {:?}", physical.query_roots); + vec![ + format!("O{id}:"), + root_label(&node.physical_node), + format!("implements=[{}]", implements.join(", ")), + role, + ] + }) + .collect(); + write_rows(out, " ", rows); + + let _ = writeln!(out, "candidates ({}):", physical.candidates.len()); + let rows = physical + .candidates + .iter() + .enumerate() + .map(|(id, candidate)| { + // One `{...}` per input: the candidates that may produce it. + let producers: Vec = candidate + .producer_candidates_per_input + .iter() + .map(|producers| { + let ids: Vec = producers.iter().map(|p| format!("C{}", p.0)).collect(); + format!("{{{}}}", ids.join(", ")) + }) + .collect(); + vec![ + format!("C{id}:"), + format!("O{}", candidate.operator_node.0), + format!("{:?}", candidate.timing), + format!("{:?}", candidate.materialization), + // It may produce a query's output. + if candidate.may_produce_query_output { + "output".to_owned() + } else { + String::new() + }, + reads(&producers), + ] + }) + .collect(); + write_rows(out, " ", rows); + + let output_groups: Vec = physical + .query_roots + .iter() + .map(|root| match root { + QueryRoot::Operator(group) => format!("G{}", group.0), + scalar => format!("{scalar:?}"), + }) + .collect(); + let _ = writeln!(out, "query_roots: [{}]", output_groups.join(", ")); let _ = writeln!( out, - " reader_constraints ({}):", + "reader_constraints ({}):", physical.reader_constraints.len() ); for constraint in &physical.reader_constraints { @@ -274,16 +425,157 @@ fn write_physical(out: &mut String, physical: &PhysicalASAPMemoCandidates) { } let _ = writeln!( out, - " pruned_candidates ({}):", + "pruned_candidates ({}):", physical.pruned_candidates.len() ); - for pruned in &physical.pruned_candidates { - let key = &pruned.candidate; - let _ = writeln!( - out, - " G{} O{} {:?} {:?} reason={:?}", - pruned.group.0, key.operator_node.0, key.timing, key.materialization, pruned.reason - ); + let rows = physical + .pruned_candidates + .iter() + .map(|pruned| { + let key = &pruned.candidate; + vec![ + format!("G{}", pruned.group.0), + format!("O{}", key.operator_node.0), + format!("{:?}", key.timing), + format!("{:?}", key.materialization), + format!("{:?}", pruned.reason), + ] + }) + .collect(); + write_rows(out, " ", rows); +} + +/// Stage 3's selection: the search's outcome, the selected DAG (each node's +/// candidate, operator and producer nodes), the plan's cost with each plan +/// node's charge, and each query's accepted accuracy; or why none was +/// selected. +fn write_selection(out: &mut String, stages: &PlanningStagesRun) { + let run = &stages.run; + let selection = match &run.selection { + Ok(selection) => selection, + Err(error) => { + let _ = writeln!(out, "error: {error}"); + return; + } + }; + let report = &selection.search_report; + let _ = writeln!( + out, + "search: {:?}, {} MILP variables, {} constraints, {} ms solving", + report.status, report.milp_variables, report.milp_constraints, report.solver_ms + ); + let dag = &selection.selected_dag; + let rows = dag + .candidate_of_node + .iter() + .enumerate() + .map(|(node, candidate)| { + let mut inputs: Vec<_> = dag + .edges + .iter() + .filter(|edge| edge.consumer.0 == node) + .collect(); + inputs.sort_by_key(|edge| edge.input); + let producers: Vec = inputs + .iter() + .map(|edge| format!("node{}", edge.producer.0)) + .collect(); + // The operator the candidate runs. + let operator = + run.physical_asap_candidates + .as_ref() + .map_or(String::new(), |physical| { + let node = physical.candidates[candidate.0].operator_node; + root_label( + &physical.operator_nodes_of_candidates_for_memo_groups[node.0] + .physical_node, + ) + }); + vec![ + format!("node{node}:"), + format!("C{}", candidate.0), + operator, + reads(&producers), + ] + }) + .collect(); + write_rows(out, "", rows); + let outputs: Vec = dag + .node_of_output_reference + .iter() + .map(|node| format!("node{}", node.0)) + .collect(); + let _ = writeln!(out, "node_of_output_reference: [{}]", outputs.join(", ")); + let cost = &selection.cost; + let _ = writeln!( + out, + "cost_per_second: {}", + cost.cost_per_second + .map_or("none".to_owned(), |cost| format!("{cost:.6}")) + ); + // Keyed by the selected plan's node ids, as Stage 2 built it. + let rows = cost + .per_node + .iter() + .map(|(node, charge)| { + vec![ + format!("#{node}"), + format!("{:?}", charge.operator), + format!("{:?}", charge.timing), + charge + .cost_per_second + .map_or("none".to_owned(), |cost| format!("{cost:.6}")), + charge.detail.clone(), + ] + }) + .collect(); + write_rows(out, " ", rows); + let _ = writeln!(out, "accepted accuracy:"); + for (query, accepted) in selection.accepted_accuracy_per_query.iter().enumerate() { + let _ = writeln!(out, " q{query}: {accepted}"); + } +} + +/// The first `limit` memo choices over `memo` (Stage 2's input), each with +/// the selected one marked and its composed DAG. +fn write_enumeration( + out: &mut String, + memo: &MemoCandidates, + selected: &Result, + limit: usize, +) { + let enumeration = enumerate(memo, limit); + section( + out, + &format!( + "Memo choices over Stage 2's input: {} listed{}", + enumeration.all_chosen_candidates.len(), + if enumeration.complete { + " (all)" + } else { + ", more not listed" + } + ), + ); + for (index, choice) in enumeration.all_chosen_candidates.iter().enumerate() { + let mark = if selected.as_ref().ok() == Some(choice) { + " <== selected" + } else { + "" + }; + let _ = writeln!(out, "\nE{} {}{mark}", index + 1, choice_text(choice)); + match build_logical_asap_dags_from_chosen_candidates(memo, choice) { + Ok(roots) => { + let roots: Vec = roots + .into_iter() + .map(|(_, node)| QueryRoot::Operator(node)) + .collect(); + write_dag(out, &roots); + } + Err(error) => { + let _ = writeln!(out, "error: {error}"); + } + } } } @@ -330,19 +622,35 @@ fn chain_label( mod tests { use super::*; use crate::planner_layering_examples::example1; + use crate::planning_stages::{lower_promql_roots, run_search_over}; + use asap_planner::{PlanSearchSettings, PlanningModels}; use asap_types::workload::Query; /// One repeating `sum(rate(...[1m]))` query: every section is printed, - /// Pass 2 adds pane candidates with their layout, and Stage 2 offers the - /// pane chain at ingestion time. + /// Pass 2 adds pane candidates with their layout, Stage 2 offers the + /// pane chain at ingestion time, and the enumeration lists the choices. #[test] fn prints_every_stage_of_a_repeating_rate_query() { let mut workload = example1(); let queries = workload.query_workload.repeating_queries.as_mut().unwrap(); queries.truncate(1); queries[0].query = Query("sum by (job) (rate(http_requests_total[1m]))".into()); + let roots = lower_promql_roots(&workload).unwrap(); + let entries: Vec<_> = workload.query_workload.entries().collect(); + let stages = run_search_over( + roots, + &entries, + &workload.data_workload.clone().unwrap(), + PlanningModels::builtin(), + PlanSearchSettings { + inspect: true, + ..PlanSearchSettings::default() + }, + None, + ) + .unwrap(); - let text = planning_stages_text(&plan_promql_with_stage_outputs(&workload).unwrap()); + let text = planning_stages_text(&stages, Some(2)); for heading in [ "Stage 0 output", @@ -352,13 +660,26 @@ mod tests { "Stage 2 input: LogicalASAPMemoCandidates", "Stage 3 input: PhysicalASAPMemoCandidates", "Stage 3 output: Selection", + "Memo choices over Stage 2's input: 2 listed, more not listed", ] { assert!(text.contains(heading), "missing {heading}:\n{text}"); } + // A group line names its operator and the groups it reads. + assert!(text.contains("G1: TimeRange 60s <- G0\n"), "{text}"); + // Pass 2 adds a pane candidate reading the scan, its layout in its chain. assert!(text.contains( - "Panes FinalizeExactAccumulator ← SummaryMerge ← PaneChain pane_count=6 pane_width_ms=10000" - )); - assert!(text.contains("IngestionTime MatAtIngestion(Memory)")); - assert!(text.contains("cost_per_second: Some(")); + " [1] Panes new FinalizeExactAccumulator ← SummaryMerge ← PaneChain pane_count=6 pane_width_ms=10000 inputs=[G0]\n" + ), "{text}"); + // Stage 2 offers the pane chain at ingestion time, read from the + // ingestion-time scan. + assert!( + text.contains("IngestionTime MatAtIngestion(Memory) <- {C1}\n"), + "{text}" + ); + assert!( + text.contains("selected choice: G0:c0 G1:c0 G2:c0 G3:c0\n"), + "{text}" + ); + assert!(text.contains("q0: "), "{text}"); } } diff --git a/crates/devtools/tests/stage_pipeline.rs b/crates/devtools/tests/stage_pipeline.rs index 057374a2c..79d2c3adb 100644 --- a/crates/devtools/tests/stage_pipeline.rs +++ b/crates/devtools/tests/stage_pipeline.rs @@ -2510,3 +2510,53 @@ fn memo_section_lists_each_step_the_selected_choice_and_the_enumeration() { std::fs::remove_file(out).unwrap(); assert!(document.get("memo").is_none()); } + +/// `--text --out -` prints the same planning run as text on standard +/// output: every stage boundary, Stage 2's physical candidates, the selected +/// plan's cost per node, and with `--enumerated` the memo choices. +#[test] +fn text_output_prints_every_stage_to_standard_output() { + let output = Command::new(env!("CARGO_BIN_EXE_stage_pipeline")) + .args([ + "--example", + "planner-layering-1", + "--text", + "--enumerated", + "--max-enumerated", + "1", + "--out", + "-", + ]) + .output() + .unwrap(); + assert!(output.status.success()); + let text = String::from_utf8(output.stdout).unwrap(); + for heading in [ + "Stage 0 output", + "Stage 1 Pass 1 output: PerComputationLogicalMemoCandidates", + "Stage 1 Pass 2 output = Stage 2 input: LogicalASAPMemoCandidates", + "Stage 2 output = Stage 3 input: PhysicalASAPMemoCandidates", + "Stage 3 output: Selection", + "Memo choices over Stage 2's input: 1 listed", + ] { + assert!(text.contains(heading), "missing {heading}:\n{text}"); + } + assert!(text.contains("\ncost_per_second: "), "{text}"); +} + +/// `--enumerated` only adds to the `--text` output. +#[test] +fn enumerated_needs_text() { + let status = Command::new(env!("CARGO_BIN_EXE_stage_pipeline")) + .args([ + "--example", + "planner-layering-1", + "--enumerated", + "--out", + "-", + ]) + .stderr(std::process::Stdio::null()) + .status() + .unwrap(); + assert_eq!(status.code(), Some(2)); +}