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 2611563e3..f98e21cdd 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", @@ -452,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/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/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 b75bd71e2..91091e79a 100644 --- a/crates/executor/src/lib.rs +++ b/crates/executor/src/lib.rs @@ -13,7 +13,11 @@ 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; /// The exact Planner contract used by these kernels. pub use planner_types as planner; 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/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/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" 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/integration-tests/tests/runtime_handoff.rs b/crates/integration-tests/tests/runtime_handoff.rs new file mode 100644 index 000000000..fcba018a7 --- /dev/null +++ b/crates/integration-tests/tests/runtime_handoff.rs @@ -0,0 +1,407 @@ +//! 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, 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; + +/// #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") + ); +} + +/// 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/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/plan-selection/src/lib.rs b/crates/plan-selection/src/lib.rs index 967a33d4d..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. //! @@ -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`]. @@ -805,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, } @@ -3044,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 @@ -3054,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, .. @@ -3074,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( diff --git a/crates/plan-selection/src/pass/mod.rs b/crates/plan-selection/src/pass/mod.rs index 59d283689..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; @@ -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. @@ -276,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) } @@ -296,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/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()); +} 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); +} 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/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, 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); +}