From 3c9931b14a16177d2f357057ae849c0e283d2a39 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Sun, 18 Oct 2026 16:51:36 +0000 Subject: [PATCH 1/7] test(planner): pin that the facade's export carries Stage 3's selection PlanOutput::physical_asap_dag's doc said every node is exported at query time, so a runtime planning Example 4b without raw data could not trust the export and read Stage 3's plan through run_stage1_to_stage3 instead. The export already keeps each node's timing and materialization (the timing pass honors a timing already written); only untimed nodes take the default. The doc now says so, and a facade test checks the export of Example 4b equals Stage 3's selected plan, ingestion-time panes included. --- .../tests/runtime_handoff.rs | 87 +++++++++++++++++++ crates/plan-selection/src/pass/mod.rs | 10 ++- 2 files changed, 94 insertions(+), 3 deletions(-) create mode 100644 crates/integration-tests/tests/runtime_handoff.rs diff --git a/crates/integration-tests/tests/runtime_handoff.rs b/crates/integration-tests/tests/runtime_handoff.rs new file mode 100644 index 000000000..a35942ee3 --- /dev/null +++ b/crates/integration-tests/tests/runtime_handoff.rs @@ -0,0 +1,87 @@ +//! A deployment runtime consumes Stage 3's selected plan through the +//! facade's exported `PhysicalASAPDAG` document alone (issue #749). + +use asap_logical_asap_optimization::DeploymentCapabilities; +use asap_planner::{e2e_plan, FrontendInput, PlanOutput, PlanningModels, UserInput}; +use asap_types::ir::physical_export::{compile_physical_asap_workload, PhysicalASAPDAG}; +use asap_types::ir::properties::ExecutionTiming; +use asap_types::physical::Materialization; +use asap_types::workload::PlanningWorkload; + +/// #509 Example 4b: one repeating `quantile_over_time` over a 60-minute +/// window, evaluated every 10 minutes, over ingesting data. +fn example4b() -> PlanningWorkload { + serde_json::from_str( + r#"{ + "query_workload": { + "language": "promql", + "repeating_queries": [{ + "query": "quantile_over_time(0.99, latency_ms[60m])", + "demand": {"fixed_interval": 600000}, + "requirements": { + "accuracy": {"accuracy_metric": "rank", "max_error": 0.01, + "failure_probability": 0.01, "accuracy_scope": "whole_result"}, + "max_latency_ms": 200.0 + }, + "predictability": {"predictable": {"known_at": null}} + }] + }, + "data_workload": { + "arrival": {"ingesting": {"rows_per_second": 1000.0}}, + "sources": {"latency_ms": {"series_count": 1000, "sample_interval": 1000}} + } + }"#, + ) + .expect("Example 4b parses") +} + +/// A deployment that keeps no raw samples: every summary must be built at +/// ingestion time. +fn without_raw_data() -> DeploymentCapabilities { + DeploymentCapabilities { + raw_data_retained: false, + ..DeploymentCapabilities::default() + } +} + +async fn plan(workload: &PlanningWorkload, capabilities: &DeploymentCapabilities) -> PlanOutput { + let models = PlanningModels::builtin().with_deployment_capabilities(capabilities); + e2e_plan(UserInput::new( + workload, + FrontendInput::Promql { histograms: None }, + models, + )) + .await + .expect("Example 4b plans") +} + +fn materialized_count(dag: &PhysicalASAPDAG) -> usize { + dag.nodes + .iter() + .filter(|n| n.materialization != Materialization::NotMaterialized) + .count() +} + +/// Without raw data Stage 3 selects ingestion-time panes; the facade's +/// export carries that selection rather than resetting every node to +/// query time. +#[tokio::test] +async fn export_carries_the_selected_materialization() { + let capabilities = without_raw_data(); + let output = plan(&example4b(), &capabilities).await; + let dag = output.physical_asap_dag().expect("exports"); + assert!( + materialized_count(&dag) > 0, + "the selected panes are stored" + ); + assert!(dag + .nodes + .iter() + .any(|n| n.output_state.timing == ExecutionTiming::IngestionTime)); + // Exactly Stage 3's selected plan, as Stage 3 timed it. + let selection = output.selection.as_ref().expect("Stage 3 selected"); + assert_eq!( + dag, + compile_physical_asap_workload(&selection.plan).expect("the selected plan exports") + ); +} diff --git a/crates/plan-selection/src/pass/mod.rs b/crates/plan-selection/src/pass/mod.rs index 59d283689..001b2b779 100644 --- a/crates/plan-selection/src/pass/mod.rs +++ b/crates/plan-selection/src/pass/mod.rs @@ -262,9 +262,13 @@ impl PlanOutput { nodes } - /// The workload as one physical ASAP DAG, every node computed at query - /// time (the default [`MaterializationAssignment`]), with shared - /// sub-DAGs exported once: operator queries become its `roots` and standalone scalar + /// The workload as one physical ASAP DAG, with shared sub-DAGs exported + /// once. A node keeps the timing and materialization its pass gave it: + /// for [`Stage1ToStage3Pass`] that is Stage 3's selected plan as + /// selected (e.g. ingestion-time panes), the document a deployment + /// runtime executes. Only a node the pass left untimed takes the default + /// [`MaterializationAssignment`] (not materialized, computed when its + /// consumer runs). Operator queries become its `roots` and standalone scalar /// queries its `scalar_roots`, keyed by position in the batch. That /// position is the entry index, because `check_contract` requires dense /// entry indices. From ff1b8fb23191f76856bd75262f91a915e4e49216 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Sun, 18 Oct 2026 21:09:16 +0000 Subject: [PATCH 2/7] feat(executor): list the summary families the executor supports A deployment fills DeploymentCapabilities.supported_summaries, but had no list from the executor, so it declared every family the Planner builds and a plan with an unimplemented one failed only when the runtime bound it. supported_summary_families() returns each sketch family the factory builds per instance with the readouts an evaluation answers from it: plain Count Sketch (built, never read) and Theta (not built) are absent, Count-Min answers its total (its item frequency comes only from a Hydra grid, which Stage 2 does not produce yet), and the heap families answer top-k. A test checks every (family, readout) pair against create_planner_accumulator and the evaluation validators. SupportedSummary moves to asap-types beside StatisticKind so the executor can return it; asap-logical-asap-optimization re-exports it. --- crates/executor/src/lib.rs | 2 + crates/executor/src/supported_summaries.rs | 194 ++++++++++++++++++ .../src/deployment_capabilities.rs | 14 +- .../types/src/ir/schema/family_capability.rs | 12 ++ crates/types/src/ir/schema/mod.rs | 4 +- 5 files changed, 214 insertions(+), 12 deletions(-) create mode 100644 crates/executor/src/supported_summaries.rs diff --git a/crates/executor/src/lib.rs b/crates/executor/src/lib.rs index b75bd71e2..649615622 100644 --- a/crates/executor/src/lib.rs +++ b/crates/executor/src/lib.rs @@ -13,7 +13,9 @@ pub use traits::*; pub use expressions::arithmetic; pub mod capability; +mod supported_summaries; pub use summary_kernels::factory; +pub use supported_summaries::supported_summary_families; /// The exact Planner contract used by these kernels. pub use planner_types as planner; diff --git a/crates/executor/src/supported_summaries.rs b/crates/executor/src/supported_summaries.rs new file mode 100644 index 000000000..c67168abd --- /dev/null +++ b/crates/executor/src/supported_summaries.rs @@ -0,0 +1,194 @@ +//! The summary families this executor builds, with the readouts it answers +//! from each: what a deployment running it declares as +//! `DeploymentCapabilities::supported_summaries`, so the Planner offers no +//! summary the executor would reject when it binds the plan. + +use planner_types::ir::schema::{SketchAlgorithm, StatisticKind, SupportedSummary}; + +/// Every sketch family this executor builds from rows and the statistics it +/// answers from one state of it, in `SketchAlgorithm` order. A family the +/// executor builds but reads nothing from (plain Count Sketch) and one it +/// does not build (Theta) are absent. Exact accumulators are always +/// supported and are not listed. +/// +/// Per-instance states only: Count-Min answers its total (an item's +/// frequency only from a Hydra grid, which Stage 2 does not produce yet; +/// `SupportedSummary` gains a grouping dimension when it does), and the +/// heap families answer top-k only. +pub fn supported_summary_families() -> Vec { + use SketchAlgorithm as A; + use StatisticKind::*; + let families: [(A, &[StatisticKind]); 8] = [ + ( + A::UnivMon, + &[Total, Distinct, FrequencyL2, FrequencyEntropy], + ), + (A::Kll, &[Quantile]), + (A::Cms, &[Total]), + (A::Hll, &[Total, Distinct]), + (A::DDSketch, &[Total, Quantile]), + (A::CmsWithHeap, &[TopK]), + (A::Kmv, &[Total, Distinct]), + (A::CountSketchWithHeap, &[TopK]), + ]; + families + .into_iter() + .map(|(family, readouts)| SupportedSummary { + family, + readouts: readouts.to_vec(), + }) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::capability::validate_sketch_evaluation; + use crate::factory::create_planner_accumulator; + use crate::summary_kernels::weighted_frequency::WeightedFrequency; + use planner_types::ir::scalar::ColumnRef; + use planner_types::ir::schema::{ + FieldDataType, GroupingStrategy, SketchKind, SketchParams, SketchStatistic, + SummaryInputExpr, SummaryUpdate, + }; + + const ALGORITHMS: [SketchAlgorithm; 10] = [ + SketchAlgorithm::UnivMon, + SketchAlgorithm::Kll, + SketchAlgorithm::Cms, + SketchAlgorithm::Hll, + SketchAlgorithm::DDSketch, + SketchAlgorithm::CmsWithHeap, + SketchAlgorithm::Kmv, + SketchAlgorithm::Theta, + SketchAlgorithm::CountSketch, + SketchAlgorithm::CountSketchWithHeap, + ]; + + const READOUTS: [StatisticKind; 7] = [ + StatisticKind::Total, + StatisticKind::ItemFrequency, + StatisticKind::Quantile, + StatisticKind::Distinct, + StatisticKind::FrequencyL2, + StatisticKind::FrequencyEntropy, + StatisticKind::TopK, + ]; + + /// Valid parameters of `algorithm`, as the Planner sizes them. + fn params(algorithm: &SketchAlgorithm) -> SketchParams { + use SketchAlgorithm as A; + match algorithm { + A::UnivMon => SketchParams::UnivMon { + heap_size: 32, + sketch_rows: 3, + sketch_cols: 1024, + layers: 8, + distinct_failure_probability: None, + }, + A::Kll => SketchParams::Kll { + k: 200, + failure_probability: None, + }, + A::Cms => SketchParams::Cms { + width: 64, + depth: 3, + }, + A::Hll => SketchParams::Hll { + precision: 12, + relative_error: None, + }, + A::DDSketch => SketchParams::DDSketch { alpha: 0.01 }, + A::CmsWithHeap => SketchParams::CmsWithHeap { + width: 64, + depth: 3, + heap_size: 10, + }, + A::Kmv => SketchParams::Kmv { + k: 1024, + relative_error: None, + }, + A::Theta => SketchParams::Theta { k: 1024 }, + A::CountSketch => SketchParams::CountSketch { + width: 64, + depth: 3, + }, + A::CountSketchWithHeap => SketchParams::CountSketchWithHeap { + width: 64, + depth: 3, + heap_size: 10, + }, + } + } + + fn per_instance(algorithm: &SketchAlgorithm) -> FieldDataType { + FieldDataType::Sketch( + SketchKind::new(algorithm.clone(), params(algorithm)), + GroupingStrategy::PerSubpopulationInstance, + ) + } + + /// A statistic of kind `readout`. + fn statistic(readout: StatisticKind) -> SketchStatistic { + let key = ColumnRef::Named("item".into()); + match readout { + StatisticKind::Total => SketchStatistic::PointCount { key, value: None }, + StatisticKind::ItemFrequency => SketchStatistic::PointCount { + key, + value: Some("a".into()), + }, + StatisticKind::Quantile => SketchStatistic::Quantile { q: 0.99 }, + StatisticKind::Distinct => SketchStatistic::Cardinality, + StatisticKind::FrequencyL2 => SketchStatistic::FrequencyL2, + StatisticKind::FrequencyEntropy => SketchStatistic::FrequencyEntropy, + StatisticKind::TopK => SketchStatistic::TopK { k: 5 }, + } + } + + /// Whether the factory builds the per-instance `family` from rows, with + /// a value column or keyed items. + fn factory_builds(family: &FieldDataType) -> bool { + let FieldDataType::Sketch(_, grouping) = family else { + return false; + }; + let value = SummaryUpdate::column(ColumnRef::SampleValue); + let keyed = SummaryUpdate { + item: Some(SummaryInputExpr::Column(ColumnRef::Named("item".into()))), + ..value.clone() + }; + create_planner_accumulator(family, &value, grouping).is_ok() + || create_planner_accumulator(family, &keyed, grouping).is_ok() + } + + /// Whether the executor answers `readout` from `family`'s state: top-k + /// through the keyed heap evaluation, every other readout through the + /// sketch evaluation. + fn evaluation_answers(family: &FieldDataType, readout: StatisticKind) -> bool { + match (readout, family) { + (StatisticKind::TopK, FieldDataType::Sketch(kind, _)) => { + WeightedFrequency::configuration(kind).is_ok() + } + _ => validate_sketch_evaluation(family, &statistic(readout)).is_ok(), + } + } + + /// The list holds exactly the (family, readout) pairs the factory builds + /// per instance and an evaluation answers from that state. + #[test] + fn supported_summary_families_match_the_factory_and_evaluations() { + let listed = supported_summary_families(); + for algorithm in &ALGORITHMS { + // Per instance: Stage 1 builds every family per instance; the + // Hydra grid is not produced outside tests yet. + let family = per_instance(algorithm); + for readout in READOUTS { + let executed = factory_builds(&family) && evaluation_answers(&family, readout); + let declared = listed + .iter() + .any(|s| &s.family == algorithm && s.readouts.contains(&readout)); + assert_eq!(declared, executed, "{algorithm:?} {readout:?}"); + } + } + assert!(listed.iter().all(|s| !s.readouts.is_empty())); + } +} diff --git a/crates/logical-asap-optimization/src/deployment_capabilities.rs b/crates/logical-asap-optimization/src/deployment_capabilities.rs index adce4a2ed..ea2d25044 100644 --- a/crates/logical-asap-optimization/src/deployment_capabilities.rs +++ b/crates/logical-asap-optimization/src/deployment_capabilities.rs @@ -31,6 +31,9 @@ //! and keeps no query-time result across evaluations, which is the //! planner's behavior without deployment inputs. +// The executor lists the families it builds as `SupportedSummary`s, so the +// type lives with the readout vocabulary in `asap-types`. +pub use asap_types::ir::schema::SupportedSummary; use asap_types::ir::schema::{SketchAlgorithm, StatisticKind}; use asap_types::workload::{DataArrival, DataWorkload}; use serde::{Deserialize, Serialize}; @@ -54,17 +57,6 @@ pub struct DeploymentCapabilities { pub query_time_results_retained: bool, } -/// One sketch family the executor can build, with the readouts it can -/// answer from the family's state. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct SupportedSummary { - /// The family the executor can build. - pub family: SketchAlgorithm, - /// The statistics it can answer from one state of the family. - pub readouts: Vec, -} - impl Default for DeploymentCapabilities { fn default() -> Self { Self { diff --git a/crates/types/src/ir/schema/family_capability.rs b/crates/types/src/ir/schema/family_capability.rs index b8d07c153..ebcce086c 100644 --- a/crates/types/src/ir/schema/family_capability.rs +++ b/crates/types/src/ir/schema/family_capability.rs @@ -47,6 +47,18 @@ pub enum StatisticKind { TopK, } +/// One sketch family a deployment's executor can build, with the readouts +/// it can answer from the family's state: an entry of +/// `DeploymentCapabilities::supported_summaries`. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] +pub struct SupportedSummary { + /// The family the executor can build. + pub family: SketchAlgorithm, + /// The statistics it can answer from one state of the family. + pub readouts: Vec, +} + impl SketchStatistic { /// The kind of statistic this is. pub fn kind(&self) -> StatisticKind { diff --git a/crates/types/src/ir/schema/mod.rs b/crates/types/src/ir/schema/mod.rs index 3fd7f3e5e..b9705c7e3 100644 --- a/crates/types/src/ir/schema/mod.rs +++ b/crates/types/src/ir/schema/mod.rs @@ -20,7 +20,9 @@ pub mod schema_support; pub mod sketch; pub use aggregate_schema::aggregate_output_schema; -pub use family_capability::{FamilyCapability, MergeInputRule, MergedError, StatisticKind}; +pub use family_capability::{ + FamilyCapability, MergeInputRule, MergedError, StatisticKind, SupportedSummary, +}; pub use sketch::{ default_hydra_params, hydra_kind_for, EntityIdentity, ExactKind, ExactParams, GroupingStrategy, HydraKind, HydraParams, NonNegativeWeightProof, SamplingKind, SamplingParams, SketchAlgorithm, From 5bf9d48ade6236f491e7c84149211402ce9e50fb Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Sun, 18 Oct 2026 21:09:16 +0000 Subject: [PATCH 3/7] feat(types): raw-sample windows, the query-time DAG and shared panes A runtime matched Stage 2 candidate ids (L to P) and repeated Stage 3's sizing to get its correctness reference, the selected plan with nothing materialized. PhysicalASAPDAG::with_every_node_at_query_time rebuilds the DAG's IR untimed, applies the default materialization and re-exports: the same operators, summaries and parameters, every node at query time. RawSampleWindow names a window of raw samples a runtime feeds the DAG: TimeRange(width) <- TimeShift(offset)* <- Scan, with bounds_ms(T). Two workload-level helpers support the installable plan: apply_materialization_timings_to_roots (the loop PlanOutput::physical_asap_dag used, moved from plan-selection) and expand_pane_chains_in_workload. Pane expansion now memoizes panes by (chain, index): two merges over one chain (p99 and p50 over the same hour) read one set of panes, not two. --- crates/plan-selection/src/pass/mod.rs | 39 ++----- crates/types/src/ir/mod.rs | 5 +- crates/types/src/ir/operator/pane_chain.rs | 101 +++++++++++++++-- crates/types/src/ir/physical_export.rs | 119 ++++++++++++++++++++- crates/types/src/ir/properties/timing.rs | 30 ++++++ crates/types/tests/pane_chain.rs | 24 +++++ crates/types/tests/physical_export.rs | 49 +++++++++ 7 files changed, 326 insertions(+), 41 deletions(-) diff --git a/crates/plan-selection/src/pass/mod.rs b/crates/plan-selection/src/pass/mod.rs index 001b2b779..686e2bf38 100644 --- a/crates/plan-selection/src/pass/mod.rs +++ b/crates/plan-selection/src/pass/mod.rs @@ -19,10 +19,10 @@ use std::rc::Rc; use asap_types::ir::physical_export::compile_physical_asap_workload; use asap_types::ir::properties::timing::{ - apply_materialization_timings, MaterializationAssignment, TimingMemo, + apply_materialization_timings_to_roots, MaterializationAssignment, }; use asap_types::ir::properties::ExecutionDataStateError; -use asap_types::ir::{OperatorNode, QueryRoot}; +use asap_types::ir::OperatorNode; use asap_types::workload::parsed_workload::ParsedWorkload; use asap_types::workload::WorkloadError; @@ -280,14 +280,11 @@ impl PlanOutput { pub fn physical_asap_dag( &self, ) -> Result { - // One memo, so a node several roots share by `Rc` is timed and exported once. - let mut memo = TimingMemo::new(); - let assignment = MaterializationAssignment::default(); - let timed = self - .roots() - .iter() - .map(|root| time_query_root(root, &assignment, &mut memo)) - .collect::, _>>()?; + // A node several roots share by `Rc` is timed and exported once. + let timed = apply_materialization_timings_to_roots( + &self.roots(), + &MaterializationAssignment::default(), + )?; compile_physical_asap_workload(&timed) } @@ -300,28 +297,6 @@ impl PlanOutput { } } -/// Time one query root: an operator root directly, a standalone scalar by -/// timing each operator it references and rebuilding it over the timed nodes. -fn time_query_root( - root: &QueryRoot, - assignment: &MaterializationAssignment, - memo: &mut TimingMemo, -) -> Result { - Ok(match root { - QueryRoot::Operator(node) => { - QueryRoot::Operator(apply_materialization_timings(node, assignment, memo)?) - } - QueryRoot::Scalar(expr) => { - for node in expr.operator_refs() { - apply_materialization_timings(node, assignment, memo)?; - } - QueryRoot::Scalar(expr.map_operator_refs(&mut |node| { - Rc::clone(memo.timed(node).expect("scalar reference was timed")) - })) - } - }) -} - #[derive(Debug, thiserror::Error)] #[non_exhaustive] pub enum OptimizeError { diff --git a/crates/types/src/ir/mod.rs b/crates/types/src/ir/mod.rs index fefa42f62..5cb6719dd 100644 --- a/crates/types/src/ir/mod.rs +++ b/crates/types/src/ir/mod.rs @@ -17,6 +17,7 @@ pub mod flat; /// Physical ASAP DAG: the flattened operators plus execution timing. pub mod physical_export; pub use properties::timing::{ - apply_materialization_timings, data_state, is_filtered_build, planned_data_state, - split_shared_by_timing, validate_at_ingestion, MaterializationAssignment, TimingMemo, + apply_materialization_timings, apply_materialization_timings_to_roots, data_state, + is_filtered_build, planned_data_state, split_shared_by_timing, validate_at_ingestion, + MaterializationAssignment, TimingMemo, }; diff --git a/crates/types/src/ir/operator/pane_chain.rs b/crates/types/src/ir/operator/pane_chain.rs index 7fb0ff61d..d810caecd 100644 --- a/crates/types/src/ir/operator/pane_chain.rs +++ b/crates/types/src/ir/operator/pane_chain.rs @@ -15,6 +15,7 @@ use super::asap::ASAPOp; use super::node::{Operator, OperatorNode}; use super::non_asap::{NonASAPOp, TimeRangeKind}; use super::operator_properties::TimeShift; +use super::query::QueryRoot; use crate::ir::properties::ExecutionTiming; use crate::ir::SchemaDerivationError; @@ -161,7 +162,74 @@ fn build_over( pub fn expand_pane_chains( root: &Rc, ) -> Result, SchemaDerivationError> { - expand(root, &mut HashMap::new()) + expand(root, &mut ExpansionMemo::default()) +} + +/// A workload with its pane chains expanded +/// ([`expand_pane_chains_in_workload`]). +#[derive(Debug, Clone)] +pub struct PaneChainExpansion { + /// The workload's roots, in order, with every chain expanded. + pub roots: Vec, + /// Each expanded chain with its panes, newest first, in the order the + /// chains were first reached. + pub chains: Vec, +} + +/// One expanded chain: the `PaneChain` node of the input workload and the +/// `SummaryAgg` panes that replace it. +#[derive(Debug, Clone)] +pub struct ExpandedPaneChain { + /// The chain node, as the input workload holds it. + pub chain: Rc, + /// Its panes in the expanded workload, newest first. + pub panes: Vec>, +} + +/// [`expand_pane_chains`] over every root of one workload, with one memo, so +/// a node several roots share stays one node and a chain several merges +/// read expands into one set of panes. A standalone scalar root has the +/// operators it reads expanded. Also returns each chain's panes, so a +/// caller can carry what it knows about a chain (e.g. its retention) to +/// them. +pub fn expand_pane_chains_in_workload( + roots: &[QueryRoot], +) -> Result { + let mut memo = ExpansionMemo::default(); + let roots = roots + .iter() + .map(|root| match root { + QueryRoot::Operator(node) => expand(node, &mut memo).map(QueryRoot::Operator), + QueryRoot::Scalar(expr) => { + let mut error = None; + let expr = expr.map_operator_refs(&mut |node| { + expand(node, &mut memo).unwrap_or_else(|e| { + error.get_or_insert(e); + Rc::clone(node) + }) + }); + match error { + Some(e) => Err(e), + None => Ok(QueryRoot::Scalar(expr)), + } + } + }) + .collect::, _>>()?; + let chains = memo + .chains + .into_iter() + .map(|chain| { + let pane_count = match chain.asap() { + Some(ASAPOp::PaneChain { pane_count, .. }) => *pane_count, + _ => unreachable!("only chains are recorded"), + }; + let panes = (0..pane_count) + .map(|index| Rc::clone(&memo.panes[&(Rc::as_ptr(&chain), index)])) + .collect(); + ExpandedPaneChain { chain, panes } + }) + .collect(); + Ok(PaneChainExpansion { roots, chains }) } /// The untimed `node` with `timing` written into it and into every untimed @@ -180,11 +248,22 @@ fn timed_like(node: &Rc, timing: ExecutionTiming) -> OperatorNode .with_timing(Some(timing)) } +/// What one expansion has built: each expanded node by its input's +/// address, and each pane by its chain's address and index, so a chain two +/// merges read expands into one set of panes. +#[derive(Default)] +struct ExpansionMemo { + nodes: HashMap<*const OperatorNode, Rc>, + panes: HashMap<(*const OperatorNode, u32), Rc>, + /// Each chain expanded, in the order first reached. + chains: Vec>, +} + fn expand( node: &Rc, - memo: &mut HashMap<*const OperatorNode, Rc>, + memo: &mut ExpansionMemo, ) -> Result, SchemaDerivationError> { - if let Some(done) = memo.get(&Rc::as_ptr(node)) { + if let Some(done) = memo.nodes.get(&Rc::as_ptr(node)) { return Ok(Rc::clone(done)); } let out = match &node.operator { @@ -197,14 +276,24 @@ fn expand( match child.asap() { Some(ASAPOp::PaneChain { pane_count, .. }) => { for index in 0..*pane_count { + let key = (Rc::as_ptr(child), index); + if let Some(done) = memo.panes.get(&key) { + inputs.push(Rc::clone(done)); + continue; + } let pane = Rc::new(pane(child, index)?); - inputs.push(match child.timing { + let pane = match child.timing { Some(timing) => Rc::new( timed_like(&pane, timing) .with_materialization(child.materialization), ), None => pane, - }); + }; + if index == 0 { + memo.chains.push(Rc::clone(child)); + } + memo.panes.insert(key, Rc::clone(&pane)); + inputs.push(pane); } } _ => inputs.push(expand(child, memo)?), @@ -243,6 +332,6 @@ fn expand( } } }; - memo.insert(Rc::as_ptr(node), Rc::clone(&out)); + memo.nodes.insert(Rc::as_ptr(node), Rc::clone(&out)); Ok(out) } diff --git a/crates/types/src/ir/physical_export.rs b/crates/types/src/ir/physical_export.rs index 3e27090c2..5efc8f51e 100644 --- a/crates/types/src/ir/physical_export.rs +++ b/crates/types/src/ir/physical_export.rs @@ -24,7 +24,10 @@ use crate::ir::properties::execution_data_state::{ ExecutionDataState, ExecutionDataStateError, ExecutionTiming, }; use crate::ir::properties::guarantee::AccuracyGuarantee; -use crate::ir::properties::timing::{data_state, is_materialization_point}; +use crate::ir::properties::timing::{ + apply_materialization_timings_to_roots, data_state, is_materialization_point, + MaterializationAssignment, +}; use crate::ir::schema::{FieldDataType, Schema}; use crate::physical::materialization::Materialization; @@ -186,7 +189,121 @@ impl PhysicalASAPDAGDocument { } } +/// A window of raw samples a deployment runtime feeds the DAG: at +/// evaluation time `T`, the samples in `(T - end_offset_ms - +/// width_ms, T - end_offset_ms]` that the `Scan` +/// `scan_node_id` selects, bound as the input rows of the `TimeRange` +/// `time_range_node_id`. The shape is `TimeRange(width) <- +/// TimeShift(offset)* <- Scan`, with no `TimeShift`, `TimeRange` or +/// subquery above the `TimeRange`, so the window depends on `T` alone. For +/// an instant selector `width_ms` is its lookback. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct RawSampleWindow { + /// The `TimeRange` node the window's input rows are bound to. + pub time_range_node_id: PhysicalASAPNodeId, + /// Width of the window, from the `TimeRange`. + pub width_ms: i64, + /// How long before the evaluation time the window ends: the sum of the + /// offset `TimeShift`s between the `TimeRange` and the `Scan`. + pub end_offset_ms: i64, + /// The `Scan` below, whose source and label matchers select the samples. + pub scan_node_id: PhysicalASAPNodeId, +} + +impl RawSampleWindow { + /// How far back from the evaluation time the window reaches: + /// `end_offset_ms + width_ms`. + pub fn reach_ms(&self) -> u64 { + u64::try_from(self.end_offset_ms.saturating_add(self.width_ms)).unwrap_or(0) + } + + /// `(start, end]` of the window at evaluation time `evaluation_time_ms`. + pub fn bounds_ms(&self, evaluation_time_ms: i64) -> (i64, i64) { + let end = evaluation_time_ms - self.end_offset_ms; + (end - self.width_ms, end) + } +} + +/// Why [`PhysicalASAPDAG::with_every_node_at_query_time`] failed. +#[derive(Debug, Error)] +pub enum ReferencePlanError { + /// Bad input: the document is not a valid physical ASAP DAG. + #[error("invalid physical ASAP DAG: {0}")] + InvalidDag(#[from] PhysicalASAPDAGValidationError), + /// Bad input: the plan cannot run at query time (a data-state rule of + /// the timing pass fails). + #[error("the plan does not run at query time: {0}")] + Timing(#[from] ExecutionDataStateError), +} + impl PhysicalASAPDAG { + /// This DAG's plan with nothing materialized: the same operators, + /// summaries and summary parameters, with every sub-DAG + /// `NotMaterialized` and so every node at query time (the default + /// [`MaterializationAssignment`]). Node ids are renumbered, and `roots` + /// and `scalar_roots` keep their order. + /// + /// The correctness reference of a selected plan: a deployment runtime + /// compares the answers of the plan it installed with the executor + /// running this one over raw data, which needs the raw data the + /// installed plan may not keep. + pub fn with_every_node_at_query_time(&self) -> Result { + let roots = self.untimed_query_roots()?; + let timed = + apply_materialization_timings_to_roots(&roots, &MaterializationAssignment::default())?; + Ok(compile_physical_asap_workload(&timed)?) + } + + /// The IR this DAG exports, one root per query in batch order, every + /// node untimed and `NotMaterialized`: the inverse of + /// [`compile_physical_asap_workload`] up to `Rc` identity and timing. + fn untimed_query_roots(&self) -> Result, PhysicalASAPDAGValidationError> { + self.validate()?; + let by_id: HashMap = + self.nodes.iter().map(|node| (node.id, node)).collect(); + let mut rebuilt: HashMap> = HashMap::new(); + // Children first, without recursion: `validate` rejected cycles. + for node in &self.nodes { + let mut pending = vec![(node.id, false)]; + while let Some((id, children_done)) = pending.pop() { + if rebuilt.contains_key(&id) { + continue; + } + let node = by_id[&id]; + if !children_done { + pending.push((id, true)); + pending.extend(node.payload.children().into_iter().map(|c| (*c, false))); + continue; + } + let operator = node + .payload + .map_children(|child| Rc::clone(&rebuilt[child])); + let restored = OperatorNode::with_schema(operator, node.output_schema.clone()) + .with_guarantee(node.guarantee.clone()); + rebuilt.insert(id, Rc::new(restored)); + } + } + let mut operator_roots = self.roots.iter(); + let mut scalar_roots = self.scalar_roots.iter().peekable(); + let mut roots = Vec::with_capacity(self.roots.len() + self.scalar_roots.len()); + for position in 0..self.roots.len() + self.scalar_roots.len() { + match scalar_roots.peek() { + Some((at, expr)) if *at == position => { + roots.push(QueryRoot::Scalar( + expr.map_operator_refs(&mut |id| Rc::clone(&rebuilt[id])), + )); + scalar_roots.next(); + } + _ => { + let id = operator_roots.next().expect("validated root positions"); + roots.push(QueryRoot::Operator(Rc::clone(&rebuilt[id]))); + } + } + } + Ok(roots) + } + pub fn validate(&self) -> Result<(), PhysicalASAPDAGValidationError> { let mut nodes = HashMap::new(); for node in &self.nodes { diff --git a/crates/types/src/ir/properties/timing.rs b/crates/types/src/ir/properties/timing.rs index ba499af5b..99fb49a1d 100644 --- a/crates/types/src/ir/properties/timing.rs +++ b/crates/types/src/ir/properties/timing.rs @@ -59,6 +59,7 @@ use crate::ir::operator::asap::ASAPOp; use crate::ir::operator::node::{Operator, OperatorNode}; use crate::ir::operator::non_asap::NonASAPOp; use crate::ir::operator::operator_properties::BinaryOpKind; +use crate::ir::operator::query::QueryRoot; use crate::ir::properties::execution_data_state::{ DataPrimitive, ExecutionDataState, ExecutionDataStateError, ExecutionTiming, }; @@ -302,6 +303,35 @@ pub fn apply_materialization_timings( Ok(timed) } +/// [`apply_materialization_timings`] over every root of one workload with +/// one memo, so a node several roots share is timed once and stays one +/// node. A standalone scalar root has each operator it reads timed, and is +/// rebuilt over the timed nodes. +pub fn apply_materialization_timings_to_roots( + roots: &[QueryRoot], + assignment: &MaterializationAssignment, +) -> Result, ExecutionDataStateError> { + let mut memo = TimingMemo::new(); + roots + .iter() + .map(|root| { + Ok(match root { + QueryRoot::Operator(node) => { + QueryRoot::Operator(apply_materialization_timings(node, assignment, &mut memo)?) + } + QueryRoot::Scalar(expr) => { + for node in expr.operator_refs() { + apply_materialization_timings(node, assignment, &mut memo)?; + } + QueryRoot::Scalar(expr.map_operator_refs(&mut |node| { + Rc::clone(memo.timed(node).expect("scalar reference was timed")) + })) + } + }) + }) + .collect() +} + /// Validate the sub-DAG below `root` with every sub-DAG `MatAtIngestion` /// where the DAG allows it ([`MaterializationAssignment::with_default`]) and /// `root` consumed at `root_timing`. For planning-time legality checks of a diff --git a/crates/types/tests/pane_chain.rs b/crates/types/tests/pane_chain.rs index eac2b2bfe..24e3064d6 100644 --- a/crates/types/tests/pane_chain.rs +++ b/crates/types/tests/pane_chain.rs @@ -231,3 +231,27 @@ fn a_chain_reads_an_instant_vector() { .unwrap(); assert!(empty.validate_structure().is_err()); } + +/// Two merges over one chain (two queries reading the same panes) expand +/// into one set of panes: the same five `SummaryAgg` nodes, not ten. +#[test] +fn merges_over_one_chain_share_its_panes() { + use asap_types::ir::operator::pane_chain::expand_pane_chains_in_workload; + use asap_types::ir::QueryRoot; + let shared = chain(scan()); + let roots = [ + QueryRoot::Operator(merge(vec![Rc::clone(&shared)])), + QueryRoot::Operator(merge(vec![shared])), + ]; + let expansion = expand_pane_chains_in_workload(&roots).unwrap(); + assert_eq!(expansion.chains.len(), 1); + assert_eq!(expansion.chains[0].panes.len(), 5); + let expanded = expansion.roots; + let panes = |root: &QueryRoot| match root { + QueryRoot::Operator(node) => node.children().into_iter().cloned().collect::>(), + QueryRoot::Scalar(_) => unreachable!(), + }; + let (first, second) = (panes(&expanded[0]), panes(&expanded[1])); + assert_eq!(first.len(), 5); + assert!(first.iter().zip(&second).all(|(a, b)| Rc::ptr_eq(a, b))); +} diff --git a/crates/types/tests/physical_export.rs b/crates/types/tests/physical_export.rs index b52f08bfa..70051a3b1 100644 --- a/crates/types/tests/physical_export.rs +++ b/crates/types/tests/physical_export.rs @@ -464,3 +464,52 @@ fn physical_node_ids_serialize_in_deterministic_binding_order() { r#"{"2":"query-2","10":"materialization-10"}"# ); } + +/// The plan with every node at query time is the same operators with +/// nothing materialized: what the default materialization exports. The +/// shared summary stays one node, and the scalar root that reads it keeps +/// its batch position. +#[test] +fn every_node_at_query_time_keeps_the_operators_and_stores_nothing() { + use asap_types::ir::scalar::ArithmeticOpKind; + use asap_types::ir::{ExprSemantics, ScalarExpr}; + let workload = |root: Rc| { + let scalar = ScalarExpr::Arithmetic { + op: ArithmeticOpKind::Add, + left: Box::new(ScalarExpr::PromqlScalarFromVector(Rc::clone(&root))), + right: Box::new(ScalarExpr::literal_f64(1.0)), + semantics: ExprSemantics::Promql, + }; + compile_physical_asap_workload(&[ + QueryRoot::Operator(Rc::clone(&root)), + QueryRoot::Scalar(scalar), + QueryRoot::Operator(root), + ]) + .unwrap() + }; + let materialized = workload(timed_at_ingestion()); + let at_query_time = apply_materialization_timings( + &plan(), + &MaterializationAssignment::default(), + &mut TimingMemo::new(), + ) + .unwrap(); + assert_eq!( + materialized.with_every_node_at_query_time().unwrap(), + workload(at_query_time) + ); +} + +/// A window 600,000 ms wide ending 300,000 ms before T = 3,600,000 covers +/// `(2,700,000, 3,300,000]`, and reaches 900,000 ms back. +#[test] +fn a_raw_sample_window_ends_its_offset_before_the_evaluation() { + let window = asap_types::ir::physical_export::RawSampleWindow { + time_range_node_id: 1, + width_ms: 600_000, + end_offset_ms: 300_000, + scan_node_id: 0, + }; + assert_eq!(window.bounds_ms(3_600_000), (2_700_000, 3_300_000)); + assert_eq!(window.reach_ms(), 900_000); +} From 282f11cfd2e708725a253a6662e571b47d45414a Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Sun, 18 Oct 2026 21:09:17 +0000 Subject: [PATCH 4/7] feat(plan-selection): load Stage 3's settings from JSON CostModel, ResourceUnitCost and PlanSelectionSettings could only be built in Rust, so a deployment could not pass its prices as a file (only --measured-operation-costs was loadable). PlanSelectionSettings, CostModel, ResourceUnitCost, ResourceSpeed, ClusterLimits and ClusterObjective now serialize: the enums in snake_case ({"usage_over_time": {...}}), every struct refusing unknown fields, and an omitted PlanSelectionSettings field taking its default. stage_pipeline --plan-selection-settings reads the whole settings from one such file; it replaces --cost-model, the --cluster-* flags and --measured-operation-costs, and combining them is refused. --- crates/devtools/src/bin/stage_pipeline.rs | 41 +++++++++-- crates/devtools/tests/stage_pipeline.rs | 39 ++++++++++ crates/plan-selection/src/lib.rs | 22 ++++-- .../tests/plan_selection_settings_json.rs | 72 +++++++++++++++++++ 4 files changed, 163 insertions(+), 11 deletions(-) create mode 100644 crates/plan-selection/tests/plan_selection_settings_json.rs diff --git a/crates/devtools/src/bin/stage_pipeline.rs b/crates/devtools/src/bin/stage_pipeline.rs index ac9335ca6..32452a36b 100644 --- a/crates/devtools/src/bin/stage_pipeline.rs +++ b/crates/devtools/src/bin/stage_pipeline.rs @@ -111,6 +111,13 @@ // writes from sketch-bench's `rqe_atomic_costs.json`); a sketch node with an // entry is priced from it, and each node's `price_source` says `measured` // or `analytical`. +// `--plan-selection-settings ` reads all of Stage 3's settings from +// one JSON file (`PlanSelectionSettings`: `cost_model`, e.g. +// `{"usage_over_time": {"cost_per_cpu_op": 1e-6, ...}}` with the +// deployment's own unit costs, `resource_speed`, +// `non_repeating_query_horizon_seconds` and `measured_operation_costs`; +// an omitted field takes its default). It replaces `--cost-model`, the +// `--cluster-*` flags and `--measured-operation-costs`. // `--sample-interval-ms` is the sample interval declared for every metric the // PromQL queries read (default 15000); the data declares no other facts. // `--table ''` (repeatable) declares a table the SQL queries read: `{"name": ..., @@ -154,7 +161,7 @@ const USAGE: &str = [--name ]... [--table ...] [--max-error --failure-probability --accuracy-metric ] [--sample-interval-ms ])) \ [--cost-model usage-over-time|aws-fargate-us-east-1|provisioned-capacity|cluster-capacity [--cluster-memory-bytes \ --cluster-disk-bytes [--cluster-cpu-cores ] --cluster-objective lowest-latency|least-memory|pareto-front]] \ -[--deployment-capabilities ] [--max-combinations ] [--max-displayed ] --out "; +[--plan-selection-settings ] [--deployment-capabilities ] [--max-combinations ] [--max-displayed ] --out "; /// The default `--max-combinations`: the most Stage 1 candidates (Pass 2's /// included) a run enumerates. @@ -245,6 +252,7 @@ fn run(args: Vec) -> Result<(), Failure> { let mut deployment_capabilities = DeploymentCapabilities::default(); let mut out = None; let mut measured_operation_costs_path: Option = None; + let mut plan_selection_settings_path: Option = None; let mut args = args.into_iter(); while let Some(flag) = args.next() { let mut value = || args.next().ok_or(format!("{flag} needs a value")); @@ -299,6 +307,7 @@ fn run(args: Vec) -> Result<(), Failure> { } "--out" => out = Some(value()?), "--measured-operation-costs" => measured_operation_costs_path = Some(value()?), + "--plan-selection-settings" => plan_selection_settings_path = Some(value()?), other => return Err(format!("unknown argument {other}").into()), } } @@ -309,6 +318,20 @@ fn run(args: Vec) -> Result<(), Failure> { return Err("--table needs --sql".into()); } let out = out.ok_or("--out is required")?; + if plan_selection_settings_path.is_some() + && (cost_model.is_some() + || cluster_memory.is_some() + || cluster_disk.is_some() + || cluster_cores.is_some() + || cluster_objective.is_some() + || measured_operation_costs_path.is_some()) + { + return Err( + "--plan-selection-settings replaces --cost-model, the --cluster-* flags \ + and --measured-operation-costs" + .into(), + ); + } let cluster = (cluster_memory, cluster_disk, cluster_objective.as_deref()); // The settings every flag below leaves at its default; the Fargate // preset converts its prices at their `resource_speed`. @@ -360,10 +383,18 @@ fn run(args: Vec) -> Result<(), Failure> { } None => Vec::new(), }; - let plan_selection = PlanSelectionSettings { - cost_model, - measured_operation_costs, - ..default_settings + let plan_selection = match &plan_selection_settings_path { + Some(path) => { + let text = std::fs::read_to_string(path) + .map_err(|e| format!("--plan-selection-settings {path}: {e}"))?; + serde_json::from_str(&text) + .map_err(|e| format!("--plan-selection-settings {path}: {e}"))? + } + None => PlanSelectionSettings { + cost_model, + measured_operation_costs, + ..default_settings + }, }; let from_file = match &workload_path { Some(path) if example.is_none() && queries.is_empty() => { diff --git a/crates/devtools/tests/stage_pipeline.rs b/crates/devtools/tests/stage_pipeline.rs index d1b6fb19c..7059a4553 100644 --- a/crates/devtools/tests/stage_pipeline.rs +++ b/crates/devtools/tests/stage_pipeline.rs @@ -949,3 +949,42 @@ fn workload_file_with_a_range_evaluation_is_planned() { ); std::fs::remove_file(path).unwrap(); } + +/// `--plan-selection-settings ` loads Stage 3's settings from JSON: +/// cost model 2 with the illustrative prices selects what +/// `--cost-model provisioned-capacity` selects for Example 4 (P1). The file +/// replaces the cost-model flags, so combining them is refused. +#[test] +fn plan_selection_settings_flag_loads_the_settings_from_json() { + let path = std::env::temp_dir().join(format!( + "stage_pipeline_{}_plan_selection_settings.json", + std::process::id() + )); + let settings = asap_plan_selection::PlanSelectionSettings { + cost_model: asap_plan_selection::CostModel::ProvisionedCapacity( + asap_plan_selection::ResourceUnitCost::illustrative(), + ), + ..Default::default() + }; + std::fs::write(&path, serde_json::to_string(&settings).unwrap()).unwrap(); + let path_arg = path.to_str().unwrap(); + let document = generate_example( + "planner-layering-3b", + &["--plan-selection-settings", path_arg], + ); + assert_eq!(document["stage3_plan_selection"]["selected"], "P1"); + let (code, stderr) = failure(&[ + "--example", + "planner-layering-3b", + "--plan-selection-settings", + path_arg, + "--cost-model", + "usage-over-time", + ]); + assert_eq!(code, Some(2)); + assert!( + stderr.contains("--plan-selection-settings replaces --cost-model"), + "{stderr}" + ); + std::fs::remove_file(path).unwrap(); +} diff --git a/crates/plan-selection/src/lib.rs b/crates/plan-selection/src/lib.rs index 967a33d4d..f7ae1cfe1 100644 --- a/crates/plan-selection/src/lib.rs +++ b/crates/plan-selection/src/lib.rs @@ -308,7 +308,8 @@ pub struct ResourceUsage { /// The price of one unit of each resource, per second of wall time. Cost /// model 1 reads the usage prices (CPU, scan, memory byte-seconds, disk); /// cost model 2 the provisioned prices (memory, CPU cores) and disk. -#[derive(Debug, Clone, PartialEq)] +#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] pub struct ResourceUnitCost { /// Price of one CPU operation. pub cost_per_cpu_op: f64, @@ -396,7 +397,8 @@ impl ResourceUnitCost { } /// How fast one core does work; the latency check reads it. -#[derive(Debug, Clone, Copy, PartialEq)] +#[derive(Debug, Clone, Copy, PartialEq, serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] pub struct ResourceSpeed { /// Milliseconds one CPU operation takes on one core. pub ms_per_cpu_op: f64, @@ -418,7 +420,10 @@ impl ResourceSpeed { /// How plan selection turns each valid candidate's [`ResourceUsage`] into a /// choice. Chosen by the deployment ([`PlanSelectionSettings::cost_model`]). -#[derive(Debug, Clone, PartialEq)] +/// In JSON, `{"usage_over_time": {...}}`, `{"provisioned_capacity": {...}}` +/// or `{"cluster_capacity": {...}}`. +#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "snake_case")] pub enum CostModel { /// Cost model 1, pay-as-you-go: Σ unit cost × usage per second (CPU /// operations, scanned bytes, memory byte-seconds, disk byte-seconds); @@ -434,7 +439,8 @@ pub enum CostModel { } /// A fixed cluster's capacity (cost model 3) and how to choose within it. -#[derive(Debug, Clone, PartialEq)] +#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] pub struct ClusterLimits { /// Memory available; a candidate whose peak memory is above it is invalid. pub memory_bytes: u64, @@ -447,7 +453,8 @@ pub struct ClusterLimits { } /// What cost model 3 optimizes among the candidates that fit. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "snake_case")] pub enum ClusterObjective { /// The smallest worst-query latency; ties by peak memory, then disk. LowestLatency, @@ -461,7 +468,10 @@ pub enum ClusterObjective { /// The deployment's choices for plan selection (Stage 3): the cost model, /// how fast a core works, and how long one-off queries are spread over. -#[derive(Debug, Clone, PartialEq)] +/// Loadable from JSON (`stage_pipeline --plan-selection-settings `); +/// an omitted field takes its default, an unknown one is refused. +#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)] +#[serde(default, deny_unknown_fields)] pub struct PlanSelectionSettings { /// How candidates are compared; default cost model 1 with /// [`ResourceUnitCost::illustrative`]. diff --git a/crates/plan-selection/tests/plan_selection_settings_json.rs b/crates/plan-selection/tests/plan_selection_settings_json.rs new file mode 100644 index 000000000..b53a6d626 --- /dev/null +++ b/crates/plan-selection/tests/plan_selection_settings_json.rs @@ -0,0 +1,72 @@ +//! A deployment passes its Stage 3 cost-model settings as JSON (issue #749). + +use asap_plan_selection::{ + ClusterLimits, ClusterObjective, CostModel, PlanSelectionSettings, ResourceSpeed, + ResourceUnitCost, +}; + +/// Cost model 2 with the deployment's own prices and speed; the fields left +/// out keep their defaults. +#[test] +fn settings_load_from_json_with_defaults_for_omitted_fields() { + let settings: PlanSelectionSettings = serde_json::from_str( + r#"{ + "cost_model": {"provisioned_capacity": { + "cost_per_cpu_op": 2e-6, + "cost_per_scanned_byte": 1e-7, + "cost_per_memory_byte_second": 1.25e-7, + "cost_per_provisioned_memory_byte_second": 3e-7, + "cost_per_provisioned_cpu_core_second": 900.0, + "cost_per_disk_byte_second": 1e-9 + }}, + "resource_speed": {"ms_per_cpu_op": 2e-6, "ms_per_scanned_byte": 1e-7} + }"#, + ) + .unwrap(); + assert_eq!( + settings, + PlanSelectionSettings { + cost_model: CostModel::ProvisionedCapacity(ResourceUnitCost { + cost_per_cpu_op: 2e-6, + cost_per_scanned_byte: 1e-7, + cost_per_memory_byte_second: 1.25e-7, + cost_per_provisioned_memory_byte_second: 3e-7, + cost_per_provisioned_cpu_core_second: 900.0, + cost_per_disk_byte_second: 1e-9, + }), + resource_speed: ResourceSpeed { + ms_per_cpu_op: 2e-6, + ms_per_scanned_byte: 1e-7, + }, + ..PlanSelectionSettings::default() + } + ); + settings.validate().unwrap(); + let empty: PlanSelectionSettings = serde_json::from_str("{}").unwrap(); + assert_eq!(empty, PlanSelectionSettings::default()); +} + +/// Every cost model round-trips, and a misspelled field is refused rather +/// than silently defaulted. +#[test] +fn settings_round_trip_and_refuse_unknown_fields() { + let cluster = PlanSelectionSettings { + cost_model: CostModel::ClusterCapacity(ClusterLimits { + memory_bytes: 64 << 30, + disk_bytes: 1 << 40, + cpu_cores: None, + objective: ClusterObjective::ParetoFront, + }), + ..PlanSelectionSettings::default() + }; + for settings in [PlanSelectionSettings::default(), cluster] { + let json = serde_json::to_string(&settings).unwrap(); + let back: PlanSelectionSettings = serde_json::from_str(&json).unwrap(); + assert_eq!(back, settings); + } + assert!(serde_json::from_str::(r#"{"costmodel": {}}"#).is_err()); + assert!(serde_json::from_str::( + r#"{"resource_speed": {"ms_per_cpu_op": 1e-6, "ms_per_scanned_byte": 1e-7, "x": 1}}"# + ) + .is_err()); +} From 0cb818c803dddfc48cb16f595dbad5ca6a448739 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Sun, 18 Oct 2026 21:09:35 +0000 Subject: [PATCH 5/7] feat(executor): feed raw samples and read answer labels through a public API A runtime turned Remote Write samples into input rows with physical_planner::promql_rows::series_row and read answer labels with decode_series_identity: the executor's row encoding. Binding rows at a pane's TimeRange also skipped the Scan, so label matchers were dropped, and an instant selector counted every sample in its lookback. RawSampleSelector::new(dag, window) prepares one raw-sample window once: metric() routes samples by metric, and rows(window_end_ms, samples) keeps the samples in (end - width, end] that the Scan's compiled predicates keep and, for an instant selector, only each series' latest one, in the TimeRange's schema. series_labels_of_output_row reads an answer row's labels; Sample is the input. promql_rows and frontier_from_timing become crate-private; the test-support feature, which the Planner's own executor and integration tests enable, keeps them reachable for those tests. --- Cargo.lock | 1 + crates/executor/Cargo.toml | 6 + crates/executor/src/lib.rs | 2 + .../src/physical_planner/candidates.rs | 1 + crates/executor/src/physical_planner/mod.rs | 17 +- .../src/physical_planner/promql_rows.rs | 5 + crates/executor/src/samples.rs | 331 ++++++++++++++++++ crates/integration-tests/Cargo.toml | 2 +- 8 files changed, 361 insertions(+), 4 deletions(-) create mode 100644 crates/executor/src/samples.rs diff --git a/Cargo.lock b/Cargo.lock index 2611563e3..2b179651d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -333,6 +333,7 @@ dependencies = [ name = "asap-executor" version = "0.1.0" dependencies = [ + "asap-executor", "asap-frontend-promql", "asap-logical-asap-optimization", "asap-plan-selection", diff --git a/crates/executor/Cargo.toml b/crates/executor/Cargo.toml index cf4c2304a..ccfce2721 100644 --- a/crates/executor/Cargo.toml +++ b/crates/executor/Cargo.toml @@ -15,7 +15,13 @@ thiserror = "2" regex = "1" +[features] +# Executor internals the Planner's own tests use (the PromQL row encoding, +# the timing frontier); a deployment runtime does not need them. +test-support = [] + [dev-dependencies] +asap-executor = { path = ".", features = ["test-support"] } rmp-serde = "1" asap-logical-asap-optimization = { path = "../logical-asap-optimization" } asap-plan-selection = { path = "../plan-selection" } diff --git a/crates/executor/src/lib.rs b/crates/executor/src/lib.rs index 649615622..91091e79a 100644 --- a/crates/executor/src/lib.rs +++ b/crates/executor/src/lib.rs @@ -13,6 +13,8 @@ pub use traits::*; pub use expressions::arithmetic; pub mod capability; +mod samples; +pub use samples::{series_labels_of_output_row, RawSampleSelector, Sample}; mod supported_summaries; pub use summary_kernels::factory; pub use supported_summaries::supported_summary_families; diff --git a/crates/executor/src/physical_planner/candidates.rs b/crates/executor/src/physical_planner/candidates.rs index 57eba3d4e..410a6d917 100644 --- a/crates/executor/src/physical_planner/candidates.rs +++ b/crates/executor/src/physical_planner/candidates.rs @@ -96,6 +96,7 @@ pub fn cut_candidate( /// A query-time node feeding an ingestion-time node has no valid placement. /// Only ingestion timing is read: a `MatAtQuery` output is not stored by /// this frontier yet (choosing and storing it is Stage 2's job). +#[cfg(any(test, feature = "test-support"))] pub fn frontier_from_timing(dag: &PhysicalASAPDAG) -> Result, Error> { use planner_types::ir::properties::ExecutionTiming::IngestionTime; let timing = dag diff --git a/crates/executor/src/physical_planner/mod.rs b/crates/executor/src/physical_planner/mod.rs index 8ee3e9a3d..6c74a5f10 100644 --- a/crates/executor/src/physical_planner/mod.rs +++ b/crates/executor/src/physical_planner/mod.rs @@ -19,6 +19,9 @@ use planner_types::ir::scalar::{ColumnRef, CompareOpKind}; use planner_types::ir::schema::{ DataType, FamilyCapability, FieldDataType, SketchStatistic, SummaryInputExpr, }; +// `promql_rows` reads `LogicalOperator` and `OperatorNode` through `super::*` +// only in its test-support helpers. +#[cfg_attr(not(any(test, feature = "test-support")), allow(unused_imports))] use planner_types::ir::{ASAPOp, NonASAPOp, Operator as LogicalOperator, OperatorNode, ScalarExpr}; mod logical; use std::{ @@ -35,14 +38,22 @@ fn invalid(message: impl Into) -> Error { pub type Source<'a> = Box + 'a>; pub mod precompute; +// The executor's PromQL row encoding and the timing frontier are internals: +// a deployment runtime feeds samples with `crate::RawSampleSelector` +// and reads the stored outputs from the installable plan. The Planner's own +// tests reach them with the `test-support` feature. +#[cfg(feature = "test-support")] pub mod promql_rows; +#[cfg(not(feature = "test-support"))] +pub(crate) mod promql_rows; pub mod promql_values; mod candidates; +#[cfg(feature = "test-support")] +pub use candidates::frontier_from_timing; pub use candidates::{ - compile_candidate, compile_candidates, cut_candidate, enumerate_frontiers, - frontier_from_timing, select_candidate, CandidateCost, CandidateSelection, - CompiledPhysicalPlan, + compile_candidate, compile_candidates, cut_candidate, enumerate_frontiers, select_candidate, + CandidateCost, CandidateSelection, CompiledPhysicalPlan, }; mod compiled; diff --git a/crates/executor/src/physical_planner/promql_rows.rs b/crates/executor/src/physical_planner/promql_rows.rs index 88eeef055..9c1ba9aad 100644 --- a/crates/executor/src/physical_planner/promql_rows.rs +++ b/crates/executor/src/physical_planner/promql_rows.rs @@ -1,5 +1,10 @@ //! A bounded PromQL source row carries the entire label set, not just labels //! mentioned by the query. The source adapter owns this lossless encoding. +// Some helpers here serve only the Planner's own tests (`test-support`). +#![cfg_attr( + not(any(test, feature = "test-support")), + allow(dead_code, unused_imports) +)] use super::*; use planner_types::ir::physical_export::{ compile_physical_asap_dag, compile_physical_asap_dag_with_node_ids, diff --git a/crates/executor/src/samples.rs b/crates/executor/src/samples.rs new file mode 100644 index 000000000..72430e5e9 --- /dev/null +++ b/crates/executor/src/samples.rs @@ -0,0 +1,331 @@ +//! Raw samples in, labeled answers out: how a deployment runtime feeds a +//! plan's inputs and reads its PromQL outputs, without the executor's row +//! encoding. +//! +//! A [`RawSampleSelector`] is prepared once per raw-sample window of the +//! installable DAG. It says which metric's samples the window reads +//! ([`RawSampleSelector::metric`]), and turns the samples a runtime +//! received into the window's input rows ([`RawSampleSelector::rows`]): +//! those in the window, kept by the window's `Scan` predicates (its label +//! matchers) and, for an instant selector, only each series' latest one. +//! [`series_labels_of_output_row`] reads the labels of an answer row back. + +use std::collections::BTreeMap; +use std::sync::Arc; + +use planner_types::ir::operator::Source as ScanSource; +use planner_types::ir::physical_export::{PhysicalASAPDAG, RawSampleWindow}; +use planner_types::ir::schema::Schema; +use planner_types::ir::{NonASAPOp, Operator, OperatorNode, TimeRangeKind}; +use std::rc::Rc; + +use crate::expressions::CompiledExpression; +use crate::physical_planner::promql_rows::{ + decode_series_identity, series_row, SERIES_IDENTITY_COLUMN, +}; +use crate::values::{Batch, SchemaRef, Value}; +use crate::Error; + +/// One raw sample, e.g. one Prometheus Remote Write sample. +#[derive(Debug, Clone, PartialEq)] +pub struct Sample { + /// Every label of the series, `__name__` (the metric) included. + pub labels: BTreeMap, + /// When the sample was taken, in Unix milliseconds. + pub timestamp_ms: i64, + /// The sampled value. + pub value: f64, +} + +/// One raw-sample window of an installable DAG, prepared to turn samples +/// into its input rows: built once per window ([`RawSampleSelector::new`]), +/// then called for each window end ([`RawSampleSelector::rows`]). +pub struct RawSampleSelector { + /// The metric the window's `Scan` reads (`__name__`). + metric: String, + /// The `Scan`'s output schema, which the rows are encoded in. + scan_schema: SchemaRef, + /// The `TimeRange`'s output schema, which the rows are returned in. + time_range_schema: SchemaRef, + /// The `Scan`'s predicates (label matchers), compiled over its schema. + predicates: Vec, + /// Width of the window: a range, or an instant selector's lookback. + width_ms: i64, + /// Whether the window is an instant selector, which keeps only each + /// series' latest sample in its lookback. + latest_sample_per_series: bool, +} + +impl RawSampleSelector { + /// The selector of `window` of `dag`. Bad input when the window's nodes + /// are not a time-series `Scan` and a `TimeRange` of `dag` with the same + /// columns, or a predicate does not compile. + pub fn new(dag: &PhysicalASAPDAG, window: &RawSampleWindow) -> Result { + let node = |id| { + dag.nodes + .iter() + .find(|n| n.id == id) + .ok_or_else(|| Error::Invalid(format!("window names unknown node {id}"))) + }; + let (scan, time_range) = (node(window.scan_node_id)?, node(window.time_range_node_id)?); + let Operator::NonASAP(NonASAPOp::Scan { + source: ScanSource::TimeSeries { metric }, + .. + }) = &scan.payload + else { + return Err(Error::Invalid( + "window's scan is not a time-series Scan".into(), + )); + }; + let Operator::NonASAP(NonASAPOp::TimeRange { kind, .. }) = &time_range.payload else { + return Err(Error::Invalid( + "window's time range is not a TimeRange".into(), + )); + }; + if time_range.output_schema.fields != scan.output_schema.fields { + return Err(Error::Invalid( + "window's TimeRange changes its Scan's columns".into(), + )); + } + let scan_schema = Arc::new(scan.output_schema.clone()); + // The Scan's predicates over its own columns. A Scan has no inputs, + // so its predicates name no operator and compile from the payload. + let Operator::NonASAP(NonASAPOp::Scan { predicates, .. }) = &scan.payload else { + unreachable!("matched above"); + }; + let predicates = predicates + .iter() + .map(|predicate| { + if !predicate.0.operator_refs().is_empty() { + return Err(Error::Invalid( + "a Scan predicate reads another operator".into(), + )); + } + let expression = predicate.0.map_operator_refs(&mut |_| -> Rc { + unreachable!("checked: no operator references") + }); + CompiledExpression::compile(&expression, &scan_schema) + }) + .collect::, _>>()?; + Ok(Self { + metric: metric.clone(), + scan_schema, + time_range_schema: Arc::new(time_range.output_schema.clone()), + predicates, + width_ms: window.width_ms, + latest_sample_per_series: *kind == TimeRangeKind::Instant, + }) + } + + /// The metric whose samples the window reads; a runtime routes only + /// those samples to [`Self::rows`]. + pub fn metric(&self) -> &str { + &self.metric + } + + /// The window's input rows for the window ending at `window_end_ms`: + /// the samples of [`Self::metric`] in `(window_end_ms - width, + /// window_end_ms]` that the `Scan`'s predicates keep, and, for an + /// instant selector, only each series' latest one; in the `TimeRange`'s + /// schema, bound to the window's `TimeRange`. `samples` may hold other + /// series and times. Bad input when a sample's labels cannot fill the + /// schema. + pub fn rows(&self, window_end_ms: i64, samples: &[Sample]) -> Result { + let start_ms = window_end_ms - self.width_ms; + let mut kept: Vec<&Sample> = samples + .iter() + .filter(|s| start_ms < s.timestamp_ms && s.timestamp_ms <= window_end_ms) + .filter(|s| s.labels.get("__name__").map(String::as_str) == Some(self.metric())) + .collect(); + if self.latest_sample_per_series { + let mut latest: BTreeMap<&BTreeMap, &Sample> = BTreeMap::new(); + for sample in kept { + let entry = latest.entry(&sample.labels).or_insert(sample); + if sample.timestamp_ms >= entry.timestamp_ms { + *entry = sample; + } + } + kept = latest.into_values().collect(); + } + let mut rows = Vec::with_capacity(kept.len()); + 'samples: for sample in kept { + let row = series_row( + &self.scan_schema, + &sample.labels, + sample.timestamp_ms, + sample.value, + )?; + for predicate in &self.predicates { + match predicate.evaluate(&row)? { + Value::Bool(true) => {} + Value::Bool(false) | Value::Null => continue 'samples, + _ => return Err(Error::Invalid("Scan predicate is not boolean".into())), + } + } + rows.push(row); + } + Batch::try_new(Arc::clone(&self.time_range_schema), rows) + } +} + +/// The labels of the series a PromQL answer row belongs to: its series +/// identity, when the schema keeps one, plus every other non-empty text +/// column (a label the query grouped by). Bad input when the identity is +/// not canonically encoded. +pub fn series_labels_of_output_row( + schema: &Schema, + row: &[Value], +) -> Result, Error> { + let mut labels = BTreeMap::new(); + for (field, cell) in schema.fields.iter().zip(row) { + let Value::Utf8(text) = cell else { continue }; + if field.name == SERIES_IDENTITY_COLUMN { + labels.extend(decode_series_identity(text)?); + } else if !text.is_empty() { + labels.insert(field.name.clone(), text.to_string()); + } + } + Ok(labels) +} + +#[cfg(test)] +mod tests { + use super::*; + use planner_types::ir::operator::Source; + use planner_types::ir::physical_export::compile_physical_asap_dag; + use planner_types::ir::scalar::{CompareOpKind, ScalarValue}; + use planner_types::ir::schema::{DataType, Field, PROMQL_SERIES_IDENTITY}; + use planner_types::ir::{ + apply_materialization_timings, ExprSemantics, MaterializationAssignment, OperatorNode, + Predicate, ScalarExpr, TimingMemo, + }; + use std::rc::Rc; + use std::time::Duration; + + /// `TimeRange(width_ms, kind)` over a `Scan` of `source` keeping + /// `instance="a"`, timed and exported, with its raw-sample window. + fn window_over( + source: Source, + width_ms: u64, + kind: TimeRangeKind, + ) -> (PhysicalASAPDAG, RawSampleWindow) { + let schema = Schema::with_time_index( + vec![ + Field::plain(PROMQL_SERIES_IDENTITY, DataType::Utf8, false), + Field::plain("ts", DataType::Timestamp, false), + Field::plain("value", DataType::Float64, false), + Field::plain("instance", DataType::Utf8, false), + ], + 1, + vec![], + ); + let instance_is_a = Predicate(ScalarExpr::Compare { + left: Box::new(ScalarExpr::Column(3)), + op: CompareOpKind::Eq, + right: Box::new(ScalarExpr::Literal(ScalarValue::Utf8("a".into()))), + semantics: ExprSemantics::Promql, + }); + let scan = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Scan { + source, + predicates: vec![instance_is_a], + schema, + })) + .unwrap(); + let range = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::TimeRange { + range: Duration::from_millis(width_ms), + kind, + child: scan, + })) + .unwrap(); + let timed: Rc = apply_materialization_timings( + &range, + &MaterializationAssignment::default(), + &mut TimingMemo::new(), + ) + .unwrap(); + let dag = compile_physical_asap_dag(&timed).unwrap(); + let window = RawSampleWindow { + time_range_node_id: dag.roots[0], + width_ms: width_ms as i64, + end_offset_ms: 0, + scan_node_id: dag.roots[0] - 1, + }; + (dag, window) + } + + fn sample(metric: &str, instance: &str, timestamp_ms: i64, value: f64) -> Sample { + Sample { + labels: BTreeMap::from([ + ("__name__".to_string(), metric.to_string()), + ("instance".to_string(), instance.to_string()), + ]), + timestamp_ms, + value, + } + } + + /// Each row's timestamp and value. + fn points(batch: &Batch) -> Vec<(i64, f64)> { + batch + .rows() + .iter() + .map(|row| match (&row[1], &row[2]) { + (Value::Timestamp(ts), Value::Float64(v)) => (*ts, *v), + other => panic!("{other:?}"), + }) + .collect() + } + + fn latency_ms() -> Source { + Source::TimeSeries { + metric: "latency_ms".into(), + } + } + + /// A range window `(end - width, end]` excludes a sample at its start + /// and includes one at its end; the Scan's matcher drops series `b` + /// and the metric routing drops another metric. + #[test] + fn a_range_window_keeps_its_bounds_matchers_and_metric() { + let (dag, window) = window_over(latency_ms(), 1_000, TimeRangeKind::Range); + let selector = RawSampleSelector::new(&dag, &window).unwrap(); + assert_eq!(selector.metric(), "latency_ms"); + let samples = [ + sample("latency_ms", "a", 9_000, 1.0), + sample("latency_ms", "a", 9_001, 2.0), + sample("latency_ms", "a", 10_000, 3.0), + sample("latency_ms", "a", 10_001, 4.0), + sample("latency_ms", "b", 9_500, 5.0), + sample("cpu_seconds", "a", 9_500, 6.0), + ]; + let rows = selector.rows(10_000, &samples).unwrap(); + assert_eq!(points(&rows), [(9_001, 2.0), (10_000, 3.0)]); + } + + /// An instant selector keeps each series' latest sample in its lookback. + #[test] + fn an_instant_window_keeps_each_series_latest_sample() { + let (dag, window) = window_over(latency_ms(), 1_000, TimeRangeKind::Instant); + let selector = RawSampleSelector::new(&dag, &window).unwrap(); + let samples = [ + sample("latency_ms", "a", 9_100, 1.0), + sample("latency_ms", "a", 9_900, 10.0), + sample("latency_ms", "a", 9_500, 5.0), + ]; + let rows = selector.rows(10_000, &samples).unwrap(); + assert_eq!(points(&rows), [(9_900, 10.0)]); + } + + /// A window over a table `Scan` has no metric to route samples by. + #[test] + fn a_window_over_a_table_scan_is_refused() { + let table = Source::Table { + table_ref: "metrics".into(), + }; + let (dag, window) = window_over(table, 1_000, TimeRangeKind::Range); + assert!(matches!( + RawSampleSelector::new(&dag, &window), + Err(Error::Invalid(_)) + )); + } +} diff --git a/crates/integration-tests/Cargo.toml b/crates/integration-tests/Cargo.toml index 5787b8aec..f0e5e832a 100644 --- a/crates/integration-tests/Cargo.toml +++ b/crates/integration-tests/Cargo.toml @@ -18,5 +18,5 @@ asap_sketchlib = { workspace = true } serde_json = "1" tokio = { version = "1", features = ["rt", "macros", "rt-multi-thread"] } -asap-executor = { path = "../executor" } +asap-executor = { path = "../executor", features = ["test-support"] } futures = "0.3" From de00ed15c5633f5ce66e971f95ea8850f2fb11b1 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Sun, 18 Oct 2026 22:24:34 +0000 Subject: [PATCH 6/7] fix(plan-selection): retain and price an offset chain's panes for its offset MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stage 3 retained a kept pane chain for its window plus the longest range span of its readers, measured as if its newest pane ended at the evaluation. A chain over an offset input (`max_over_time(m[60m] offset 10m)`) has its newest pane end 10 minutes earlier, so its panes reach back 70 minutes: Selection::pane_retention_ms reported 3,600,000 ms instead of 4,200,000, and the kept memory priced from it was one pane short. The retention now adds the offset of the TimeShifts directly below the chain (end offset + N·w + span); without an offset nothing changes, and every pinned number stays as it was. --- .../tests/range_evaluation.rs | 37 +++++++++++++++ crates/plan-selection/src/lib.rs | 47 +++++++++++++++---- 2 files changed, 76 insertions(+), 8 deletions(-) diff --git a/crates/integration-tests/tests/range_evaluation.rs b/crates/integration-tests/tests/range_evaluation.rs index a161e165f..2ae33b7b8 100644 --- a/crates/integration-tests/tests/range_evaluation.rs +++ b/crates/integration-tests/tests/range_evaluation.rs @@ -291,3 +291,40 @@ fn stage3_plan_selection_prefers_kept_panes_once_the_range_is_declared() { let without_retention = run_promql(&exact_panel(query, 60_000, Some(LAST_HOUR_BY_MINUTE))); assert_eq!(without_retention.selection.selected, ingestion); } + +/// (e) A chain over an offset input reaches back its offset further: with +/// no raw data kept, `max_over_time(m[60m] offset 10m)` every 10 minutes +/// keeps six 10-minute panes ending 10 minutes before each evaluation, so +/// its panes must be kept for 10 min + 60 min = 4,200,000 ms, and the kept +/// memory Stage 3 prices covers that seventh pane; without the offset the +/// retention is the window, 3,600,000 ms. +#[test] +fn an_offset_chain_is_retained_and_priced_for_its_offset() { + let no_raw_data = DeploymentCapabilities { + raw_data_retained: false, + ..DeploymentCapabilities::default() + }; + let plain = run_promql_for( + &exact_panel("max_over_time(m[60m])", 600_000, None), + &no_raw_data, + ); + let offset = run_promql_for( + &exact_panel("max_over_time(m[60m] offset 10m)", 600_000, None), + &no_raw_data, + ); + let retention = + |run: &Run| -> Vec { run.selection.pane_retention_ms.values().copied().collect() }; + assert_eq!(retention(&plain), [3_600_000]); + assert_eq!(retention(&offset), [4_200_000]); + let memory = |run: &Run| { + run.selection.costs[&run.selection.selected] + .usage + .average_memory_bytes + }; + assert!( + memory(&offset) > memory(&plain), + "{} vs {}", + memory(&offset), + memory(&plain) + ); +} diff --git a/crates/plan-selection/src/lib.rs b/crates/plan-selection/src/lib.rs index f7ae1cfe1..a625fc592 100644 --- a/crates/plan-selection/src/lib.rs +++ b/crates/plan-selection/src/lib.rs @@ -88,8 +88,8 @@ //! steady state each request builds its newest pane and every other pane //! any of its steps reads was built by an earlier request or step; //! * a pane chain kept beyond one evaluation (ingestion time or kept -//! across evaluations) retains its window plus the longest -//! `span_ms` among its readers ([`Selection::pane_retention_ms`]), +//! across evaluations) retains its input's offset plus its window plus +//! the longest `span_ms` among its readers ([`Selection::pane_retention_ms`]), //! so the earliest step of the next request still finds its panes: that //! many more panes are kept every second. //! @@ -815,10 +815,12 @@ pub struct Selection { /// one evaluation (maintained at ingestion time, or built at query time /// and kept across evaluations), by node id in its physical DAG: how /// far back from the newest evaluation the runtime must keep its - /// completed panes, in milliseconds. It is the chain's window plus the - /// longest range-evaluation span of its readers, so the earliest step - /// of the next request still finds its panes; just the window without - /// range evaluations. Empty when the plan keeps no panes. + /// completed panes, in milliseconds. It is how far its panes reach + /// back: the offset its newest pane ends before the evaluation (`[60m] + /// offset 10m`: 10 min), plus the chain's window, plus the longest + /// range-evaluation span of its readers, so the earliest step of the + /// next request still finds its panes; without an offset or range + /// evaluations, just the window. Empty when the plan keeps no panes. pub pane_retention_ms: BTreeMap, } @@ -3054,7 +3056,9 @@ fn ingest_workers( /// For each tumbling-pane chain of `dag` kept beyond one evaluation (at /// ingestion time, or at query time across evaluations), how far back /// from the newest evaluation its completed panes must be kept, in -/// milliseconds: its window `pane_count · pane_width_ms`, plus the longest +/// milliseconds: how long before the evaluation its newest pane ends (the +/// offset of its input, [`chain_input_offset_ms`]), plus its window +/// `pane_count · pane_width_ms`, plus the longest /// `span_ms` ([`RangeEvaluation`]) among the queries reaching it, /// whose earliest step reads that much further back. `reach` is /// [`query_reach`] of `dag`; `range_evaluations[q]` is query `q`'s (none @@ -3064,11 +3068,13 @@ fn pane_retention_ms( reach: &[HashSet], range_evaluations: &[Option], ) -> BTreeMap { + let nodes = node_map(dag); dag.nodes .iter() .filter_map(|node| match (&node.payload, node.materialization) { ( Payload::ASAP(ASAPOp::PaneChain { + child, pane_width_ms, pane_count, .. @@ -3084,13 +3090,38 @@ fn pane_retention_ms( .map(|range| range.span_ms) .max() .unwrap_or(0); - Some((node.id, pane_width_ms * u64::from(*pane_count) + span_ms)) + let end_offset_ms = chain_input_offset_ms(*child, &nodes); + Some(( + node.id, + end_offset_ms + pane_width_ms * u64::from(*pane_count) + span_ms, + )) } _ => None, }) .collect() } +/// How long before the evaluation a chain's newest pane ends: the offsets +/// of the `TimeShift`s (without an `@` anchor) directly below the chain, +/// added up; 0 for a negative total. +fn chain_input_offset_ms( + input: PhysicalASAPNodeId, + nodes: &HashMap, +) -> u64 { + let mut offset_ms: i64 = 0; + let mut current = input; + while let Some(Payload::NonASAP(NonASAPOp::TimeShift { shift, child })) = + nodes.get(¤t).map(|node| &node.payload) + { + if shift.at.is_some() { + break; + } + offset_ms = offset_ms.saturating_add(shift.offset_ms); + current = *child; + } + u64::try_from(offset_ms).unwrap_or(0) +} + /// The panes a chain in `retention_ms` ([`pane_retention_ms`]) keeps /// completed: its retention over its pane width, at least its pane count. fn retained_panes( From 729901b5d9069f1301de540956386bec8baa1ae8 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Sun, 18 Oct 2026 21:09:35 +0000 Subject: [PATCH 7/7] feat(planner): plan_installable, the one call a deployment runtime makes A runtime had to link Planner internals to install Stage 3's plan: run the stages, expand pane chains, pick the executor's summary families, find the nodes to store (frontier_from_timing) and walk the DAG for each pane's window. plan_installable(UserInput) plans the workload as e2e_plan does and returns an InstallablePlan, the runtime's plan file: - dag: the selected physical ASAP DAG in its versioned document, with Stage 3's timings and materialization and pane chains expanded; - queries: each query's text and the root node that answers it; a standalone scalar query is refused (UnsupportedQueryRoot); - stored_outputs: each node stored across evaluations, with the run that builds it (an ingestion run, or the query run for a result kept across evaluations), a computation id shared by panes that differ only in offset (so a runtime builds each pane once), its pane (a RawSampleWindow), and how long to keep it (Stage 3's pane_retention_ms of its chain, or how far the pane reaches back); - query_time_raw_windows: the other raw-sample windows read at query time; a window that is not a fixed offset from the evaluation time (a subquery or TimeShift above the range, an @ anchor below it) is refused; - selection: the selected candidate id and its cost per second. Unrestricted supported summaries default to the executor's families. InstallablePlan::query_time_reference (validating first) gives the correctness reference for a runtime's tests. The stage export keeps the chains for costing and display. asap_planner re-exports what a runtime needs to build UserInput, so it depends on asap_planner and asap_executor alone; CI checks asap-planner's tests on their own, where the executor links without test-support, so the runtime test uses public APIs only. crates/planner/tests/runtime_install.rs runs #509 Example 4b the way a runtime does, on samples whose values name their pane, plus a label matcher, an instant selector, two queries whose panes build once, and query-time results kept across evaluations; the integration tests check the plan against frontier_from_timing, two queries on one chain, the refused shapes and scalar queries, and the panes' retention under a range evaluation and an offset. --- .github/workflows/rust.yml | 8 + Cargo.lock | 5 + .../tests/runtime_handoff.rs | 326 +++++++- crates/planner/Cargo.toml | 10 +- crates/planner/src/installable.rs | 773 ++++++++++++++++++ crates/planner/src/lib.rs | 37 +- crates/planner/tests/runtime_install.rs | 564 +++++++++++++ 7 files changed, 1714 insertions(+), 9 deletions(-) create mode 100644 crates/planner/src/installable.rs create mode 100644 crates/planner/tests/runtime_install.rs diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 29245afbd..69a1b3f5d 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -62,6 +62,14 @@ jobs: env: RUSTC_WRAPPER: sccache + # A deployment runtime uses only the facade's and the executor's public + # APIs. asap-planner's tests check that, but only when the executor links + # without its `test-support` feature, which a workspace build turns on. + - name: Check the runtime tests against the public APIs alone + run: cargo check -p asap-planner --tests --locked + env: + RUSTC_WRAPPER: sccache + test: if: github.event_name != 'pull_request' || github.event.pull_request.draft == false runs-on: ubuntu-latest diff --git a/Cargo.lock b/Cargo.lock index 2b179651d..f98e21cdd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -453,12 +453,17 @@ dependencies = [ name = "asap-planner" version = "0.1.0" dependencies = [ + "asap-executor", "asap-frontend-metricsql", "asap-frontend-promql", "asap-frontend-sql", "asap-logical-asap-optimization", "asap-plan-selection", "asap-types", + "futures", + "serde", + "serde_json", + "sha2", "thiserror 2.0.18", "tokio", ] diff --git a/crates/integration-tests/tests/runtime_handoff.rs b/crates/integration-tests/tests/runtime_handoff.rs index a35942ee3..fcba018a7 100644 --- a/crates/integration-tests/tests/runtime_handoff.rs +++ b/crates/integration-tests/tests/runtime_handoff.rs @@ -1,10 +1,19 @@ -//! A deployment runtime consumes Stage 3's selected plan through the -//! facade's exported `PhysicalASAPDAG` document alone (issue #749). +//! What `plan_installable` returns for a deployment runtime (issue #749), +//! checked against the Planner's internals; the runtime's own end-to-end +//! run, with public APIs only, is `crates/planner/tests/runtime_install.rs`. +use std::collections::BTreeMap; +use std::sync::Arc; + +use asap_executor::physical_planner::{compile, frontier_from_timing, InputContract}; use asap_logical_asap_optimization::DeploymentCapabilities; -use asap_planner::{e2e_plan, FrontendInput, PlanOutput, PlanningModels, UserInput}; +use asap_planner::{ + e2e_plan, plan_installable, FrontendInput, InstallablePlan, InstallablePlanError, PlanError, + PlanOutput, PlanningModels, UserInput, +}; use asap_types::ir::physical_export::{compile_physical_asap_workload, PhysicalASAPDAG}; use asap_types::ir::properties::ExecutionTiming; +use asap_types::ir::{ASAPOp, NonASAPOp, Operator}; use asap_types::physical::Materialization; use asap_types::workload::PlanningWorkload; @@ -85,3 +94,314 @@ async fn export_carries_the_selected_materialization() { compile_physical_asap_workload(&selection.plan).expect("the selected plan exports") ); } + +/// What a runtime that keeps no raw samples gets from one facade call. +async fn installable(workload: &PlanningWorkload) -> InstallablePlan { + let capabilities = without_raw_data(); + let models = PlanningModels::builtin().with_deployment_capabilities(&capabilities); + plan_installable(UserInput::new( + workload, + FrontendInput::Promql { histograms: None }, + models, + )) + .await + .expect("Example 4b plans") +} + +/// The deployment's raw input contract for each raw `TimeRange` of `dag`: +/// the rows a runtime binds to each range it scans. +fn raw_range_inputs(dag: &PhysicalASAPDAG) -> BTreeMap { + dag.nodes + .iter() + .filter(|n| matches!(n.payload, Operator::NonASAP(NonASAPOp::TimeRange { .. }))) + .map(|n| { + ( + n.id as u64, + InputContract::bounded(Arc::new(n.output_schema.clone())), + ) + }) + .collect() +} + +fn pane_chain_count(dag: &PhysicalASAPDAG) -> usize { + dag.nodes + .iter() + .filter(|n| matches!(n.payload, Operator::ASAP(ASAPOp::PaneChain { .. }))) + .count() +} + +/// The installable plan has the selected pane chain expanded into the six +/// ten-minute panes the executor runs, each stored at ingestion time, and +/// the executor compiles it. The stage export keeps the chain for costing +/// and display. Its one query is answered by the DAG's one root. +#[tokio::test] +async fn installable_plan_has_its_pane_chains_expanded() { + let workload = example4b(); + let installed = installable(&workload).await; + installed.validate().unwrap(); + let plan = &installed; + let dag = &installed.dag.dag; + assert_eq!(pane_chain_count(dag), 0); + assert_eq!(materialized_count(dag), 6, "six ten-minute panes"); + let roots: Vec = dag.roots.iter().map(|&r| r as u64).collect(); + compile(dag, raw_range_inputs(dag), &roots).expect("the executor compiles it"); + assert_eq!(plan.queries.len(), 1); + assert_eq!( + plan.queries[0].query, + "quantile_over_time(0.99, latency_ms[60m])" + ); + assert_eq!(plan.queries[0].root_node_id, dag.roots[0]); + // The stored outputs are the ingestion frontier: the six panes, each + // ten minutes wide, ending 0, 10, ..., 50 minutes before evaluation. + let stored: Vec = plan + .stored_outputs + .iter() + .map(|o| o.node_id as u64) + .collect(); + assert_eq!(stored, frontier_from_timing(dag).unwrap()); + let mut offsets: Vec = plan + .stored_outputs + .iter() + .map(|o| { + assert_eq!(o.pane.width_ms, 600_000); + o.pane.end_offset_ms + }) + .collect(); + offsets.sort(); + assert_eq!(offsets, (0..6).map(|i| i * 600_000).collect::>()); + assert!(plan.query_time_raw_windows.is_empty(), "no raw data kept"); + let report = plan.selection.as_ref().expect("Stage 3 selected"); + assert!(report.cost_per_second.is_some()); + + let staged = plan_for_stages(&workload).await; + assert_eq!( + pane_chain_count(&staged), + 1, + "the stage export keeps the chain" + ); +} + +async fn plan_for_stages(workload: &PlanningWorkload) -> PhysicalASAPDAG { + plan(workload, &without_raw_data()) + .await + .physical_asap_dag() + .unwrap() +} + +/// Each summary build of `dag` (`SummaryAgg` or `PaneChain`) as its JSON +/// payload, sorted: which families with which parameters it builds. +fn summary_builds(dag: &PhysicalASAPDAG) -> Vec { + let mut builds: Vec = dag + .nodes + .iter() + .filter(|n| { + matches!( + n.payload, + Operator::ASAP(ASAPOp::SummaryAgg { .. } | ASAPOp::PaneChain { .. }) + ) + }) + .map(|n| serde_json::to_string(&n.payload).unwrap()) + .collect(); + builds.sort(); + builds +} + +/// The correctness reference of the installable plan: the same summaries +/// with the same parameters, every node at query time and nothing stored, +/// with the same queries bound to its roots. +#[tokio::test] +async fn reference_plan_runs_the_installable_plan_at_query_time() { + let plan = installable(&example4b()).await; + let reference = plan.query_time_reference().expect("re-times"); + reference.validate().unwrap(); + assert_eq!(materialized_count(&reference.dag.dag), 0); + assert!(reference.stored_outputs.is_empty()); + assert_eq!(reference.query_time_raw_windows.len(), 6); + assert!(reference + .dag + .dag + .nodes + .iter() + .all(|n| n.output_state.timing == ExecutionTiming::QueryTime)); + assert_eq!( + summary_builds(&reference.dag.dag), + summary_builds(&plan.dag.dag) + ); + assert_eq!( + reference.queries[0].root_node_id, + reference.dag.dag.roots[0] + ); +} + +/// A repeating PromQL workload over ingesting `latency_ms`: `queries`, each +/// every 10 minutes at rank error 0.01, δ 0.01. +fn repeating(queries: &[&str]) -> PlanningWorkload { + let entries: Vec = queries + .iter() + .map(|q| { + format!( + r#"{{"query": {q:?}, "demand": {{"fixed_interval": 600000}}, + "requirements": {{"accuracy": {{"accuracy_metric": "rank", "max_error": 0.01, + "failure_probability": 0.01, "accuracy_scope": "whole_result"}}, + "max_latency_ms": 200.0}}, + "predictability": {{"predictable": {{"known_at": null}}}}}}"# + ) + }) + .collect(); + serde_json::from_str(&format!( + r#"{{"query_workload": {{"language": "promql", "repeating_queries": [{}]}}, + "data_workload": {{"arrival": {{"ingesting": {{"rows_per_second": 1000.0}}}}, + "sources": {{"latency_ms": {{"series_count": 1000, "sample_interval": 1000}}}}}}}}"#, + entries.join(",") + )) + .unwrap() +} + +/// p99 and p50 over the same 60 minutes share one chain of panes: the +/// installable plan stores six panes, not one set per query. +#[tokio::test] +async fn two_queries_on_one_chain_store_its_six_panes_once() { + let plan = installable(&repeating(&[ + "quantile_over_time(0.99, latency_ms[60m])", + "quantile_over_time(0.5, latency_ms[60m])", + ])) + .await; + assert_eq!(plan.stored_outputs.len(), 6); + assert_eq!(plan.queries.len(), 2); +} + +/// Planned with raw data kept (everything at query time), so each query's +/// range is a query-time raw-sample window. +async fn installable_at_query_time(query: &str) -> Result { + let workload = repeating(&[query]); + plan_installable(UserInput::new( + &workload, + FrontendInput::Promql { histograms: None }, + PlanningModels::builtin(), + )) + .await +} + +/// A raw-sample window whose position is not a fixed offset from the +/// evaluation time (a subquery or an offset above the range, an `@` +/// anchor below it), or that ends after the evaluation (a negative offset, +/// which no ingestion run can build in time), is refused with its reason, +/// not passed on with a wrong window or dropped; a plain range is listed. +#[tokio::test] +async fn raw_sample_windows_of_unsupported_shapes_are_refused() { + for (query, reason) in [ + ( + "max_over_time(rate(latency_ms[5m])[1h:1m])", + "a subquery or a TimeShift above the range", + ), + ( + "max_over_time(latency_ms[10m] @ 1000)", + "an @ anchor below the range", + ), + ( + "max_over_time(latency_ms[60m] offset -10m)", + "a negative offset below the range", + ), + ] { + match installable_at_query_time(query).await { + Err(PlanError::Installable(InstallablePlanError::UnsupportedRawSampleWindow { + reason: refused, + .. + })) => assert_eq!(refused, reason, "{query}"), + other => panic!("{query}: {:?}", other.map(|p| p.query_time_raw_windows)), + } + } + let plain = installable_at_query_time("max_over_time(latency_ms[10m] offset 5m)") + .await + .unwrap(); + let [window] = plain.query_time_raw_windows.as_slice() else { + panic!("{:?}", plain.query_time_raw_windows); + }; + assert_eq!((window.width_ms, window.end_offset_ms), (600_000, 300_000)); +} + +/// Example 4b answered as the last hour at 10-minute steps (a range +/// evaluation): each of the six panes is kept for the hour its window reads +/// plus the hour the range spans, 7,200,000 ms, as Stage 3 records for the +/// chain (`Selection::pane_retention_ms`); without the range, for the hour. +#[tokio::test] +async fn stored_panes_carry_stage3_pane_retention() { + let mut workload = example4b(); + let plain = installable(&workload).await; + assert!(plain + .stored_outputs + .iter() + .all(|o| o.retention_ms == 3_600_000)); + let mut json = serde_json::to_value(&workload).unwrap(); + json["query_workload"]["repeating_queries"][0]["range_evaluation"] = + serde_json::json!({"step_ms": 600000, "span_ms": 3600000}); + workload = serde_json::from_value(json).unwrap(); + let ranged = installable(&workload).await; + assert_eq!(ranged.stored_outputs.len(), 6); + assert!(ranged + .stored_outputs + .iter() + .all(|o| o.retention_ms == 7_200_000)); + let selection = plan(&workload, &without_raw_data()) + .await + .selection + .unwrap(); + assert_eq!( + selection + .pane_retention_ms + .values() + .copied() + .collect::>(), + [7_200_000] + ); + // A plan file that keeps a pane for less than it reads is refused. + let mut short = ranged.clone(); + short.stored_outputs[0].retention_ms = 0; + assert!(matches!( + short.validate(), + Err(InstallablePlanError::StoredOutputRetentionTooShort { .. }) + )); +} + +/// A chain over an offset input keeps its panes for its window plus the +/// offset: `[60m] offset 10m` reaches 70 minutes back (4,200,000 ms), the +/// unshifted query's chain 60 minutes. Their panes are one computation, so +/// each pane is kept for the longer of the two on every output. +#[tokio::test] +async fn an_offset_chain_keeps_its_panes_for_its_window_plus_the_offset() { + let plan = installable(&repeating(&[ + "quantile_over_time(0.99, latency_ms[60m])", + "quantile_over_time(0.99, latency_ms[60m] offset 10m)", + ])) + .await; + // The twins' panes share one computation, so every output of it + // carries the longest retention: the offset chain's 70 minutes. + let computations: std::collections::BTreeSet<_> = plan + .stored_outputs + .iter() + .map(|o| &o.computation_id) + .collect(); + assert_eq!(computations.len(), 1); + assert!( + plan.stored_outputs + .iter() + .all(|o| o.retention_ms == 4_200_000), + "{:?}", + plan.stored_outputs + .iter() + .map(|o| o.retention_ms) + .collect::>() + ); +} + +/// A standalone scalar query (`scalar(sum(latency_ms))`) has no DAG node +/// to answer it through the public API, so it is refused with its entry. +#[tokio::test] +async fn a_scalar_query_is_refused() { + match installable_at_query_time("scalar(sum(latency_ms))").await { + Err(PlanError::Installable(InstallablePlanError::UnsupportedQueryRoot { + entry_index: 0, + })) => {} + other => panic!("{:?}", other.map(|p| p.queries)), + } +} diff --git a/crates/planner/Cargo.toml b/crates/planner/Cargo.toml index 9a6c00610..0406ba896 100644 --- a/crates/planner/Cargo.toml +++ b/crates/planner/Cargo.toml @@ -16,10 +16,18 @@ asap-plan-selection = { path = "../plan-selection" } asap-frontend-sql = { path = "../frontend-sql" } asap-frontend-promql = { path = "../frontend-promql" } asap-frontend-metricsql = { path = "../frontend-metricsql" } +asap-logical-asap-optimization = { path = "../logical-asap-optimization" } +# The Planner's own executor: `plan_installable` offers only the summary +# families it runs when the caller leaves them unrestricted. +asap-executor = { path = "../executor" } +serde = { version = "1", features = ["derive"] } +serde_json = "1" +sha2 = "0.11" thiserror = "2" [dev-dependencies] # The SQL frontend plans through DataFusion, which is async; `e2e_plan` is # therefore async and its tests need a runtime. tokio = { version = "1", features = ["rt", "macros", "rt-multi-thread"] } -asap-logical-asap-optimization = { path = "../logical-asap-optimization" } +# The runtime end-to-end test runs the executor as a runtime does. +futures = "0.3" diff --git a/crates/planner/src/installable.rs b/crates/planner/src/installable.rs new file mode 100644 index 000000000..152f90992 --- /dev/null +++ b/crates/planner/src/installable.rs @@ -0,0 +1,773 @@ +//! The plan a deployment runtime installs: one call from a workload and the +//! deployment's inputs to the single selected physical ASAP DAG, ready to +//! execute ([`plan_installable`]). +//! +//! The runtime writes the returned [`InstallablePlan`] as its plan file and +//! executes its DAG with the executor, as given (#509 Assumption 8). It +//! needs three things from it: the DAG to install (`dag`), the outputs to +//! store (`stored_outputs`), and where raw samples go (each stored output's +//! `pane` and `query_time_raw_windows`, fed with +//! `asap_executor::RawSampleSelector`). How the Planner represents the +//! plan internally (e.g. a tumbling-pane chain, which the stage export keeps +//! for costing and display) is not the runtime's concern: the installable +//! DAG has every chain expanded into the panes the executor runs. + +use std::collections::{HashMap, HashSet}; +use std::rc::Rc; + +use asap_logical_asap_optimization::DeploymentCapabilities; +use asap_plan_selection::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, +}; +use asap_types::ir::properties::{ExecutionDataStateError, ExecutionTiming}; +use asap_types::ir::SchemaDerivationError; +use asap_types::ir::{ + apply_materialization_timings_to_roots, ASAPOp, MaterializationAssignment, NonASAPOp, Operator, + OperatorNode, QueryRoot, +}; +use asap_types::physical::Materialization; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; + +use crate::{e2e_plan, PlanError, UserInput}; + +/// The selected plan, ready to install and execute: what +/// [`plan_installable`] returns and a runtime writes as its plan file. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct InstallablePlan { + /// The selected physical ASAP DAG with Stage 3's timings and + /// materialization, every pane chain expanded into its panes, in its + /// versioned envelope; the executor compiles `dag.dag` as is. + pub dag: PhysicalASAPDAGDocument, + /// One per workload query, in `QueryWorkload::entries()` order: its + /// text and the root of the DAG that answers it. + pub queries: Vec, + /// Every node whose output the runtime stores across evaluations (its + /// node's materialization is `MatAtIngestion` or + /// `MatAtQueryKeptAcrossEvaluations`), in node order, with the pane of + /// raw samples each one is built from. + pub stored_outputs: Vec, + /// Every window of raw samples read at query time other than a stored + /// output's pane, in node order: the runtime feeds each its samples + /// when a query evaluates. + pub query_time_raw_windows: Vec, + /// Stage 3's record of the selected plan, for reporting only; `None` for + /// [`InstallablePlan::query_time_reference`] and for a pass + /// that selects without pricing. + pub selection: Option, +} + +/// One workload query and the root of the installable DAG that answers it. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct InstalledQuery { + /// The query text, as the workload gave it. + pub query: String, + /// The node whose output answers it, one of `dag.dag.roots`. + pub root_node_id: PhysicalASAPNodeId, +} + +/// One node of the installable DAG whose output the runtime stores, and the +/// pane of raw samples it is built from. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct StoredOutput { + /// The node whose output is stored. + pub node_id: PhysicalASAPNodeId, + /// Which run builds it: an ingestion run over its pane's window, or + /// the query run of the evaluation that first reads it. + pub built_at: StoredOutputBuild, + /// Hex SHA-256 of what the output computes, independent of node ids + /// and of which pane it reads: stored outputs that differ only in + /// `pane.end_offset_ms` share it, so a runtime builds each pane + /// once; equal across plans of one DAG wire version. + pub computation_id: String, + /// The pane of raw samples it is built from. + pub pane: RawSampleWindow, + /// How far back from the newest evaluation time the runtime keeps this + /// pane once built (a pane whose window starts earlier is dropped): + /// Stage 3's retention of its chain (`Selection::pane_retention_ms`: + /// its input's offset plus its window plus the longest + /// range-evaluation span of its readers); or, for an output Stage 3 + /// records none for, how far back its pane reaches + /// (`pane.end_offset_ms + pane.width_ms`). Every output with the same + /// `computation_id` carries the largest of these among them, since a + /// runtime keeps their panes as one. + pub retention_ms: u64, +} + +/// Which run builds a stored output. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum StoredOutputBuild { + /// Built at ingestion time (`MatAtIngestion`): the runtime runs the + /// node in an ingestion run over its pane's window once the pane's + /// samples have arrived, and stores the result. + IngestionRun, + /// Built at query time and kept across evaluations + /// (`MatAtQueryKeptAcrossEvaluations`): when an evaluation needs the + /// pane and the store lacks it, the runtime runs the node in a query + /// run over its pane's samples, stores the result, and binds it like + /// a stored pane. Its `TimeRange` is fed only through this build, so + /// it is not listed in `query_time_raw_windows`. + QueryRun, +} + +/// What Stage 3 recorded about the plan it selected, for reporting. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct SelectedPlanReport { + /// The selected physical candidate's id, e.g. `P3-m1`. + pub selected_candidate_id: String, + /// Its price per second; `None` under cost model 3, which has no + /// prices, or when Stage 3 recorded no cost for it. + pub cost_per_second: Option, +} + +/// Why an installable plan could not be built or is not consistent. +#[derive(Debug, thiserror::Error)] +#[non_exhaustive] +pub enum InstallablePlanError { + /// Internal bug: the selected plan's timings do not hold, or it does + /// not export; or, for the query-time reference, bad input: the plan + /// cannot run at query time. + #[error("timing: {0}")] + Timing(#[from] ExecutionDataStateError), + /// Internal bug: a pane chain of the selected plan does not expand. + #[error("pane chain expansion: {0}")] + PaneChainExpansion(#[from] SchemaDerivationError), + /// Bad input (a plan file): the DAG is not a valid physical ASAP DAG + /// of this wire version. + #[error("invalid physical ASAP DAG: {0}")] + InvalidDag(#[from] PhysicalASAPDAGValidationError), + /// Bad input (a plan file): `queries` does not bind each query, in + /// order, to the DAG's roots. + #[error("queries do not match the DAG's {roots} roots")] + QueriesDoNotMatchRoots { roots: usize }, + /// Unsatisfiable request: query `entry_index` is a standalone scalar + /// query (e.g. `scalar(sum(m))`), which no node of the DAG answers, so + /// a runtime cannot answer it through the executor's public API. + #[error("query {entry_index} is a standalone scalar query, which is not supported yet")] + UnsupportedQueryRoot { entry_index: usize }, + /// Bad input (a plan file): `stored_outputs` or `query_time_raw_windows` + /// is not what the DAG stores or reads. + #[error("stored_outputs or query_time_raw_windows do not match the DAG")] + WindowsOrStoredOutputsDoNotMatchDag, + /// Bad input (a plan file): stored output `node_id` is kept for less + /// than its pane reaches back, so a later evaluation would miss it. + #[error("stored output {node_id} is kept for less than its pane reaches back")] + StoredOutputRetentionTooShort { node_id: PhysicalASAPNodeId }, + /// Internal bug: Stage 3 records a pane retention for node `node_id` + /// of the selected plan, which is not one of its pane chains. + #[error("pane retention for node {node_id}, which is not a pane chain of the selected plan")] + PaneRetentionOfUnknownChain { node_id: PhysicalASAPNodeId }, + /// Unsatisfiable request: the plan stores node `node_id`, which is not + /// built from one pane of raw samples (`TimeRange <- TimeShift* <- + /// Scan` below single-input operators), so a runtime cannot build it. + #[error("stored output {node_id} is not built from one pane of raw samples")] + StoredOutputWithoutPane { node_id: PhysicalASAPNodeId }, + /// Unsatisfiable request: the plan reads raw samples through + /// `TimeRange` `time_range_node_id` in a shape whose window is not a + /// fixed offset from the evaluation time (`reason`: a subquery or a + /// `TimeShift` above it, or an `@` anchor below it), or that ends after + /// the evaluation (a negative offset below it), which no ingestion run + /// can build in time. Not supported yet. + #[error("raw-sample window of node {time_range_node_id} is unsupported: {reason}")] + UnsupportedRawSampleWindow { + time_range_node_id: PhysicalASAPNodeId, + reason: &'static str, + }, +} + +impl From for InstallablePlanError { + fn from(error: ReferencePlanError) -> Self { + match error { + ReferencePlanError::InvalidDag(e) => Self::InvalidDag(e), + ReferencePlanError::Timing(e) => Self::Timing(e), + } + } +} + +/// Plan `input` for a deployment runtime: the query-language frontends and +/// Stages 1-3 as [`e2e_plan`] runs them, returning the single selected DAG +/// ready to install. Supported summaries the caller leaves unrestricted +/// (`DeploymentCapabilities::supported_summaries` `None`) default to the +/// families the Planner's executor runs; raw-data and query-time-result +/// retention stay the caller's. +pub async fn plan_installable(input: UserInput<'_>) -> Result { + let capabilities = executor_summaries_when_unrestricted(input.models.deployment_capabilities); + let query_texts: Vec = input + .workload + .query_workload + .entries() + .map(|entry| entry.query.0) + .collect(); + let input = UserInput { + workload: input.workload, + frontend_specific: input.frontend_specific, + models: input.models.with_deployment_capabilities(&capabilities), + pass: input.pass, + }; + let output = e2e_plan(input).await?; + let selection = output.selection.as_ref().map(report_of); + let installable = || -> Result { + // Every node keeps the timing Stage 3 gave it. + let timed = apply_materialization_timings_to_roots( + &output.roots(), + &MaterializationAssignment::default(), + )?; + let expanded = pane_retention_of_expanded_panes(&timed, output.selection.as_ref())?; + InstallablePlan::of_dag( + expanded.dag, + query_texts, + selection, + &expanded.retention_ms_by_node, + ) + }; + installable().map_err(PlanError::Installable) +} + +impl InstallablePlan { + /// The plan of `dag`, with what the runtime reads from it derived. + fn of_dag( + dag: PhysicalASAPDAG, + query_texts: Vec, + selection: Option, + retention_ms_by_node: &HashMap, + ) -> Result { + let (stored_outputs, query_time_raw_windows) = + runtime_inputs_of(&dag, retention_ms_by_node)?; + let plan = InstallablePlan { + queries: queries_bound_to_roots(&dag, query_texts)?, + stored_outputs, + query_time_raw_windows, + dag: PhysicalASAPDAGDocument::new(dag), + selection, + }; + // Derived from the exported (validated) DAG, so it validates. + debug_assert!(plan.validate().is_ok()); + Ok(plan) + } + + /// Checks a plan, e.g. one read from a plan file: the DAG is valid and + /// of this wire version, and `queries`, `stored_outputs` and + /// `query_time_raw_windows` are what it binds, stores and reads. + pub fn validate(&self) -> Result<(), InstallablePlanError> { + self.dag.validate()?; + let dag = &self.dag.dag; + if queries_bound_to_roots(dag, self.query_texts())? != self.queries { + return Err(InstallablePlanError::QueriesDoNotMatchRoots { + roots: dag.roots.len(), + }); + } + // Retention comes from Stage 3, not from the DAG: take the plan's + // own, and check only that each pane is kept while it is read. + let retention_ms_by_node = self + .stored_outputs + .iter() + .map(|output| (output.node_id, output.retention_ms)) + .collect(); + if let Some(short) = self + .stored_outputs + .iter() + .find(|output| output.retention_ms < output.pane.reach_ms()) + { + return Err(InstallablePlanError::StoredOutputRetentionTooShort { + node_id: short.node_id, + }); + } + let (stored_outputs, query_time_raw_windows) = + runtime_inputs_of(dag, &retention_ms_by_node)?; + if stored_outputs != self.stored_outputs + || query_time_raw_windows != self.query_time_raw_windows + { + return Err(InstallablePlanError::WindowsOrStoredOutputsDoNotMatchDag); + } + Ok(()) + } + + /// For correctness tests only, never for installing: this plan with + /// nothing materialized, the same operators, summaries and parameters + /// with every node at query time, and the same queries. A runtime's + /// test compares its installed plan's answers with the executor running + /// this one over raw samples. Its `selection` is `None`. Validates this + /// plan first. + pub fn query_time_reference(&self) -> Result { + self.validate()?; + let dag = self.dag.dag.with_every_node_at_query_time()?; + InstallablePlan::of_dag(dag, self.query_texts(), None, &HashMap::new()) + } + + fn query_texts(&self) -> Vec { + self.queries.iter().map(|q| q.query.clone()).collect() + } +} + +/// `declared` with `supported_summaries` set to the executor's families +/// when the caller left them unrestricted (`None`). +fn executor_summaries_when_unrestricted( + declared: &DeploymentCapabilities, +) -> DeploymentCapabilities { + let mut capabilities = declared.clone(); + if capabilities.supported_summaries.is_none() { + capabilities.supported_summaries = Some(asap_executor::supported_summary_families()); + } + capabilities +} + +/// Query `i` of `query_texts` bound to root `i` of `dag`; a standalone +/// scalar query, which no node answers, is refused. +fn queries_bound_to_roots( + dag: &PhysicalASAPDAG, + query_texts: Vec, +) -> Result, InstallablePlanError> { + if let Some((entry_index, _)) = dag.scalar_roots.first() { + return Err(InstallablePlanError::UnsupportedQueryRoot { + entry_index: *entry_index, + }); + } + if query_texts.len() != dag.roots.len() { + return Err(InstallablePlanError::QueriesDoNotMatchRoots { + roots: dag.roots.len(), + }); + } + Ok(query_texts + .into_iter() + .zip(&dag.roots) + .map(|(query, &root_node_id)| InstalledQuery { + query, + root_node_id, + }) + .collect()) +} + +/// The nodes of a validated (acyclic) DAG by id, each node's inputs in edge +/// order, and two facts every raw-sample window check reads, computed in +/// one pass each over the DAG (O(nodes + edges)). +struct IndexedDag<'a> { + nodes: HashMap, + inputs: HashMap>, + /// The nodes whose rows come from raw samples: a `Scan` is reachable + /// below them without passing a `TimeRange`. + reading_raw_samples: HashSet, + /// The nodes with a `TimeShift`, `TimeRange` or `PromqlSubquery` above + /// them, which evaluates them at other times than the evaluation time. + re_timed_from_above: HashSet, +} + +/// Whether `operator` evaluates its input at other times than its own. +fn re_times_its_input(operator: &Operator) -> bool { + matches!( + operator, + Operator::NonASAP( + NonASAPOp::TimeShift { .. } + | NonASAPOp::TimeRange { .. } + | NonASAPOp::PromqlSubquery { .. } + ) + ) +} + +impl<'a> IndexedDag<'a> { + fn new(dag: &'a PhysicalASAPDAG) -> Self { + let nodes: HashMap<_, _> = dag.nodes.iter().map(|n| (n.id, n)).collect(); + let mut inputs: HashMap> = HashMap::new(); + for edge in &dag.edges { + inputs.entry(edge.consumer).or_default().push(edge.producer); + } + // Every node after its inputs: an iterative depth-first postorder. + let mut children_first = Vec::with_capacity(dag.nodes.len()); + let mut visited = HashSet::new(); + for node in &dag.nodes { + let mut pending = vec![(node.id, false)]; + while let Some((id, inputs_done)) = pending.pop() { + if inputs_done { + children_first.push(id); + } else if visited.insert(id) { + pending.push((id, true)); + for &input in inputs.get(&id).into_iter().flatten() { + pending.push((input, false)); + } + } + } + } + let payload = |id: &PhysicalASAPNodeId| &nodes[id].payload; + let mut reading_raw_samples = HashSet::new(); + for id in &children_first { + let reads = inputs + .get(id) + .into_iter() + .flatten() + .any(|input| match payload(input) { + Operator::NonASAP(NonASAPOp::Scan { .. }) => true, + Operator::NonASAP(NonASAPOp::TimeRange { .. }) => false, + _ => reading_raw_samples.contains(input), + }); + if reads { + reading_raw_samples.insert(*id); + } + } + let mut re_timed_from_above = HashSet::new(); + for id in children_first.iter().rev() { + if re_timed_from_above.contains(id) || re_times_its_input(payload(id)) { + re_timed_from_above.extend(inputs.get(id).into_iter().flatten().copied()); + } + } + IndexedDag { + nodes, + inputs, + reading_raw_samples, + re_timed_from_above, + } + } + + fn relational(&self, id: PhysicalASAPNodeId) -> Option<&'a NonASAPOp> { + match &self.nodes.get(&id)?.payload { + Operator::NonASAP(op) => Some(op), + Operator::ASAP(_) => None, + } + } + + fn only_input(&self, id: PhysicalASAPNodeId) -> Option { + match self.inputs.get(&id).map(Vec::as_slice) { + Some([input]) => Some(*input), + _ => None, + } + } + + /// The raw-sample window read through the `TimeRange` `time_range`, or + /// why its shape is unsupported. + fn window_at( + &self, + time_range: PhysicalASAPNodeId, + ) -> Result { + let unsupported = |reason| InstallablePlanError::UnsupportedRawSampleWindow { + time_range_node_id: time_range, + reason, + }; + let Some(NonASAPOp::TimeRange { range, .. }) = self.relational(time_range) else { + return Err(unsupported("not a TimeRange")); + }; + if self.re_timed_from_above.contains(&time_range) { + return Err(unsupported("a subquery or a TimeShift above the range")); + } + let width_ms = + i64::try_from(range.as_millis()).map_err(|_| unsupported("window width overflows"))?; + let mut end_offset_ms = 0i64; + let mut current = time_range; + loop { + current = self.only_input(current).ok_or(unsupported( + "an operator with several inputs below the range", + ))?; + match self.relational(current) { + Some(NonASAPOp::TimeShift { shift, .. }) if shift.at.is_some() => { + return Err(unsupported("an @ anchor below the range")) + } + Some(NonASAPOp::TimeShift { shift, .. }) => { + end_offset_ms = end_offset_ms + .checked_add(shift.offset_ms) + .ok_or(unsupported("window offset overflows"))?; + } + Some(NonASAPOp::Scan { .. }) if end_offset_ms < 0 => { + return Err(unsupported("a negative offset below the range")) + } + Some(NonASAPOp::Scan { .. }) => { + return Ok(RawSampleWindow { + time_range_node_id: time_range, + width_ms, + end_offset_ms, + scan_node_id: current, + }) + } + _ => { + return Err(unsupported( + "an operator other than TimeShift below the range", + )) + } + } + } + } + + /// The pane stored output `node` is built from: follow single inputs + /// down to a `TimeRange`, then [`Self::window_at`] it. + fn pane_of(&self, node: PhysicalASAPNodeId) -> Result { + let without_pane = || InstallablePlanError::StoredOutputWithoutPane { node_id: node }; + let mut current = node; + loop { + match self.relational(current) { + Some(NonASAPOp::TimeRange { .. }) => return self.window_at(current), + Some(NonASAPOp::TimeShift { .. } | NonASAPOp::Scan { .. }) => { + return Err(without_pane()) + } + _ => current = self.only_input(current).ok_or_else(without_pane)?, + } + } + } +} + +/// Every node of `dag` stored across evaluations with its pane, in node +/// order, and every window of raw samples read at query time, in node +/// order; an error for a stored output without a pane or a raw-sample +/// window of an unsupported shape. +fn runtime_inputs_of( + dag: &PhysicalASAPDAG, + retention_ms_by_node: &HashMap, +) -> Result<(Vec, Vec), InstallablePlanError> { + let indexed = IndexedDag::new(dag); + let mut computation_hashes = HashMap::new(); + let mut stored_outputs = Vec::new(); + // Every raw-sample window with its timing, and the ranges of stored + // outputs' panes, which are fed through those outputs' builds. + let mut raw_windows = Vec::new(); + let mut stored_pane_ranges = HashSet::new(); + for node in &dag.nodes { + if matches!( + node.materialization, + Materialization::MatAtIngestion(_) + | Materialization::MatAtQueryKeptAcrossEvaluations(_) + ) { + let pane = indexed.pane_of(node.id)?; + stored_pane_ranges.insert(pane.time_range_node_id); + stored_outputs.push(StoredOutput { + node_id: node.id, + built_at: match node.materialization { + Materialization::MatAtIngestion(_) => StoredOutputBuild::IngestionRun, + _ => StoredOutputBuild::QueryRun, + }, + retention_ms: retention_ms_by_node + .get(&node.id) + .copied() + .unwrap_or_else(|| pane.reach_ms()), + computation_id: computation_hash(node.id, &indexed, &mut computation_hashes) + .iter() + .map(|b| format!("{b:02x}")) + .collect(), + pane, + }); + } + if matches!(node.payload, Operator::NonASAP(NonASAPOp::TimeRange { .. })) + && indexed.reading_raw_samples.contains(&node.id) + { + // Every raw-sample window must be one the runtime can feed; a + // stored output's pane is fed through that output's build. + raw_windows.push((node.output_state.timing, indexed.window_at(node.id)?)); + } + } + // Outputs of one computation are one pane store: each keeps its panes + // for the longest retention any of them needs. + let mut longest_retention_ms: HashMap = HashMap::new(); + for output in &stored_outputs { + let longest = longest_retention_ms + .entry(output.computation_id.clone()) + .or_default(); + *longest = (*longest).max(output.retention_ms); + } + for output in &mut stored_outputs { + output.retention_ms = longest_retention_ms[&output.computation_id]; + } + let query_time_raw_windows = raw_windows + .into_iter() + .filter(|(timing, window)| { + *timing == ExecutionTiming::QueryTime + && !stored_pane_ranges.contains(&window.time_range_node_id) + }) + .map(|(_, window)| window) + .collect(); + Ok((stored_outputs, query_time_raw_windows)) +} + +/// SHA-256 over a node's payload (input ids zeroed) and output schema, then +/// its inputs' hashes in edge order. An offset `TimeShift` with one input +/// takes its input's hash: the offset says which pane an output reads, not +/// what it computes. +fn computation_hash( + node: PhysicalASAPNodeId, + dag: &IndexedDag, + done: &mut HashMap, +) -> [u8; 32] { + if let Some(hash) = done.get(&node) { + return *hash; + } + let dag_node = dag.nodes[&node]; + let node_inputs = dag.inputs.get(&node).map(Vec::as_slice).unwrap_or_default(); + let hash = match (&dag_node.payload, node_inputs) { + (Operator::NonASAP(NonASAPOp::TimeShift { shift, .. }), [input]) if shift.at.is_none() => { + computation_hash(*input, dag, done) + } + _ => { + let payload = dag_node.payload.map_children(|_| 0); + let mut hasher = Sha256::new(); + hasher.update( + serde_json::to_vec(&(&payload, &dag_node.output_schema)).expect("serializes"), + ); + for &input in node_inputs { + hasher.update(computation_hash(input, dag, done)); + } + hasher.finalize().into() + } + }; + done.insert(node, hash); + hash +} + +/// The installable DAG of the timed selected plan, with each expanded +/// pane's retention by node id. +struct ExpandedWithRetention { + dag: PhysicalASAPDAG, + retention_ms_by_node: HashMap, +} + +/// Expand the pane chains of `timed` and export it; each pane takes the +/// retention Stage 3 gives its chain (`Selection::pane_retention_ms`, by +/// node id in the selected candidate's physical DAG, which is `timed` +/// exported). +fn pane_retention_of_expanded_panes( + timed: &[QueryRoot], + selection: Option<&Selection>, +) -> Result { + let selected = compile_physical_asap_workload_with_node_ids(timed)?; + let mut retention_by_chain: Vec<(Rc, u64)> = Vec::new(); + for (&node_id, &retention_ms) in selection.iter().flat_map(|s| &s.pane_retention_ms) { + let chain = selected + .node_ids + .operator_node(node_id) + .filter(|node| matches!(node.asap(), Some(ASAPOp::PaneChain { .. }))) + .ok_or(InstallablePlanError::PaneRetentionOfUnknownChain { node_id })?; + retention_by_chain.push((Rc::clone(chain), retention_ms)); + } + let expansion = expand_pane_chains_in_workload(timed)?; + let compiled = compile_physical_asap_workload_with_node_ids(&expansion.roots)?; + let mut retention_ms_by_node = HashMap::new(); + for expanded in &expansion.chains { + let Some((_, retention_ms)) = retention_by_chain + .iter() + .find(|(chain, _)| Rc::ptr_eq(chain, &expanded.chain)) + else { + continue; + }; + for pane in &expanded.panes { + if let Some(node_id) = compiled.node_ids.node_id(pane) { + retention_ms_by_node.insert(node_id, *retention_ms); + } + } + } + Ok(ExpandedWithRetention { + dag: compiled.dag, + retention_ms_by_node, + }) +} + +/// Stage 3's record of its selected candidate. +fn report_of(selection: &Selection) -> SelectedPlanReport { + SelectedPlanReport { + selected_candidate_id: selection.selected.clone(), + cost_per_second: selection + .costs + .get(&selection.selected) + .and_then(|cost| cost.cost_per_second), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use asap_types::ir::schema::{SketchAlgorithm, StatisticKind, SupportedSummary}; + + /// `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 + /// the shift the window is listed. + #[test] + fn a_time_shift_above_the_range_is_refused() { + use asap_types::ir::operator::operator_properties::TimeShift; + use asap_types::ir::operator::Source; + use asap_types::ir::physical_export::compile_physical_asap_dag; + use asap_types::ir::schema::{DataType, Field, Schema}; + use asap_types::ir::{ + apply_materialization_timings, OperatorNode, TimeRangeKind, TimingMemo, + }; + use std::time::Duration; + let scan = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Scan { + source: Source::TimeSeries { + metric: "latency_ms".into(), + }, + predicates: vec![], + schema: Schema::with_time_index( + vec![ + Field::plain("ts", DataType::Timestamp, false), + Field::plain("value", DataType::Float64, false), + ], + 0, + vec![], + ), + })) + .unwrap(); + let range = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::TimeRange { + range: Duration::from_secs(300), + kind: TimeRangeKind::Range, + child: scan, + })) + .unwrap(); + let shifted = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::TimeShift { + shift: TimeShift { + offset_ms: 600_000, + at: None, + }, + child: std::rc::Rc::clone(&range), + })) + .unwrap(); + let export = |root| { + let timed = apply_materialization_timings( + &root, + &MaterializationAssignment::default(), + &mut TimingMemo::new(), + ) + .unwrap(); + compile_physical_asap_dag(&timed).unwrap() + }; + match runtime_inputs_of(&export(shifted), &HashMap::new()) { + Err(InstallablePlanError::UnsupportedRawSampleWindow { reason, .. }) => { + assert_eq!(reason, "a subquery or a TimeShift above the range") + } + other => panic!("{other:?}"), + } + let (stored, windows) = runtime_inputs_of(&export(range), &HashMap::new()).unwrap(); + assert!(stored.is_empty()); + assert_eq!( + windows + .iter() + .map(|w| (w.width_ms, w.end_offset_ms)) + .collect::>(), + [(300_000, 0)] + ); + } + + /// Unrestricted summaries become the executor's families; a caller's + /// own list and its retention flags are kept. + #[test] + fn unrestricted_summaries_default_to_the_executors_families() { + let unrestricted = DeploymentCapabilities { + raw_data_retained: false, + ..DeploymentCapabilities::default() + }; + let defaulted = executor_summaries_when_unrestricted(&unrestricted); + assert_eq!( + defaulted.supported_summaries, + Some(asap_executor::supported_summary_families()) + ); + assert!(!defaulted.raw_data_retained); + assert!(!defaulted.supports(&SketchAlgorithm::Theta, StatisticKind::Distinct)); + let kll_only = DeploymentCapabilities { + supported_summaries: Some(vec![SupportedSummary { + family: SketchAlgorithm::Kll, + readouts: vec![StatisticKind::Quantile], + }]), + ..DeploymentCapabilities::default() + }; + assert_eq!(executor_summaries_when_unrestricted(&kll_only), kll_only); + } +} diff --git a/crates/planner/src/lib.rs b/crates/planner/src/lib.rs index 7d4a683cf..57f3b85e3 100644 --- a/crates/planner/src/lib.rs +++ b/crates/planner/src/lib.rs @@ -10,20 +10,21 @@ //! 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 +//! [`e2e_plan`] runs both steps. [`plan_installable`] runs them for a +//! deployment runtime and returns the single selected DAG ready to install +//! ([`InstallablePlan`]). A caller who already holds pre-ASAP IR — a //! new frontend, a deserialized plan, a test that does not want to build SQL //! and a catalog — skips this crate and calls //! [`asap_plan_selection::optimize`] directly. use asap_types::workload::parsed_workload::{ParsedWorkload, ParsedWorkloadError}; -use asap_types::workload::{PlanningWorkload, QueryLanguage, SqlDialect, WorkloadError}; +use asap_types::workload::{QueryLanguage, SqlDialect, WorkloadError}; use asap_frontend_metricsql::{lower_metricsql_query_workload, MetricsqlError}; use asap_frontend_promql::{ - lower_promql_query_workload, lower_promql_query_workload_with_histograms, HistogramCatalog, - PromqlError, + lower_promql_query_workload, lower_promql_query_workload_with_histograms, PromqlError, }; -use asap_frontend_sql::{lower_sql_dialect, SqlCatalog, SqlError}; +use asap_frontend_sql::{lower_sql_dialect, SqlError}; // The optimization stage's vocabulary is this facade's vocabulary too: a caller // configures the same models and reads the same output whether it goes through @@ -33,6 +34,27 @@ pub use asap_plan_selection::pass::{ PassRegistry, PlanOutput, PlanningModels, QueryPlan, Stage1ToStage3Pass, Stage1ToStage3Run, }; +mod installable; +pub use installable::{ + plan_installable, InstallablePlan, InstallablePlanError, InstalledQuery, SelectedPlanReport, + StoredOutput, StoredOutputBuild, +}; + +// What a deployment runtime needs to build a `UserInput` and read an +// `InstallablePlan`, so it depends on this crate and `asap-executor` alone. +pub use asap_frontend_promql::{HistogramCatalog, HistogramKind}; +pub use asap_frontend_sql::SqlCatalog; +pub use asap_logical_asap_optimization::{DeploymentCapabilities, SupportedSummary}; +pub use asap_plan_selection::{ + ClusterLimits, ClusterObjective, CostModel, MeasuredOperationCost, PlanSelectionSettings, + ResourceSpeed, ResourceUnitCost, SummaryOperation, +}; +pub use asap_types::ir::physical_export::{ + PhysicalASAPDAG, PhysicalASAPDAGDocument, PhysicalASAPNodeId, RawSampleWindow, +}; +pub use asap_types::ir::schema::{SketchAlgorithm, SketchParams, StatisticKind}; +pub use asap_types::workload::PlanningWorkload; + // ── Input ──────────────────────────────────────────────────────────────── /// The lowering dependencies of one frontend. Which variant applies is fixed by @@ -154,6 +176,11 @@ pub enum PlanError { Parsed(#[from] ParsedWorkloadError), #[error(transparent)] Optimize(OptimizeError), + /// The selected plan cannot be installed ([`plan_installable`]): an + /// unsupported raw-sample window or stored output (unsatisfiable + /// request), or an internal bug; see [`InstallablePlanError`]. + #[error("installable plan: {0}")] + Installable(InstallablePlanError), } // ── Entry point ────────────────────────────────────────────────────────── diff --git a/crates/planner/tests/runtime_install.rs b/crates/planner/tests/runtime_install.rs new file mode 100644 index 000000000..fe248dfd5 --- /dev/null +++ b/crates/planner/tests/runtime_install.rs @@ -0,0 +1,564 @@ +//! A deployment runtime installs and runs the Planner's plan with public +//! APIs only: `asap_planner` to plan, `asap_executor` to execute (issue +//! #749). The executor links here without its `test-support` feature, so +//! the compiler checks that no internal is used (CI runs +//! `cargo check -p asap-planner --tests --locked` on its own for that). + +use std::collections::{BTreeMap, BTreeSet}; + +use asap_executor::operators::Operator as ExecutorOperator; +use asap_executor::physical_planner::{compile, InputContract, Source}; +use asap_executor::runtime::{Limits, RunContext, Scope}; +use asap_executor::values::{Batch, Value}; +use asap_executor::{series_labels_of_output_row, RawSampleSelector, Sample}; +use asap_planner::{ + plan_installable, DeploymentCapabilities, FrontendInput, InstallablePlan, InstallablePlanError, + PhysicalASAPDAG, PlanSelectionSettings, PlanningModels, PlanningWorkload, StoredOutputBuild, + UserInput, +}; +use futures::{executor::block_on, StreamExt}; + +/// Example 4b's evaluation time: the end of the 60-minute window. +const EVALUATION_MS: i64 = 3_600_000; + +/// Repeating PromQL `queries`, each every 10 minutes at rank error 0.01 and +/// δ 0.01, over 1,000 ingesting `latency_ms` series sampled every second: +/// #509 Example 4b when the one query is its p99. +fn workload(queries: &[&str]) -> PlanningWorkload { + let entries: Vec = queries + .iter() + .map(|query| { + format!( + r#"{{"query": {query:?}, "demand": {{"fixed_interval": 600000}}, + "requirements": {{"accuracy": {{"accuracy_metric": "rank", "max_error": 0.01, + "failure_probability": 0.01, "accuracy_scope": "whole_result"}}, + "max_latency_ms": 200.0}}, + "predictability": {{"predictable": {{"known_at": null}}}}}}"# + ) + }) + .collect(); + serde_json::from_str(&format!( + r#"{{"query_workload": {{"language": "promql", "repeating_queries": [{}]}}, + "data_workload": {{"arrival": {{"ingesting": {{"rows_per_second": 1000.0}}}}, + "sources": {{"latency_ms": {{"series_count": 1000, "sample_interval": 1000}}}}}}}}"#, + entries.join(",") + )) + .unwrap() +} + +/// One sample of `metric` for series `instance`. +fn sample(metric: &str, instance: &str, timestamp_ms: i64, value: f64) -> Sample { + Sample { + labels: BTreeMap::from([ + ("__name__".to_string(), metric.to_string()), + ("instance".to_string(), instance.to_string()), + ]), + timestamp_ms, + value, + } +} + +/// Raw `latency_ms` samples of series `a` and `b`, one every 10 s from 70 +/// minutes before `EVALUATION_MS` to a minute after it (for a second +/// evaluation), off the pane boundaries, and one +/// sample of another metric the plan never reads. A sample's value says +/// which 10-minute pane before `EVALUATION_MS` it falls in: `1000 × pane + +/// base + 0..=100`, with base 100 for `a` and 300 for `b` (pane 0 is the +/// newest), so an answer read from the wrong panes is visibly off. +fn remote_write_samples() -> Vec { + let mut samples = Vec::new(); + for (instance, base) in [("a", 100.0), ("b", 300.0)] { + for step in 0..426 { + let timestamp_ms = EVALUATION_MS - 4_195_000 + step * 10_000; + let pane = (EVALUATION_MS - timestamp_ms) / 600_000; + samples.push(sample( + "latency_ms", + instance, + timestamp_ms, + (1000 * pane) as f64 + base + (step * 37 % 101) as f64, + )); + } + } + samples.push(sample("cpu_seconds", "a", 5_000, 1e9)); + samples +} + +/// Compile `ids` of `dag` over `inputs` and run them in `scope`, as a +/// runtime does with the executor: each id's output batches. +fn run( + dag: &PhysicalASAPDAG, + inputs: BTreeMap, + ids: &[u64], + scope: Scope, +) -> Vec> { + // What each bound input holds: its rows' schema. + let contracts = inputs + .iter() + .map(|(&id, batch)| (id, InputContract::bounded(batch.schema().clone()))) + .collect(); + // The executor's plan for `ids`, cut at the bound inputs. + let compiled = compile(dag, contracts, ids).expect("the executor compiles the plan"); + // Each bound input as an in-memory source of its rows. + let sources = inputs + .into_iter() + .map(|(id, batch)| { + let source = ExecutorOperator::source(batch.schema().clone(), vec![batch]).unwrap(); + (id, Box::new(source) as Source<'_>) + }) + .collect(); + // The plan bound to its sources, ready to run. + let instance = compiled.instantiate(sources).unwrap(); + block_on(async { + // The run's scope (ingestion window or query time) and limits. + let context = RunContext::new(scope, Limits::default()).unwrap(); + // One output stream per id. + let outputs = instance.execute(compiled.roots(), context).unwrap(); + futures::future::join_all(outputs.into_iter().map(|mut output| async move { + // Its batches, collected. + let mut batches = Vec::new(); + while let Some(batch) = output.next().await { + batches.push((*batch.unwrap()).clone()); + } + batches + })) + .await + }) +} + +/// One answer row: its series labels and its value. +type Answer = (BTreeMap, f64); + +/// Each answer row as its series labels and value, sorted by labels. +fn answers(batches: &[Batch]) -> Vec { + let mut answers: Vec<_> = batches + .iter() + .flat_map(|batch| { + batch.rows().iter().map(|row| { + let labels = series_labels_of_output_row(batch.schema(), row).unwrap(); + let value = row.iter().find_map(|v| match v { + Value::Float64(v) => Some(*v), + _ => None, + }); + (labels, value.expect("a Float64 answer")) + }) + }) + .collect(); + answers.sort_by(|a, b| a.0.cmp(&b.0)); + answers +} + +/// Each answer is a p99 read from pane `pane` of its series +/// ([`remote_write_samples`]): within `1000 × pane + base + 0..=100`. +fn assert_answers_from_pane(answers: &[Answer], pane: i64) { + for (labels, value) in answers { + let base = if labels["instance"] == "a" { + 100.0 + } else { + 300.0 + }; + let low = (1000 * pane) as f64 + base; + assert!( + (low..=low + 100.0).contains(value), + "{labels:?}: {value} is not from pane {pane}" + ); + } +} + +/// The control plane: plan `queries` for a deployment that keeps no raw +/// samples, with its cost model read from JSON, and write the plan file. +async fn plan_json(queries: &[&str]) -> String { + // What the deployment keeps: no raw samples, no query-time results + // across evaluations. Supported summaries are left unrestricted, so the + // Planner offers only what its own executor runs. + let capabilities = DeploymentCapabilities { + raw_data_retained: false, + ..DeploymentCapabilities::default() + }; + plan_json_for(workload(queries), &capabilities).await +} + +/// [`plan_json`] for `workload` on a deployment with `capabilities`. +async fn plan_json_for( + workload: PlanningWorkload, + capabilities: &DeploymentCapabilities, +) -> String { + // The deployment's Stage 3 prices, from its JSON file: cost model 1. + let settings: PlanSelectionSettings = serde_json::from_str( + r#"{"cost_model": {"usage_over_time": { + "cost_per_cpu_op": 1e-6, "cost_per_scanned_byte": 1e-7, + "cost_per_memory_byte_second": 1.25e-7, + "cost_per_provisioned_memory_byte_second": 1.25e-7, + "cost_per_provisioned_cpu_core_second": 1000.0, + "cost_per_disk_byte_second": 0.0}}}"#, + ) + .unwrap(); + // The models Stages 1-3 read: built-in accuracy, the deployment inputs. + let models = PlanningModels::builtin() + .with_deployment_capabilities(capabilities) + .with_plan_selection(&settings); + // The one Planner call: the selected plan, ready to install. + let plan = plan_installable(UserInput::new( + &workload, + FrontendInput::Promql { histograms: None }, + models, + )) + .await + .unwrap(); + // `plan.json`, as the control plane writes it. + serde_json::to_string(&plan).unwrap() +} + +/// What one data-plane evaluation of a plan file returned. +struct DataPlaneRun { + /// Per query, in workload order, the installed plan's answers. + answers_per_query: Vec>, + /// Per query, the query-time reference's answers. + reference_answers_per_query: Vec>, + /// How many panes this evaluation built (ingestion or query runs); + /// the others it read came from the store. + panes_built: usize, +} + +/// One pane the data plane stores: its built state, and how long to keep it. +struct StoredPane { + /// The built state, bound at each stored output of its computation. + state: Batch, + /// When the pane's window starts, in Unix milliseconds. + start_ms: i64, + /// How far back from the newest evaluation the pane is kept. + retention_ms: u64, +} + +/// The data plane's pane store, by computation and pane end: stored +/// outputs with the same key are the same state, across queries and +/// evaluations. +type PaneStore = BTreeMap<(String, i64), StoredPane>; + +/// The node that answers each query of `plan`, in workload order. +fn query_roots(plan: &InstallablePlan) -> Vec { + plan.queries + .iter() + .map(|query| query.root_node_id as u64) + .collect() +} + +/// The data plane at evaluation time `evaluation_ms`: from `plan_json`, +/// `samples` and the panes earlier evaluations left in `store`, build each +/// distinct stored pane the store lacks once, answer every query, drop the +/// panes past their retention, and answer the queries again with the +/// query-time reference. +fn run_data_plane( + plan_json: &str, + samples: &[Sample], + evaluation_ms: i64, + store: &mut PaneStore, +) -> DataPlaneRun { + // The installed plan, read back from its file and checked. + let installed: InstallablePlan = serde_json::from_str(plan_json).unwrap(); + installed.validate().unwrap(); + // The DAG the executor runs. + let dag = &installed.dag.dag; + // Every input of the query run, keyed by the node it is bound to. + let mut inputs = BTreeMap::new(); + // How many panes this evaluation builds. + let mut panes_built = 0; + for output in &installed.stored_outputs { + // The pane's window at this evaluation, `(start_ms, end_ms]`. + let (start_ms, end_ms) = output.pane.bounds_ms(evaluation_ms); + // The pane's identity: its computation and where it ends. + let pane_key = (output.computation_id.clone(), end_ms); + if !store.contains_key(&pane_key) { + // Prepared once per window: routes by metric, applies the Scan. + let selector = RawSampleSelector::new(dag, &output.pane).unwrap(); + // Only the samples of the selector's metric reach it. + let routed: Vec = samples + .iter() + .filter(|s| s.labels["__name__"] == selector.metric()) + .cloned() + .collect(); + // The pane's input rows: the samples its Scan selects in its window. + let rows = selector.rows(end_ms, &routed).unwrap(); + // The run that builds the pane: an ingestion run over its + // window, or, for a query-time result kept across evaluations, + // the query run of this evaluation. + let scope = match output.built_at { + StoredOutputBuild::IngestionRun => Scope::Ingestion { + window_start_ms: start_ms, + window_end_ms: end_ms, + revision: 1, + }, + StoredOutputBuild::QueryRun => Scope::Query { + evaluation_time_ms: evaluation_ms, + revision: 1, + }, + }; + // The rows bound to the pane's TimeRange. + let pane_input = BTreeMap::from([(output.pane.time_range_node_id as u64, rows)]); + // The batches that run built. + let built = run(dag, pane_input, &[output.node_id as u64], scope).remove(0); + // The pane's state, as the store keeps it, in one batch. + let schema = built[0].schema().clone(); + let state_rows = built.iter().flat_map(|b| b.rows().to_vec()).collect(); + let state = Batch::try_new(schema, state_rows).unwrap(); + store.insert( + pane_key.clone(), + StoredPane { + state, + start_ms, + retention_ms: output.retention_ms, + }, + ); + panes_built += 1; + } + inputs.insert(output.node_id as u64, store[&pane_key].state.clone()); + } + // Raw samples read at query time, other than stored outputs' panes. + for window in &installed.query_time_raw_windows { + // The window's selector and its end at this evaluation. + let selector = RawSampleSelector::new(dag, window).unwrap(); + let (_, end_ms) = window.bounds_ms(evaluation_ms); + inputs.insert( + window.time_range_node_id as u64, + selector.rows(end_ms, samples).unwrap(), + ); + } + // The query evaluation, which merges the stored panes. + let query = Scope::Query { + evaluation_time_ms: evaluation_ms, + revision: 1, + }; + // Each query's answers: (series labels, p99) per series. + let answers_per_query = run(dag, inputs, &query_roots(&installed), query.clone()) + .iter() + .map(|batches| answers(batches)) + .collect(); + // The store drops each pane whose window starts more than its + // retention before this evaluation. + store.retain(|_, pane| pane.start_ms >= evaluation_ms - pane.retention_ms as i64); + + // The correctness reference: nothing stored, every window fed raw + // samples at query time. + let reference = installed.query_time_reference().unwrap(); + // The reference's DAG. + let reference_dag = &reference.dag.dag; + // Each query-time window's input rows, keyed by its TimeRange. + let raw: BTreeMap = reference + .query_time_raw_windows + .iter() + .map(|window| { + // The window's selector and its end at this evaluation. + let selector = RawSampleSelector::new(reference_dag, window).unwrap(); + let (_, end_ms) = window.bounds_ms(evaluation_ms); + ( + window.time_range_node_id as u64, + selector.rows(end_ms, samples).unwrap(), + ) + }) + .collect(); + // The reference's answers, which the installed plan's must equal. + let reference_answers_per_query = run(reference_dag, raw, &query_roots(&reference), query) + .iter() + .map(|batches| answers(batches)) + .collect(); + DataPlaneRun { + answers_per_query, + reference_answers_per_query, + panes_built, + } +} + +/// #509 Example 4b, as a runtime runs it: one `plan_installable` call, +/// `plan.json`, then each stored pane built in an ingestion run and merged +/// in the query run. One p99 per series, read from the hour's oldest pane, +/// equal to the query-time reference; the other metric is not read. +#[tokio::test] +async fn a_runtime_installs_and_runs_the_plan_from_one_facade_call() { + let plan_json = plan_json(&["quantile_over_time(0.99, latency_ms[60m])"]).await; + let run = run_data_plane( + &plan_json, + &remote_write_samples(), + EVALUATION_MS, + &mut PaneStore::new(), + ); + assert_eq!(run.panes_built, 6, "six ten-minute panes"); + // The p99 of the hour comes from its oldest pane, pane 5. + assert_answers_from_pane(&run.answers_per_query[0], 5); + assert_eq!(run.answers_per_query[0][0].0["instance"], "a"); + assert_eq!(run.answers_per_query, run.reference_answers_per_query); +} + +/// The query's label matcher selects series `a` only: the runtime feeds +/// every sample, and the window's selector applies the plan's `Scan`. +#[tokio::test] +async fn label_matchers_select_the_series_a_pane_reads() { + let plan_json = + plan_json(&[r#"quantile_over_time(0.99, latency_ms{instance="a"}[60m])"#]).await; + let run = run_data_plane( + &plan_json, + &remote_write_samples(), + EVALUATION_MS, + &mut PaneStore::new(), + ); + let answered = &run.answers_per_query[0]; + assert_eq!(answered.len(), 1, "{answered:?}"); + assert_eq!(answered[0].0["instance"], "a"); + assert_answers_from_pane(answered, 5); + assert_eq!(run.answers_per_query, run.reference_answers_per_query); +} + +/// The p99 over the last hour and over the hour before the last 10 +/// minutes store 12 panes of one computation; five panes are the same +/// (offsets 10..50 min), so the data plane builds 7, and both answers equal +/// the reference. +#[tokio::test] +async fn panes_of_one_computation_build_once_across_queries() { + let plan_json = plan_json(&[ + "quantile_over_time(0.99, latency_ms[60m])", + "quantile_over_time(0.99, latency_ms[60m] offset 10m)", + ]) + .await; + let installed: InstallablePlan = serde_json::from_str(&plan_json).unwrap(); + assert_eq!(installed.stored_outputs.len(), 12); + let computations: BTreeSet<&str> = installed + .stored_outputs + .iter() + .map(|o| o.computation_id.as_str()) + .collect(); + assert_eq!(computations.len(), 1); + let run = run_data_plane( + &plan_json, + &remote_write_samples(), + EVALUATION_MS, + &mut PaneStore::new(), + ); + assert_eq!(run.panes_built, 7); + assert_eq!(run.answers_per_query.len(), 2); + // The last hour's p99 comes from pane 5, the offset hour's from pane 6. + assert_answers_from_pane(&run.answers_per_query[0], 5); + assert_answers_from_pane(&run.answers_per_query[1], 6); + assert_eq!(run.answers_per_query, run.reference_answers_per_query); +} + +/// A plan file of another DAG wire version is refused, and so is the +/// reference of a tampered plan file, with an error rather than a panic. +#[tokio::test] +async fn a_plan_file_of_another_version_or_tampered_is_refused() { + let plan_json = plan_json(&["quantile_over_time(0.99, latency_ms[60m])"]).await; + let mut other_version: InstallablePlan = serde_json::from_str(&plan_json).unwrap(); + other_version.dag.schema_version += 1; + assert!(matches!( + other_version.validate(), + Err(InstallablePlanError::InvalidDag(_)) + )); + let mut tampered: InstallablePlan = serde_json::from_str(&plan_json).unwrap(); + tampered.queries.clear(); + assert!(matches!( + tampered.query_time_reference(), + Err(InstallablePlanError::QueriesDoNotMatchRoots { .. }) + )); + let mut missing_node: InstallablePlan = serde_json::from_str(&plan_json).unwrap(); + missing_node.dag.dag.nodes.remove(0); + assert!(matches!( + missing_node.query_time_reference(), + Err(InstallablePlanError::InvalidDag(_)) + )); +} + +/// `sum(latency_ms)` reads an instant selector: per series, only the latest +/// sample in its 1 s lookback counts. Series `a` has samples at T-900 ms (1) +/// and T-100 ms (10), series `b` at T-200 ms (100): the sum is 10 + 100. +#[tokio::test] +async fn an_instant_selector_reads_the_latest_sample_per_series() { + let workload = workload(&["sum(latency_ms)"]); + let plan = plan_installable(UserInput::new( + &workload, + FrontendInput::Promql { histograms: None }, + PlanningModels::builtin(), + )) + .await + .unwrap(); + let dag = &plan.dag.dag; + let [window] = plan.query_time_raw_windows.as_slice() else { + panic!("{:?}", plan.query_time_raw_windows); + }; + assert_eq!(window.width_ms, 1_000); + let samples = [ + sample("latency_ms", "a", EVALUATION_MS - 900, 1.0), + sample("latency_ms", "a", EVALUATION_MS - 100, 10.0), + sample("latency_ms", "b", EVALUATION_MS - 200, 100.0), + ]; + let selector = RawSampleSelector::new(dag, window).unwrap(); + let rows = selector.rows(EVALUATION_MS, &samples).unwrap(); + let query = Scope::Query { + evaluation_time_ms: EVALUATION_MS, + revision: 1, + }; + let inputs = BTreeMap::from([(window.time_range_node_id as u64, rows)]); + let answered = answers(&run(dag, inputs, &query_roots(&plan), query)[0]); + assert_eq!(answered.len(), 1); + assert_eq!(answered[0].1, 110.0); +} + +/// A deployment that keeps raw samples and query-time results across +/// evaluations (#509 Example 4's B3), with a dashboard panel +/// (`max_over_time(latency_ms[5m])` every minute over 1M series, answered +/// as the last hour at 1-minute steps): Stage 3 keeps five 1-minute panes +/// built at query time, and feeds their ranges only through those builds, +/// so no window is listed twice. The first evaluation, with an empty +/// store, builds all five in its query run; the next, a minute later, +/// reuses the four still in its window and builds only the newest. Each +/// evaluation's answers (each series' newest-pane maximum) equal the +/// reference. +#[tokio::test] +async fn query_time_results_kept_across_evaluations_are_built_in_query_runs() { + let capabilities = DeploymentCapabilities { + query_time_results_retained: true, + ..DeploymentCapabilities::default() + }; + let workload: PlanningWorkload = serde_json::from_str( + r#"{"query_workload": {"language": "promql", "repeating_queries": [{ + "query": "max_over_time(latency_ms[5m])", + "demand": {"fixed_interval": 60000}, + "requirements": {}, + "predictability": {"predictable": {"known_at": null}}, + "range_evaluation": {"step_ms": 60000, "span_ms": 3600000}}]}, + "data_workload": {"arrival": {"ingesting": {"rows_per_second": 66666.67}}, + "sources": {"latency_ms": {"series_count": 1000000, "sample_interval": 15000}}}}"#, + ) + .unwrap(); + let plan_json = plan_json_for(workload, &capabilities).await; + let installed: InstallablePlan = serde_json::from_str(&plan_json).unwrap(); + assert_eq!(installed.stored_outputs.len(), 5); + assert!(installed + .stored_outputs + .iter() + .all(|o| o.built_at == StoredOutputBuild::QueryRun && o.pane.width_ms == 60_000)); + let pane_ranges: BTreeSet<_> = installed + .stored_outputs + .iter() + .map(|o| o.pane.time_range_node_id) + .collect(); + assert!(installed + .query_time_raw_windows + .iter() + .all(|w| !pane_ranges.contains(&w.time_range_node_id))); + // The store the data plane keeps across evaluations. + let mut store = PaneStore::new(); + let first = run_data_plane( + &plan_json, + &remote_write_samples(), + EVALUATION_MS, + &mut store, + ); + assert_eq!(first.panes_built, 5); + assert_answers_from_pane(&first.answers_per_query[0], 0); + assert_eq!(first.answers_per_query, first.reference_answers_per_query); + let next = run_data_plane( + &plan_json, + &remote_write_samples(), + EVALUATION_MS + 60_000, + &mut store, + ); + assert_eq!(next.panes_built, 1, "four panes reused from the store"); + assert_eq!(next.answers_per_query, next.reference_answers_per_query); +}