From 718e8259cc7bddc93a4558a9d0958a7e6a7edda1 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Wed, 21 Oct 2026 08:30:20 +0000 Subject: [PATCH 1/2] perf(planner): compile each stored output's build over the nodes below it plan_installable (and Stage 3's executor compile check, through check_compiles) compiled every stored output's build with the whole plan: the executor's compile runs its whole-plan checks (depth, restore, validate) and indexes every node and edge before lowering the build, so a plan of N nodes with S stored outputs cost O(S x N). A pane-expanded plan of a long window has thousands of panes, each a stored output. The first build is still compiled in the whole plan, which runs the executor's whole-plan checks once; every other build is compiled over the sub-DAG of its node and every node below it. check_compiles also looks up nodes and inputs through the index instead of scanning the node and edge lists for each one. --- crates/planner/src/installable.rs | 149 ++++++++++++++++++++++++++---- 1 file changed, 133 insertions(+), 16 deletions(-) diff --git a/crates/planner/src/installable.rs b/crates/planner/src/installable.rs index cba43961e..b84fadf9f 100644 --- a/crates/planner/src/installable.rs +++ b/crates/planner/src/installable.rs @@ -46,8 +46,8 @@ use asap_plan_selection::{ use asap_types::ir::operator::pane_chain::expand_pane_chains_in_workload; use asap_types::ir::physical_export::{ compile_physical_asap_workload_with_node_ids, PhysicalASAPDAG, PhysicalASAPDAGDocument, - PhysicalASAPDAGNode, PhysicalASAPDAGValidationError, PhysicalASAPNodeId, RawSampleWindow, - ReferencePlanError, + PhysicalASAPDAGEdge, PhysicalASAPDAGNode, PhysicalASAPDAGValidationError, PhysicalASAPNodeId, + RawSampleWindow, ReferencePlanError, }; use asap_types::ir::properties::summary_coverage::{SelectionBox, SummaryCoverage}; use asap_types::ir::properties::{ExecutionDataStateError, ExecutionTiming}; @@ -439,11 +439,11 @@ impl InstallablePlan { use asap_executor::physical_planner::{compile, InputContract}; use std::collections::BTreeMap; let dag = &self.dag.dag; + let indexed = IndexedDag::new(dag); let contract = |id: PhysicalASAPNodeId| -> Result<(u64, InputContract), String> { - let node = dag + let node = indexed .nodes - .iter() - .find(|node| node.id == id) + .get(&id) .ok_or_else(|| format!("node {id} is not in the plan"))?; Ok(( id as u64, @@ -452,7 +452,9 @@ impl InstallablePlan { }; let stored: HashSet = self.stored_outputs.iter().map(|o| o.node_id).collect(); - for output in &self.stored_outputs { + // The first build in the whole plan, which runs the executor's checks + // of the whole plan once; every other over the nodes below it. + for (position, output) in self.stored_outputs.iter().enumerate() { let inputs = output .inputs .raw_sample_windows @@ -461,7 +463,15 @@ impl InstallablePlan { .chain(output.inputs.stored_outputs.iter().copied()) .map(contract) .collect::, _>>()?; - compile(dag, inputs, &[output.node_id as u64]) + let sub_dag; + let below = match position { + 0 => dag, + _ => { + sub_dag = indexed.dag_below(output.node_id); + &sub_dag + } + }; + compile(below, inputs, &[output.node_id as u64]) .map_err(|error| format!("stored output {}: {error}", output.node_id))?; } // The query run reads the stored outputs it reaches from the roots @@ -477,12 +487,7 @@ impl InstallablePlan { if stored.contains(&id) { read.push(id); } else { - pending.extend( - dag.edges - .iter() - .filter(|edge| edge.consumer == id) - .map(|edge| edge.producer), - ); + pending.extend(indexed.inputs.get(&id).into_iter().flatten().copied()); } } let inputs = read @@ -625,7 +630,12 @@ fn queries_bound_to_roots( /// order, and two facts every raw-sample window check reads, computed in /// one pass each over the DAG (O(nodes + edges)). struct IndexedDag<'a> { + dag: &'a PhysicalASAPDAG, nodes: HashMap, + /// Each node's position in the DAG's node list. + position: HashMap, + /// Each consumer's input edges, in the DAG's edge order. + edges_into: HashMap>, inputs: HashMap>, /// Every node after its inputs, from the DAG's node order. children_first: Vec, @@ -694,8 +704,21 @@ impl<'a> IndexedDag<'a> { re_timed_from_above.extend(inputs.get(id).into_iter().flatten().copied()); } } + let position = dag + .nodes + .iter() + .enumerate() + .map(|(i, node)| (node.id, i)) + .collect(); + let mut edges_into: HashMap> = HashMap::new(); + for edge in &dag.edges { + edges_into.entry(edge.consumer).or_default().push(edge); + } IndexedDag { + dag, nodes, + position, + edges_into, inputs, children_first, reading_raw_samples, @@ -703,6 +726,32 @@ impl<'a> IndexedDag<'a> { } } + /// The sub-DAG of `root` and every node below it, with their edges, in + /// the DAG's node and edge order, rooted at `root`: all a build of + /// `root` reads, at the cost of its size rather than the plan's. + fn dag_below(&self, root: PhysicalASAPNodeId) -> PhysicalASAPDAG { + let mut below = HashSet::new(); + let mut pending = vec![root]; + while let Some(id) = pending.pop() { + if below.insert(id) { + pending.extend(self.inputs.get(&id).into_iter().flatten().copied()); + } + } + let mut ids: Vec = below.into_iter().collect(); + ids.sort_unstable_by_key(|id| self.position[id]); + let edges = ids + .iter() + .flat_map(|id| self.edges_into.get(id).into_iter().flatten()) + .map(|edge| (*edge).clone()) + .collect(); + PhysicalASAPDAG { + nodes: ids.iter().map(|id| self.nodes[id].clone()).collect(), + edges, + roots: vec![root], + scalar_roots: Vec::new(), + } + } + fn relational(&self, id: PhysicalASAPNodeId) -> Option<&'a NonASAPOp> { match &self.nodes.get(&id)?.payload { Operator::NonASAP(op) => Some(op), @@ -910,6 +959,10 @@ fn runtime_inputs_of( // The ranges of stored outputs' raw-sample windows, which are fed // through those outputs' builds. let mut stored_window_ranges = HashSet::new(); + // The first build is compiled in the whole plan, which runs the + // executor's checks of the whole plan once; every other over the nodes + // below it ([`executor_builds`]). + let mut whole_plan_checked = false; for &id in indexed .children_first .iter() @@ -924,7 +977,8 @@ fn runtime_inputs_of( } })?; coverage_by_node.insert(id, coverage.clone()); - executor_builds(dag, id, &inputs, &indexed)?; + executor_builds(id, &inputs, &indexed, !whole_plan_checked)?; + whole_plan_checked = true; let reach_ms = reach_ms_of(&inputs, &reach_ms_by_node); let span_ms = match inputs.stored_outputs.is_empty() { true => 0, @@ -1067,12 +1121,24 @@ fn with_replaced( /// `inputs` bound as sources, as a runtime runs it; the executor's reason /// when it does not. fn executor_builds( - dag: &PhysicalASAPDAG, node: PhysicalASAPNodeId, inputs: &StoredOutputInputs, indexed: &IndexedDag<'_>, + in_whole_plan: bool, ) -> Result<(), InstallablePlanError> { use asap_executor::physical_planner::{compile, InputContract}; + // The build reads only the nodes below it: compile them, or the whole + // plan once, for the executor's checks of the whole plan. + let sub_dag; + let below = match in_whole_plan { + true => indexed.dag, + false => { + sub_dag = indexed.dag_below(node); + &sub_dag + } + }; + #[cfg(test)] + tests::COMPILED_NODES.with(|count| count.set(count.get() + below.nodes.len())); let bound = inputs .raw_sample_windows .iter() @@ -1083,7 +1149,7 @@ fn executor_builds( (id as u64, InputContract::bounded(schema)) }) .collect(); - compile(dag, bound, &[node as u64]) + compile(below, bound, &[node as u64]) .map(|_| ()) .map_err(|error| InstallablePlanError::StoredOutputNotExecutable { node_id: node, @@ -1230,6 +1296,57 @@ mod tests { use super::*; use asap_types::ir::schema::{SketchAlgorithm, StatisticKind, SupportedSummary}; + thread_local! { + /// The plan nodes handed to the executor's compiler by + /// [`runtime_inputs_of`], summed over its compiles. + pub(super) static COMPILED_NODES: std::cell::Cell = const { std::cell::Cell::new(0) }; + } + + /// A p99 panel over a 300 s window every second, with no raw data kept: + /// one DDSketch per series in each of 300 one-second panes built at + /// ingestion, about 300 stored outputs in a DAG of over a thousand + /// nodes. Each stored output's build is compiled over the nodes below + /// it, and the whole plan's checks run once, so the nodes handed to the + /// executor grow with the plan, not with the plan times its stored + /// outputs (before, every build compiled the whole plan). + #[tokio::test] + async fn stored_output_builds_compile_only_the_nodes_below_them() { + let workload: crate::PlanningWorkload = serde_json::from_str( + r#"{"query_workload": {"language": "promql", "repeating_queries": [{ + "query": "avg by (job) (quantile_over_time(0.99, data[300s]))", + "demand": {"fixed_interval": 1000}, + "requirements": {"accuracy": {"max_error": 0.05, "failure_probability": 0.3, + "accuracy_scope": "whole_result"}}, + "predictability": {"predictable": {"known_at": null}}}]}, + "data_workload": {"arrival": {"ingesting": {"rows_per_second": 10000.0}}, + "sources": {"data": {"series_count": 1000, "metric_type": "gauge", + "sample_interval": 100, "values_non_negative": true}}}}"#, + ) + .unwrap(); + let capabilities = DeploymentCapabilities { + raw_data_retained: false, + ..DeploymentCapabilities::default() + }; + COMPILED_NODES.with(|count| count.set(0)); + let plan = plan_installable(crate::UserInput::new( + &workload, + crate::FrontendInput::Promql { histograms: None }, + crate::PlanningModels::builtin().with_deployment_capabilities(&capabilities), + )) + .await + .expect("a plan"); + let nodes = plan.dag.dag.nodes.len(); + let stored = plan.stored_outputs.len(); + assert!(stored >= 300, "{stored} stored outputs"); + // Stage 3's compile check builds the plan of each candidate it may + // select, so a few plans are compiled; each costs about its size. + let compiled = COMPILED_NODES.with(std::cell::Cell::get); + assert!( + compiled <= 40 * nodes, + "{compiled} nodes compiled for a plan of {nodes} nodes and {stored} stored outputs" + ); + } + /// `TimeShift(10 min) <- TimeRange(5 min) <- Scan`, all at query time: /// the shift above the range moves its window, which a window of a /// fixed offset below the range cannot say, so it is refused; without From 4ca2a9406e7c9e822b4d7f79b5b1f0d6e455bf98 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Wed, 21 Oct 2026 09:48:10 +0000 Subject: [PATCH 2/2] test(planner): count Stage 3 compile check's builds in the compiled-nodes test --- crates/planner/src/installable.rs | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/crates/planner/src/installable.rs b/crates/planner/src/installable.rs index b84fadf9f..e091ad4dd 100644 --- a/crates/planner/src/installable.rs +++ b/crates/planner/src/installable.rs @@ -471,6 +471,8 @@ impl InstallablePlan { &sub_dag } }; + #[cfg(test)] + tests::COMPILED_NODES.with(|count| count.set(count.get() + below.nodes.len())); compile(below, inputs, &[output.node_id as u64]) .map_err(|error| format!("stored output {}: {error}", output.node_id))?; } @@ -1297,18 +1299,20 @@ mod tests { use asap_types::ir::schema::{SketchAlgorithm, StatisticKind, SupportedSummary}; thread_local! { - /// The plan nodes handed to the executor's compiler by - /// [`runtime_inputs_of`], summed over its compiles. + /// The plan nodes handed to the executor's compiler by the stored + /// outputs' builds in [`runtime_inputs_of`] and + /// [`InstallablePlan::check_compiles`], summed over their compiles. pub(super) static COMPILED_NODES: std::cell::Cell = const { std::cell::Cell::new(0) }; } /// A p99 panel over a 300 s window every second, with no raw data kept: /// one DDSketch per series in each of 300 one-second panes built at - /// ingestion, about 300 stored outputs in a DAG of over a thousand - /// nodes. Each stored output's build is compiled over the nodes below - /// it, and the whole plan's checks run once, so the nodes handed to the + /// ingestion, about 300 stored outputs in a DAG of about 900 nodes. + /// Each stored output's build is compiled over the nodes below it, and + /// the whole plan's checks run once, so the nodes handed to the /// executor grow with the plan, not with the plan times its stored - /// outputs (before, every build compiled the whole plan). + /// outputs (before, every build compiled the whole plan). The count + /// covers both `plan_installable`'s builds and Stage 3's compile check. #[tokio::test] async fn stored_output_builds_compile_only_the_nodes_below_them() { let workload: crate::PlanningWorkload = serde_json::from_str(