From 15af6a750224abe038a3e545537e751569b3a157 Mon Sep 17 00:00:00 2001 From: zzylol Date: Sun, 18 Oct 2026 17:14:06 +0000 Subject: [PATCH 1/2] feat(query): keep query-time panes across evaluations A plan node materialized `MatAtQueryKeptAcrossEvaluations` (#509 Example 4's B3) is now executed instead of rejected. Its output is kept in memory by (definition, pane end), with the SDS identity rule, built from its pane's raw samples the first time an evaluation reads it, reused by later evaluations, and evicted like stored panes (older than the largest read offset plus two pane widths, measured from the newest evaluation). The control plane's default DeploymentCapabilities now set query_time_results_retained. examples/planner-layering-example4-b3.json is a workload for which the Planner selects B3 under the default cost model. --- README.md | 16 +- control_plane/src/lib.rs | 8 +- control_plane/tests/plan_workload.rs | 34 ++++ data_plane/src/engine.rs | 2 + data_plane/src/precompute.rs | 165 ++++++++++-------- data_plane/src/query.rs | 170 ++++++++++++++++-- data_plane/src/sds.rs | 105 +++++++---- data_plane/tests/common/mod.rs | 16 +- data_plane/tests/e2e_examples.rs | 54 +++++- data_plane/tests/kept.rs | 194 +++++++++++++++++++++ data_plane/tests/query.rs | 14 +- data_plane/tests/sds.rs | 53 +++++- examples/planner-layering-example4-b3.json | 44 +++++ 13 files changed, 716 insertions(+), 159 deletions(-) create mode 100644 data_plane/tests/kept.rs create mode 100644 examples/planner-layering-example4-b3.json diff --git a/README.md b/README.md index 8d11d1d4..3b87370e 100644 --- a/README.md +++ b/README.md @@ -35,6 +35,7 @@ The backend is being rebuilt on the Planner's physical DAG: | Stored summaries (SDS): one definition per computation, panes by end time | done | | Precompute engine: build ingestion-time outputs once per tumbling pane | done | | Query engine: answer planned instant queries from stored panes and raw samples | done | +| Query engine: keep query-time panes across evaluations (`MatAtQueryKeptAcrossEvaluations`) | done | Panes close `--allowed-lateness-ms` (default 15 s) after the newest sample passes their end; later samples for them are dropped. The first pane after @@ -43,7 +44,11 @@ startup is skipped because it may be partial. An instant query whose text is one of the plan's queries is answered by the plan (`x-asap-execution: plan`). Its stored inputs read the panes ending at `T - offset`, so `T` must be on the pane grid and those panes stored; its -query-time raw ranges are read from Prometheus. Anything else is forwarded +query-time raw ranges are read from Prometheus. An input the plan keeps +across evaluations (`MatAtQueryKeptAcrossEvaluations`) is computed from its +pane's raw samples the first time an evaluation reads it, kept in memory by +definition and pane end, and reused by later evaluations; it is evicted like +a stored pane, measured from the newest evaluation. Anything else is forwarded to Prometheus (`x-asap-execution: forwarded`); a planned query that could not be answered also says why in `x-asap-forward-reason`. Range queries, SQL and ClickHouse are not supported yet. @@ -61,8 +66,7 @@ cargo run -p data_plane -- --plan plan.json --prometheus-url http://localhost:90 `--deployment-capabilities caps.json` replaces the default `DeploymentCapabilities`: every summary the Planner can build, raw data kept -in Prometheus, and no query-time result kept across evaluations -(`MatAtQueryKeptAcrossEvaluations` is not implemented). For example +in Prometheus, and query-time results kept across evaluations. For example `{"raw_data_retained": false}` makes the Planner build every input at ingestion time. `--measured-operation-costs costs.json` gives Stage 3 the deployment's measured sketch operation costs; otherwise it prices every node @@ -70,8 +74,10 @@ with cost model 1 and illustrative unit costs. ## Examples -`examples/` holds #509's PromQL example workloads (1, 3a, 3b, 4a, 4b) as -Planner `PlanningWorkload`s. `data_plane/tests/e2e_examples.rs` plans each +`examples/` holds #509's PromQL example workloads (1, 3a, 3b, 4a, 4b, and +`4-b3`: Example 3b's panel over ten series sampled every millisecond, for +which the Planner keeps query-time panes across evaluations) as Planner +`PlanningWorkload`s. `data_plane/tests/e2e_examples.rs` plans each one, runs the `data_plane` binary against a Prometheus holding generated samples, Remote Writes the same samples, and checks every planned query against the Planner executor running the selected plan with nothing diff --git a/control_plane/src/lib.rs b/control_plane/src/lib.rs index 8cc82602..15e8a4da 100644 --- a/control_plane/src/lib.rs +++ b/control_plane/src/lib.rs @@ -224,14 +224,14 @@ pub fn plan_selected_at_query_time( } /// What this backend's data plane can execute and keep: every summary the -/// Planner executor builds, raw samples kept in Prometheus, and no -/// query-time result kept across evaluations (the data plane does not -/// implement `MatAtQueryKeptAcrossEvaluations`). +/// Planner executor builds, raw samples kept in Prometheus, and query-time +/// results kept across evaluations in memory +/// (`MatAtQueryKeptAcrossEvaluations`). pub fn backend_capabilities() -> DeploymentCapabilities { DeploymentCapabilities { supported_summaries: None, raw_data_retained: true, - query_time_results_retained: false, + query_time_results_retained: true, } } diff --git a/control_plane/tests/plan_workload.rs b/control_plane/tests/plan_workload.rs index 782f9a96..3aae8c61 100644 --- a/control_plane/tests/plan_workload.rs +++ b/control_plane/tests/plan_workload.rs @@ -99,3 +99,37 @@ fn reference_plan_materializes_nothing() { assert_eq!(materialized(&reference), 0); assert_eq!(reference.queries, installed.queries); } + +/// The backend keeps query-time results across evaluations, so for a few +/// series sampled every millisecond the Planner selects #509 Example 4's +/// B3: five 1-minute KLL panes built at query time and kept, installed as +/// plain panes. +#[test] +fn selects_panes_kept_across_evaluations() { + let inputs = DeploymentInputs::default(); + assert!(inputs.deployment_capabilities.query_time_results_retained); + let path = + Path::new(env!("CARGO_MANIFEST_DIR")).join("../examples/planner-layering-example4-b3.json"); + let workload: PlanningWorkload = + serde_json::from_str(&std::fs::read_to_string(path).unwrap()).unwrap(); + let plan = plan_workload(&workload, &inputs).unwrap(); + let kept = plan + .dag + .dag + .nodes + .iter() + .filter(|n| { + matches!( + n.materialization, + Materialization::MatAtQueryKeptAcrossEvaluations(_) + ) + }) + .inspect(|n| { + assert!(matches!( + n.payload, + Operator::ASAP(ASAPOp::SummaryAgg { .. }) + )) + }) + .count(); + assert_eq!(kept, 5, "five one-minute panes"); +} diff --git a/data_plane/src/engine.rs b/data_plane/src/engine.rs index e2e220a5..282798e6 100644 --- a/data_plane/src/engine.rs +++ b/data_plane/src/engine.rs @@ -135,6 +135,8 @@ impl Engine { tracing::info!( definitions = sds.definitions.len(), outputs = sds.outputs.len(), + kept_definitions = sds.kept_definitions.len(), + kept_outputs = sds.kept_outputs.len(), "stored summaries" ); let query = QueryEngine::new(&plan, &sds); diff --git a/data_plane/src/precompute.rs b/data_plane/src/precompute.rs index e0cbc9d1..0231fcd6 100644 --- a/data_plane/src/precompute.rs +++ b/data_plane/src/precompute.rs @@ -29,7 +29,7 @@ use planner_types::ir::operator::operator_properties::Source as ScanSource; use planner_types::ir::physical_export::PhysicalASAPDAG; use planner_types::ir::{NonASAPOp, Operator as IrOperator}; -use crate::sds::{DefinitionId, Sds, SummaryDefinition, SummaryStore}; +use crate::sds::{DefinitionId, PaneBinding, Sds, SummaryDefinition, SummaryStore}; /// One raw sample: its series' full label set (including `__name__`), time /// and value. @@ -69,13 +69,75 @@ pub struct ScanInput { pub schema: SchemaRef, } -struct Builder { - definition: SummaryDefinition, - scan: ScanInput, +/// How the panes of one definition are built from raw samples. The +/// precompute engine builds stored panes with it, and the query engine kept +/// ones. +pub(crate) struct PaneBuild { + pub(crate) definition: SummaryDefinition, + pub(crate) scan: ScanInput, build: CompiledPhysicalDAG, - /// Oldest pane offset any output reads, plus one pane: panes ending - /// before `watermark - retention_ms` are no longer read. - retention_ms: i64, + /// Largest offset any output of the definition reads, plus two panes: + /// panes ending at or before `newest - retention_ms` are no longer read. + pub(crate) retention_ms: i64, +} + +impl PaneBuild { + /// Compiles `definition`'s build from its `TimeRange` up to its builder; + /// `outputs` are the outputs that read it. + pub(crate) fn new( + dag: &PhysicalASAPDAG, + definition: &SummaryDefinition, + outputs: &BTreeMap, + ) -> Result { + let error = + |message: &str| PrecomputeError::Definition(definition.id.clone(), message.into()); + let scan = scan_below(dag, definition.raw_input) + .ok_or_else(|| error("no time-series Scan below its TimeRange"))?; + let raw_schema = Arc::new(output_schema(dag, definition.raw_input).clone()); + if raw_schema != scan.schema { + return Err(error("TimeRange rows differ from its Scan's")); + } + let build = compile( + dag, + BTreeMap::from([(definition.raw_input, InputContract::bounded(raw_schema))]), + &[definition.builder], + )?; + let max_offset = outputs + .values() + .filter(|(id, _)| id == &definition.id) + .map(|(_, binding)| binding.offset_ms) + .max() + .unwrap_or(0); + Ok(Self { + definition: definition.clone(), + scan, + build, + retention_ms: max_offset + 2 * definition.pane_ms, + }) + } + + /// Runs the plan's `Scan` over `samples`, then the build over the rows + /// it selects, under `scope`. + pub(crate) fn run( + &self, + dag: &PhysicalASAPDAG, + samples: &[RawSample], + scope: Scope, + ) -> Result { + let selected = select(dag, &self.scan, samples, scope.clone())?; + let input: Source<'_> = + Box::new(Operator::source(self.scan.schema.clone(), vec![selected])?); + let build = self + .build + .instantiate(BTreeMap::from([(self.definition.raw_input, input)]))?; + let rows = run(&build, self.build.roots(), scope)?; + let schema = Arc::new(self.definition.output_schema.clone()); + Ok(Batch::try_new(schema, rows)?) + } +} + +struct Builder { + pane: PaneBuild, /// Samples of open panes, by pane end. open: BTreeMap>, /// End of the newest pane built or skipped; samples in it or before it @@ -99,37 +161,17 @@ impl Precompute { sds: &Sds, allowed_lateness_ms: i64, ) -> Result { - let mut builders = Vec::new(); - for definition in sds.definitions.values() { - let error = - |message: String| PrecomputeError::Definition(definition.id.clone(), message); - let scan = scan_below(&dag, definition.raw_input) - .ok_or_else(|| error("no time-series Scan below its TimeRange".into()))?; - let raw_schema = Arc::new(output_schema(&dag, definition.raw_input).clone()); - if raw_schema != scan.schema { - return Err(error("TimeRange rows differ from its Scan's".into())); - } - let build = compile( - &dag, - BTreeMap::from([(definition.raw_input, InputContract::bounded(raw_schema))]), - &[definition.builder], - )?; - let max_offset = sds - .outputs - .values() - .filter(|(id, _)| id == &definition.id) - .map(|(_, binding)| binding.offset_ms) - .max() - .unwrap_or(0); - builders.push(Builder { - definition: definition.clone(), - scan, - build, - retention_ms: max_offset + 2 * definition.pane_ms, - open: BTreeMap::new(), - built_through: None, - }); - } + let builders = sds + .definitions + .values() + .map(|definition| { + Ok(Builder { + pane: PaneBuild::new(&dag, definition, &sds.outputs)?, + open: BTreeMap::new(), + built_through: None, + }) + }) + .collect::>()?; Ok(Self { dag, builders, @@ -153,10 +195,10 @@ impl Precompute { let metric = sample.labels.get("__name__").map(String::as_str); let mut accepted = false; for builder in &mut self.builders { - if metric != Some(builder.scan.metric.as_str()) { + if metric != Some(builder.pane.scan.metric.as_str()) { continue; } - let end = pane_end(sample.timestamp_ms, builder.definition.pane_ms); + let end = pane_end(sample.timestamp_ms, builder.pane.definition.pane_ms); // The first pane seen may be partial: skip it. let built = *builder.built_through.get_or_insert(end); if end <= built { @@ -182,27 +224,33 @@ impl Precompute { let Some(built) = builder.built_through else { continue; }; - let pane_ms = builder.definition.pane_ms; + let pane_ms = builder.pane.definition.pane_ms; // Every pane after the first through the watermark is built, so // a pane without samples is stored as empty, not missing. let mut end = built + pane_ms; while end <= watermark { let samples = builder.open.remove(&end).unwrap_or_default(); - match build_pane(&self.dag, builder, end, &samples) { + let scope = Scope::Ingestion { + window_start_ms: end - pane_ms, + window_end_ms: end, + revision: 1, + }; + match builder.pane.run(&self.dag, &samples, scope) { Ok(state) => { - self.store.insert(&builder.definition.id, end, state); + self.store.insert(&builder.pane.definition.id, end, state); self.stats.panes_built += 1; } Err(error) => { - tracing::error!(definition = ?builder.definition.id, end, %error, "pane build failed"); + tracing::error!(definition = ?builder.pane.definition.id, end, %error, "pane build failed"); self.stats.panes_failed += 1; } } builder.built_through = Some(end); end += pane_ms; } + let definition = &builder.pane.definition.id; self.store - .evict_through(&builder.definition.id, watermark - builder.retention_ms); + .evict_through(definition, watermark - builder.pane.retention_ms); } } } @@ -214,33 +262,6 @@ pub fn pane_end(timestamp_ms: i64, pane_ms: i64) -> i64 { -(-timestamp_ms).div_euclid(pane_ms) * pane_ms } -/// Runs the plan's `Scan` over `samples`, then the definition's build over -/// the rows it selects, for the pane ending at `end`. -fn build_pane( - dag: &PhysicalASAPDAG, - builder: &Builder, - end: i64, - samples: &[RawSample], -) -> Result { - let start = end - builder.definition.pane_ms; - let scope = Scope::Ingestion { - window_start_ms: start, - window_end_ms: end, - revision: 1, - }; - let selected = select(dag, &builder.scan, samples, scope.clone())?; - let input: Source<'_> = Box::new(Operator::source( - builder.scan.schema.clone(), - vec![selected], - )?); - let build = builder - .build - .instantiate(BTreeMap::from([(builder.definition.raw_input, input)]))?; - let rows = run(&build, builder.build.roots(), scope)?; - let schema = Arc::new(builder.definition.output_schema.clone()); - Ok(Batch::try_new(schema, rows)?) -} - /// The rows of `samples` the plan's `scan` selects: its label matchers /// applied by the executor's `Scan`. pub fn select( diff --git a/data_plane/src/query.rs b/data_plane/src/query.rs index 59ca6d85..488a5138 100644 --- a/data_plane/src/query.rs +++ b/data_plane/src/query.rs @@ -1,9 +1,14 @@ //! Query engine: answers a planned query at evaluation time `T` by running //! its root of the plan's DAG over stored panes and raw samples. //! -//! Each root is compiled once with two kinds of inputs: +//! Each root is compiled once with three kinds of inputs: //! - stored outputs (the SDS frontier): bound to the pane of their definition //! ending at `T - offset`, which must be stored; +//! - kept outputs (`MatAtQueryKeptAcrossEvaluations`): bound to the kept +//! pane of their definition ending at `T - offset`. A missing one is built +//! from its pane's raw samples, which the caller fetches, and kept for the +//! next evaluations; kept panes are evicted like stored ones, measured +//! from the newest evaluation; //! - query-time raw ranges, `TimeRange(range) <- TimeShift(offset)* <- Scan`: //! bound to the samples in `(T - offset - range, T - offset]`, which the //! caller fetches, filtered by the plan's `Scan`. @@ -26,7 +31,7 @@ use planner_types::ir::physical_export::PhysicalASAPDAG; use planner_types::ir::schema::{DataType, FieldDataType, Schema}; use planner_types::ir::{NonASAPOp, Operator as IrOperator}; -use crate::precompute::{run, scan_below, select, RawSample, ScanInput}; +use crate::precompute::{run, scan_below, select, PaneBuild, RawSample, ScanInput}; use crate::sds::{DefinitionId, Sds, SummaryStore}; /// Raw samples a query needs: `metric`'s samples in `(start_ms, end_ms]`. @@ -65,6 +70,14 @@ struct StoredInput { pane_ms: i64, } +struct KeptInput { + node: u64, + definition: DefinitionId, + offset_ms: i64, + /// The output's own `TimeRange`, which its raw samples feed. + raw_input: u64, +} + struct RawInput { node: u64, scan: ScanInput, @@ -75,11 +88,30 @@ struct RawInput { struct Program { compiled: CompiledPhysicalDAG, stored: Vec, + kept: Vec, raw: Vec, output: Schema, } +/// What happened to the kept panes evaluations read. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +pub struct KeptResultStats { + /// Kept panes computed and kept, because no evaluation had kept them. + pub built: u64, + /// Kept panes read from an earlier evaluation instead of recomputed. + pub reused: u64, +} + pub struct QueryEngine { + /// Kept panes (`MatAtQueryKeptAcrossEvaluations`), by definition and + /// pane end, in memory. + pub kept: SummaryStore, + /// How many kept panes evaluations built and reused. + pub kept_stats: KeptResultStats, + /// Newest evaluation time, from which kept panes are evicted. + newest_evaluation_ms: Option, + /// How each kept definition's panes are built, or why they cannot be. + kept_builds: BTreeMap>, dag: PhysicalASAPDAG, /// Each query, with its compiled root or why it cannot run. roots: Vec<(String, Result)>, @@ -88,17 +120,37 @@ pub struct QueryEngine { impl QueryEngine { pub fn new(plan: &InstalledPlan, sds: &Sds) -> Self { let dag = plan.dag.dag.clone(); + let kept_builds: BTreeMap<_, _> = sds + .kept_definitions + .values() + .map(|definition| { + let build = PaneBuild::new(&dag, definition, &sds.kept_outputs); + (definition.id.clone(), build.map_err(|e| e.to_string())) + }) + .collect(); let roots = plan .queries .iter() .zip(&dag.roots) - .map(|(query, root)| (query.trim().to_string(), program(&dag, sds, *root as u64))) + .map(|(query, root)| { + let program = program(&dag, sds, &kept_builds, *root as u64); + (query.trim().to_string(), program) + }) .collect(); - Self { dag, roots } + Self { + kept: SummaryStore::new(), + kept_stats: KeptResultStats::default(), + newest_evaluation_ms: None, + kept_builds, + dag, + roots, + } } - /// The root answering `query` at `time_ms`, and the raw samples it needs. - /// Fails if the query is not planned or a stored pane it reads is missing. + /// The root answering `query` at `time_ms`, and the raw samples it needs: + /// its query-time ranges and the panes of the kept outputs no evaluation + /// has kept. Fails if the query is not planned or a stored pane it reads + /// is missing. pub fn raw_ranges( &self, query: &str, @@ -109,7 +161,7 @@ impl QueryEngine { for input in &program.stored { pane(input, time_ms, store)?; } - let ranges = program + let mut ranges: Vec = program .raw .iter() .map(|input| RawRange { @@ -119,13 +171,27 @@ impl QueryEngine { end_ms: time_ms - input.offset_ms, }) .collect(); + for input in &program.kept { + let end_ms = time_ms - input.offset_ms; + if self.kept.get(&input.definition, end_ms).is_none() { + let build = self.kept_build(&input.definition); + ranges.push(RawRange { + node: input.raw_input, + metric: build.scan.metric.clone(), + start_ms: end_ms - build.definition.pane_ms, + end_ms, + }); + } + } Ok((root, ranges)) } - /// Runs `root` at `time_ms` over stored panes and `raw`, the samples of - /// each of its [`RawRange`]s by node. + /// Runs `root` at `time_ms` over stored panes, kept panes and `raw`, the + /// samples of each of its [`RawRange`]s by node. A kept pane that is + /// missing is built from its raw samples and kept; then kept panes no + /// evaluation after the newest one reads are evicted. pub fn evaluate( - &self, + &mut self, root: usize, time_ms: i64, store: &SummaryStore, @@ -147,17 +213,69 @@ impl QueryEngine { Operator::source(pane.schema().clone(), vec![pane.clone()]).map_err(failed)?; sources.insert(input.node, Box::new(source)); } + for input in &program.kept { + let end_ms = time_ms - input.offset_ms; + let pane = match self.kept.get(&input.definition, end_ms) { + Some(pane) => { + self.kept_stats.reused += 1; + pane.clone() + } + None => { + // `raw_ranges` asked for these samples unless the pane + // was kept then and evicted since. + let samples = raw.get(&input.raw_input).ok_or_else(|| { + Unanswered::MissingPane(format!( + "kept pane ending at {end_ms} of {:?} was evicted", + input.definition + )) + })?; + // A kept pane is computed as at its end, so every + // evaluation that reads it reads the same pane. + let pane_scope = Scope::Query { + evaluation_time_ms: end_ms, + revision: 1, + }; + let pane = self + .kept_build(&input.definition) + .run(&self.dag, samples, pane_scope) + .map_err(|e| Unanswered::Execution(e.to_string()))?; + self.kept.insert(&input.definition, end_ms, pane.clone()); + self.kept_stats.built += 1; + pane + } + }; + let source = Operator::source(pane.schema().clone(), vec![pane]).map_err(failed)?; + sources.insert(input.node, Box::new(source)); + } for input in &program.raw { let samples = raw.get(&input.node).map(Vec::as_slice).unwrap_or_default(); let rows = select(&self.dag, &input.scan, samples, scope.clone()).map_err(failed)?; let source = Operator::source(input.scan.schema.clone(), vec![rows]).map_err(failed)?; sources.insert(input.node, Box::new(source)); } + // Evict the kept panes older than each definition's retention + // before the newest evaluation, this one or a later one. + let newest = time_ms.max(self.newest_evaluation_ms.unwrap_or(time_ms)); + self.newest_evaluation_ms = Some(newest); + for (definition, build) in &self.kept_builds { + if let Ok(build) = build { + self.kept + .evict_through(definition, newest - build.retention_ms); + } + } let dag = program.compiled.instantiate(sources).map_err(failed)?; let rows = run(&dag, program.compiled.roots(), scope).map_err(failed)?; result_samples(&program.output, rows) } + /// The build of a kept definition a compiled root reads; roots whose + /// kept builds failed are not compiled. + fn kept_build(&self, definition: &DefinitionId) -> &PaneBuild { + self.kept_builds[definition] + .as_ref() + .expect("compiled roots read only kept definitions that build") + } + fn program(&self, query: &str) -> Result<(usize, &Program), Unanswered> { let query = query.trim(); let root = self @@ -193,8 +311,13 @@ fn pane<'a>( }) } -/// Compiles `root` with its stored and raw inputs, or says why it cannot. -fn program(dag: &PhysicalASAPDAG, sds: &Sds, root: u64) -> Result { +/// Compiles `root` with its stored, kept and raw inputs, or says why it cannot. +fn program( + dag: &PhysicalASAPDAG, + sds: &Sds, + kept_builds: &BTreeMap>, + root: u64, +) -> Result { let nodes: HashMap = dag.nodes.iter().map(|n| (n.id as u64, n)).collect(); let mut producers: HashMap> = HashMap::new(); for edge in &dag.edges { @@ -203,7 +326,7 @@ fn program(dag: &PhysicalASAPDAG, sds: &Sds, root: u64) -> Result Result { raw.push(raw_input( @@ -237,9 +370,13 @@ fn program(dag: &PhysicalASAPDAG, sds: &Sds, root: u64) -> Result Result, /// Every frontier output, with its definition and pane. pub outputs: BTreeMap, + /// Definitions of the outputs computed at query time and kept across + /// evaluations, by identity. + pub kept_definitions: BTreeMap, + /// Every output kept across evaluations, with its definition and pane. + pub kept_outputs: BTreeMap, } #[derive(Debug, thiserror::Error, PartialEq)] pub enum SdsError { #[error("node {0} is not in the DAG")] UnknownNode(u64), - #[error("ingestion-time output {node} does not read one pane of one Scan: {reason}")] + #[error("output {node} does not read one pane of one Scan: {reason}")] UnsupportedShape { node: u64, reason: String }, - /// The data plane keeps no query-time result across evaluations; it - /// declares `query_time_results_retained: false`, so the Planner should - /// not select such a plan. - #[error("node {0} is kept across evaluations (MatAtQueryKeptAcrossEvaluations), which the data plane does not support")] - KeptAcrossEvaluations(u64), } impl Sds { /// The definitions behind `frontier`, the DAG outputs computed at - /// ingestion time and read at query time. + /// ingestion time and read at query time, and behind the outputs kept + /// across evaluations (`MatAtQueryKeptAcrossEvaluations`). pub fn from_dag(dag: &PhysicalASAPDAG, frontier: &[u64]) -> Result { let nodes: HashMap = dag.nodes.iter().map(|n| (n.id as u64, n)).collect(); @@ -85,40 +95,63 @@ impl Sds { .or_default() .push(edge.producer as u64); } - if let Some(node) = dag.nodes.iter().find(|n| { - matches!( - n.materialization, - Materialization::MatAtQueryKeptAcrossEvaluations(_) - ) - }) { - return Err(SdsError::KeptAcrossEvaluations(node.id as u64)); - } let mut sds = Sds::default(); let mut hashes = HashMap::new(); for &node in frontier { - let (raw_input, pane_ms, offset_ms) = pane_of(node, &nodes, &inputs)?; - let id = DefinitionId(hex(&identity(node, &nodes, &inputs, &mut hashes)?)); - let output_schema = nodes[&node].output_schema.clone(); - let definition = - sds.definitions - .entry(id.clone()) - .or_insert_with(|| SummaryDefinition { - id: id.clone(), - pane_ms, - builder: node, - raw_input, - output_schema, - }); - if node < definition.builder { - definition.builder = node; - definition.raw_input = raw_input; + let (definitions, outputs) = (&mut sds.definitions, &mut sds.outputs); + add_output(node, &nodes, &inputs, &mut hashes, definitions, outputs)?; + } + for node in &dag.nodes { + if let Materialization::MatAtQueryKeptAcrossEvaluations(_) = node.materialization { + let (definitions, outputs) = (&mut sds.kept_definitions, &mut sds.kept_outputs); + add_output( + node.id as u64, + &nodes, + &inputs, + &mut hashes, + definitions, + outputs, + )?; } - sds.outputs.insert(node, (id, PaneBinding { offset_ms })); } Ok(sds) } } +/// Adds `node`, which must read one pane, to `outputs`, and its definition +/// to `definitions` if it is new. +fn add_output( + node: u64, + nodes: &HashMap, + inputs: &HashMap>, + hashes: &mut HashMap, + definitions: &mut BTreeMap, + outputs: &mut BTreeMap, +) -> Result<(), SdsError> { + let (raw_input, pane_ms, offset_ms) = pane_of(node, nodes, inputs)?; + let id = DefinitionId(hex(&identity(node, nodes, inputs, hashes)?)); + let output_schema = nodes[&node].output_schema.clone(); + let definition = definitions + .entry(id.clone()) + .or_insert_with(|| SummaryDefinition { + id: id.clone(), + pane_ms, + builder: node, + raw_input, + output_schema, + }); + if node < definition.builder { + definition.builder = node; + definition.raw_input = raw_input; + } + let binding = PaneBinding { + offset_ms, + raw_input, + }; + outputs.insert(node, (id, binding)); + Ok(()) +} + /// `(raw input, pane width, offset)` of a frontier output: follow its single /// inputs down to the `TimeRange`, then sum the `TimeShift`s down to the `Scan`. fn pane_of( diff --git a/data_plane/tests/common/mod.rs b/data_plane/tests/common/mod.rs index 6029a045..bf670ba3 100644 --- a/data_plane/tests/common/mod.rs +++ b/data_plane/tests/common/mod.rs @@ -1,6 +1,7 @@ use std::path::Path; use control_plane::{plan_workload, DeploymentInputs, InstalledPlan}; +use planner_types::workload::PlanningWorkload; /// The deployment inputs of a backend that keeps raw data or not. pub fn inputs(raw_data_retained: bool) -> DeploymentInputs { @@ -9,13 +10,18 @@ pub fn inputs(raw_data_retained: bool) -> DeploymentInputs { inputs } -/// The selected plan of a #509 example workload in `examples/`, for a -/// deployment that keeps raw data or not. -pub fn plan(example: &str, raw_data_retained: bool) -> InstalledPlan { +/// A #509 example workload in `examples/`. +pub fn workload(example: &str) -> PlanningWorkload { let path = Path::new(env!("CARGO_MANIFEST_DIR")) .join(format!("../examples/planner-layering-{example}.json")); - let workload = serde_json::from_str(&std::fs::read_to_string(path).unwrap()).unwrap(); - plan_workload(&workload, &inputs(raw_data_retained)).unwrap() + serde_json::from_str(&std::fs::read_to_string(path).unwrap()).unwrap() +} + +/// The selected plan of a #509 example workload, for a deployment that +/// keeps raw data or not. +#[allow(dead_code)] +pub fn plan(example: &str, raw_data_retained: bool) -> InstalledPlan { + plan_workload(&workload(example), &inputs(raw_data_retained)).unwrap() } #[allow(dead_code)] diff --git a/data_plane/tests/e2e_examples.rs b/data_plane/tests/e2e_examples.rs index abb49732..76b655a8 100644 --- a/data_plane/tests/e2e_examples.rs +++ b/data_plane/tests/e2e_examples.rs @@ -9,6 +9,7 @@ mod common; use std::collections::BTreeMap; use std::path::{Path, PathBuf}; use std::process::{Child, Command}; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -79,7 +80,7 @@ fn reference( ) -> Vec { let sds = Sds::from_dag(&plan.dag.dag, &frontier_from_timing(&plan.dag.dag).unwrap()).unwrap(); assert!(sds.definitions.is_empty()); - let engine = QueryEngine::new(plan, &sds); + let mut engine = QueryEngine::new(plan, &sds); let store = SummaryStore::new(); let (root, ranges) = engine.raw_ranges(query, time_ms, &store).unwrap(); let raw = ranges @@ -107,10 +108,14 @@ fn sorted(mut samples: Vec) -> Vec { /// A Prometheus holding `all`: range selectors `{__name__="m"}[Nms]` at /// `time` return the samples in `[time - N, time]`, as Prometheus 2 does; -/// every other query returns an empty vector. -async fn prometheus(all: Arc>) -> String { +/// every other query returns an empty vector. Returns its URL and how many +/// range selectors it has answered. +async fn prometheus(all: Arc>) -> (String, Arc) { + let range_reads = Arc::new(AtomicUsize::new(0)); + let reads = range_reads.clone(); let handler = move |Query(params): Query>| { let all = all.clone(); + let reads = reads.clone(); async move { let query = ¶ms["query"]; // Anything but a raw range is a query the data plane forwarded @@ -125,6 +130,7 @@ async fn prometheus(all: Arc>) -> String { ); }; let range: i64 = rest.strip_suffix("ms]").unwrap().parse().unwrap(); + reads.fetch_add(1, Ordering::SeqCst); let time = (params["time"].parse::().unwrap() * 1000.0).round() as i64; let mut by_series: BTreeMap<&BTreeMap, Vec> = BTreeMap::new(); for s in all.iter() { @@ -149,7 +155,7 @@ async fn prometheus(all: Arc>) -> String { let url = format!("http://{}", listener.local_addr().unwrap()); let app = Router::new().route("/api/v1/query", get(handler)); tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); - url + (url, range_reads) } /// The `data_plane` binary serving `plan`, killed on drop. @@ -331,7 +337,7 @@ async fn check_example( let plan = plan_workload(&workload, inputs).unwrap(); let at_query_time = plan_selected_at_query_time(&workload, inputs).unwrap(); let all = Arc::new(all); - let process = start(&plan, &prometheus(all.clone()).await).await; + let process = start(&plan, &prometheus(all.clone()).await.0).await; remote_write(&process.url, &all).await; for query in &plan.queries { let expected = reference(&at_query_time, query, time_ms, &all); @@ -440,3 +446,41 @@ async fn example4b() { let definitions = check_example("example4b", &without_raw_data(), all, 70 * MINUTE, 0.0).await; assert_eq!(definitions, 1); } + +/// #509 Example 4's B3: Example 3b's p99 panel over ten series sampled every +/// millisecond, where the Planner keeps five 1-minute KLL panes built at +/// query time across evaluations. Three evaluations a minute apart match +/// the reference; the first reads all five panes' raw samples from +/// Prometheus, each later one only its newest pane. +#[tokio::test] +async fn example4_b3() { + let series = [ + series("latency_ms", &[("service", "a")]), + series("latency_ms", &[("service", "b")]), + ]; + let all = samples(&series, 0, 13 * MINUTE, 15_000, |i, t| { + (i * 100) as f64 + (t / 15_000 * 25 % 101) as f64 + }); + let workload = workload("example4-b3"); + let inputs = DeploymentInputs::default(); + let plan = plan_workload(&workload, &inputs).unwrap(); + let at_query_time = plan_selected_at_query_time(&workload, &inputs).unwrap(); + let dag = &plan.dag.dag; + let sds = Sds::from_dag(dag, &frontier_from_timing(dag).unwrap()).unwrap(); + assert_eq!((sds.definitions.len(), sds.kept_definitions.len()), (0, 1)); + let all = Arc::new(all); + let (prometheus_url, range_reads) = prometheus(all.clone()).await; + let process = start(&plan, &prometheus_url).await; + remote_write(&process.url, &all).await; + let query = &plan.queries[0]; + let mut reads = Vec::new(); + for minute in [10, 11, 12] { + let before = range_reads.load(Ordering::SeqCst); + let expected = reference(&at_query_time, query, minute * MINUTE, &all); + let actual = planned_answer(&process.url, query, minute * MINUTE).await; + assert_same(query, &actual, &expected, 0.0); + reads.push(range_reads.load(Ordering::SeqCst) - before); + } + eprintln!("{query}: raw ranges read per evaluation {reads:?}"); + assert_eq!(reads, vec![5, 1, 1]); +} diff --git a/data_plane/tests/kept.rs b/data_plane/tests/kept.rs new file mode 100644 index 00000000..50107bd3 --- /dev/null +++ b/data_plane/tests/kept.rs @@ -0,0 +1,194 @@ +//! Query-time results kept across evaluations (`MatAtQueryKeptAcrossEvaluations`): +//! #509 Example 4's B3, five 1-minute KLL panes built at query time, each +//! kept for the next evaluations of `quantile_over_time(0.99, latency_ms[5m])` +//! every minute. + +mod common; + +use std::collections::BTreeMap; +use std::sync::Arc; + +use asap_executor::physical_planner::frontier_from_timing; +use control_plane::{plan_selected_at_query_time, InstalledPlan}; +use data_plane::precompute::RawSample; +use data_plane::query::{KeptResultStats, QueryEngine, RawRange, ResultSample}; +use data_plane::sds::{Sds, SummaryStore}; + +const MINUTE: i64 = 60_000; +const P99: &str = "quantile_over_time(0.99, latency_ms[5m])"; + +fn engine(plan: &InstalledPlan) -> QueryEngine { + let dag = &plan.dag.dag; + let sds = Sds::from_dag(dag, &frontier_from_timing(dag).unwrap()).unwrap(); + assert!(sds.definitions.is_empty(), "nothing is built at ingestion"); + QueryEngine::new(plan, &sds) +} + +/// Two series every 15 s in `[0, 30 min]`, the `n`th sample `25n mod 101` +/// (plus 100 for series b): each 5-minute window's p99 falls in an older +/// pane and changes every minute. KLL is exact below k. +fn samples() -> Vec { + let series: Vec<_> = ["a", "b"] + .iter() + .map(|service| { + Arc::new(BTreeMap::from([ + ("__name__".to_string(), "latency_ms".to_string()), + ("service".to_string(), service.to_string()), + ])) + }) + .collect(); + (0..=30 * MINUTE) + .step_by(15_000) + .flat_map(|t| { + series.iter().enumerate().map(move |(i, labels)| RawSample { + labels: labels.clone(), + timestamp_ms: t, + value: (i * 100) as f64 + (t / 15_000 * 25 % 101) as f64, + }) + }) + .collect() +} + +fn fetch(ranges: &[RawRange], all: &[RawSample]) -> BTreeMap> { + ranges + .iter() + .map(|range| { + let samples = all + .iter() + .filter(|s| { + s.labels["__name__"] == range.metric + && range.start_ms < s.timestamp_ms + && s.timestamp_ms <= range.end_ms + }) + .cloned() + .collect(); + (range.node, samples) + }) + .collect() +} + +/// Answers `P99` at `time_ms`, returning the answer and the raw ranges it read. +fn answer( + engine: &mut QueryEngine, + time_ms: i64, + all: &[RawSample], +) -> (Vec, Vec) { + let store = SummaryStore::new(); + let (root, ranges) = engine.raw_ranges(P99, time_ms, &store).unwrap(); + let mut result = engine + .evaluate(root, time_ms, &store, &fetch(&ranges, all)) + .unwrap(); + result.sort_by(|a, b| a.labels.cmp(&b.labels)); + (result, ranges) +} + +/// The installed B3 plan and the Planner executor's reference for it: the +/// same selection with nothing materialized. +fn plans() -> (InstalledPlan, InstalledPlan) { + let workload = common::workload("example4-b3"); + let inputs = common::inputs(true); + ( + control_plane::plan_workload(&workload, &inputs).unwrap(), + plan_selected_at_query_time(&workload, &inputs).unwrap(), + ) +} + +/// The first evaluation builds all five kept panes and answers like the +/// reference; the next one, a minute later, builds only the newest pane, +/// reads only its minute of raw samples, reuses the other four, and still +/// answers like the reference. +#[test] +fn kept_panes_answer_like_the_reference_and_are_reused() { + let (plan, reference_plan) = plans(); + let all = samples(); + let mut kept = engine(&plan); + let mut reference = engine(&reference_plan); + + let (first, ranges) = answer(&mut kept, 10 * MINUTE, &all); + assert_eq!(ranges.len(), 5, "every pane is built: {ranges:?}"); + assert_eq!(first, answer(&mut reference, 10 * MINUTE, &all).0); + assert_eq!(first.len(), 2); + assert_eq!( + kept.kept_stats, + KeptResultStats { + built: 5, + reused: 0 + } + ); + + let (second, ranges) = answer(&mut kept, 11 * MINUTE, &all); + assert_eq!( + ranges + .iter() + .map(|r| (r.start_ms, r.end_ms)) + .collect::>(), + vec![(10 * MINUTE, 11 * MINUTE)], + "only the newest pane is read" + ); + assert_eq!(second, answer(&mut reference, 11 * MINUTE, &all).0); + assert_ne!(second, first, "the answer moves with the window"); + assert_eq!( + kept.kept_stats, + KeptResultStats { + built: 6, + reused: 4 + } + ); + + // The same evaluation again recomputes nothing. + let (again, ranges) = answer(&mut kept, 11 * MINUTE, &all); + assert!(ranges.is_empty()); + assert_eq!(again, second); + assert_eq!( + kept.kept_stats, + KeptResultStats { + built: 6, + reused: 9 + } + ); + assert_eq!( + reference.kept_stats, + KeptResultStats::default(), + "the reference keeps nothing" + ); +} + +/// A kept pane is dropped once it is older than the largest offset any +/// output reads (4 min) plus two pane widths, measured from the newest +/// evaluation: 6 min. +#[test] +fn kept_panes_are_evicted_like_stored_panes() { + let (plan, _) = plans(); + let dag = &plan.dag.dag; + let sds = Sds::from_dag(dag, &frontier_from_timing(dag).unwrap()).unwrap(); + let definition = sds.kept_definitions.keys().next().unwrap().clone(); + let all = samples(); + let mut kept = engine(&plan); + let ends = |engine: &QueryEngine| { + (0..=30) + .map(|m| m * MINUTE) + .filter(|end| engine.kept.get(&definition, *end).is_some()) + .map(|end| end / MINUTE) + .collect::>() + }; + + answer(&mut kept, 10 * MINUTE, &all); + assert_eq!(ends(&kept), vec![6, 7, 8, 9, 10]); + answer(&mut kept, 11 * MINUTE, &all); + assert_eq!( + ends(&kept), + vec![6, 7, 8, 9, 10, 11], + "6 is not older than 11 - 6" + ); + answer(&mut kept, 12 * MINUTE, &all); + assert_eq!(ends(&kept), vec![7, 8, 9, 10, 11, 12]); + // An evaluation far ahead builds its own five panes and drops the rest. + answer(&mut kept, 20 * MINUTE, &all); + assert_eq!(ends(&kept), vec![16, 17, 18, 19, 20]); + // An older evaluation is answered, but its panes are not kept. + let (old, ranges) = answer(&mut kept, 12 * MINUTE, &all); + assert_eq!(ranges.len(), 5); + assert_eq!(old.len(), 2); + assert_eq!(ends(&kept), vec![16, 17, 18, 19, 20]); + assert_eq!(kept.kept.len(), 5); +} diff --git a/data_plane/tests/query.rs b/data_plane/tests/query.rs index 04af1d49..26a49e36 100644 --- a/data_plane/tests/query.rs +++ b/data_plane/tests/query.rs @@ -77,7 +77,7 @@ fn fetch(ranges: &[RawRange], all: &[RawSample]) -> BTreeMap .collect() } -fn answer(f: &Fixture, query: &str, time_ms: i64, all: &[RawSample]) -> Vec { +fn answer(f: &mut Fixture, query: &str, time_ms: i64, all: &[RawSample]) -> Vec { let (root, ranges) = f .query .raw_ranges(query, time_ms, &f.precompute.store) @@ -118,9 +118,9 @@ fn stored_panes_answer_like_a_single_build() { .raw_ranges(P99, 70 * MINUTE, &panes.precompute.store) .unwrap(); assert!(ranges.is_empty(), "every input is a stored pane"); - let from_panes = answer(&panes, P99, 70 * MINUTE, &all); + let from_panes = answer(&mut panes, P99, 70 * MINUTE, &all); - let single = fixture(common::plan("example4b", true)); + let mut single = fixture(common::plan("example4b", true)); let (_, ranges) = single .query .raw_ranges(P99, 70 * MINUTE, &single.precompute.store) @@ -133,7 +133,7 @@ fn stored_panes_answer_like_a_single_build() { vec![(10 * MINUTE, 70 * MINUTE)], "one raw hour, left-open" ); - let from_raw = answer(&single, P99, 70 * MINUTE, &all); + let from_raw = answer(&mut single, P99, 70 * MINUTE, &all); assert_eq!(from_panes, from_raw); assert_eq!(from_panes.len(), 2); @@ -155,9 +155,9 @@ fn query_time_plans_read_raw_ranges() { ("http_requests_total", &[("job", "b"), ("instance", "1")]), ]; let all = samples(series, 0, 10 * MINUTE, |_, t| t as f64 / 1000.0); - let f = fixture(common::plan("example1", true)); + let mut f = fixture(common::plan("example1", true)); let result = answer( - &f, + &mut f, "sum by (job) (rate(http_requests_total[1m]))", 5 * MINUTE, &all, @@ -173,7 +173,7 @@ fn query_time_plans_read_raw_ranges() { assert!((by_job[1].1 - 1.0).abs() < 1e-9, "{by_job:?}"); let topk = answer( - &f, + &mut f, "topk by (job) (10, sum_over_time(http_requests_total[1m]))", 5 * MINUTE, &all, diff --git a/data_plane/tests/sds.rs b/data_plane/tests/sds.rs index 68e15786..60008c07 100644 --- a/data_plane/tests/sds.rs +++ b/data_plane/tests/sds.rs @@ -107,17 +107,52 @@ fn rejects_outputs_without_a_pane() { ); } -/// A plan that keeps a query-time result across evaluations is not -/// installed: the data plane declares it cannot keep one. +/// B3's five query-time KLL panes kept across evaluations are one kept +/// definition, keyed like ingestion-time panes: a 1-minute pane, each output +/// reading the pane `i * 1min` before evaluation. Nothing is built at +/// ingestion time. #[test] -fn rejects_results_kept_across_evaluations() { - let mut dag = plan("example4b", true).dag.dag; - let node = &mut dag.nodes[0]; - node.materialization = Materialization::MatAtQueryKeptAcrossEvaluations(Medium::Memory); - let id = node.id as u64; +fn kept_panes_share_one_definition() { + let plan = plan("example4-b3", true); + let dag = &plan.dag.dag; + assert!(dag.nodes.iter().any(|n| matches!( + n.materialization, + Materialization::MatAtQueryKeptAcrossEvaluations(_) + ))); + let sds = sds_of(dag).unwrap(); + assert!(sds.definitions.is_empty() && sds.outputs.is_empty()); + assert_eq!(sds.kept_definitions.len(), 1); + let definition = sds.kept_definitions.values().next().unwrap(); + assert_eq!(definition.pane_ms, 60_000); + let mut offsets: Vec = sds + .kept_outputs + .values() + .map(|(id, binding)| { + assert_eq!(id, &definition.id); + binding.offset_ms + }) + .collect(); + offsets.sort(); + assert_eq!(offsets, (0..5).map(|i| i * 60_000).collect::>()); +} + +/// A kept output must read one pane of raw samples, like a stored one. +#[test] +fn rejects_kept_outputs_without_a_pane() { + let mut dag = plan("example4-b3", true).dag.dag; + let scan = dag + .nodes + .iter_mut() + .find(|n| matches!(n.payload, Operator::NonASAP(NonASAPOp::Scan { .. }))) + .unwrap(); + scan.materialization = Materialization::MatAtQueryKeptAcrossEvaluations(Medium::Memory); + let scan = scan.id as u64; assert_eq!( - Sds::from_dag(&dag, &[]), - Err(SdsError::KeptAcrossEvaluations(id)) + sds_of(&dag), + Err(SdsError::UnsupportedShape { + node: scan, + reason: "no TimeRange".into() + }) ); } diff --git a/examples/planner-layering-example4-b3.json b/examples/planner-layering-example4-b3.json new file mode 100644 index 00000000..9ad9481d --- /dev/null +++ b/examples/planner-layering-example4-b3.json @@ -0,0 +1,44 @@ +{ + "query_workload": { + "language": "promql", + "query_batch": null, + "repeating_queries": [ + { + "query": "quantile_over_time(0.99, latency_ms[5m])", + "demand": { + "fixed_interval": 60000 + }, + "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": 10000.0 + } + }, + "sources": { + "latency_ms": { + "series_count": 10, + "metric_type": null, + "sample_interval": 1, + "stream_norms": null, + "max_distinct_per_evaluation": null + } + } + } +} From e0beb601e3e32fcb261794f293cfe6d983d4362e Mon Sep 17 00:00:00 2001 From: zzylol Date: Sun, 18 Oct 2026 17:25:59 +0000 Subject: [PATCH 2/2] feat(query): keep a query-time pane only once it is closed A pane computed at query time is kept across evaluations only once its end is at or before the precompute engine's watermark (newest sample minus the allowed lateness). A pane read while still open is computed for that evaluation but not kept, so a later evaluation computes it again, including samples that arrived late, and keeps it once it is closed. --- README.md | 9 ++-- data_plane/src/engine.rs | 1 + data_plane/src/query.rs | 21 ++++++--- data_plane/tests/e2e_examples.rs | 5 ++- data_plane/tests/kept.rs | 75 +++++++++++++++++++++++++++++++- data_plane/tests/query.rs | 8 +++- 6 files changed, 105 insertions(+), 14 deletions(-) diff --git a/README.md b/README.md index 3b87370e..9ebd9ae3 100644 --- a/README.md +++ b/README.md @@ -46,9 +46,12 @@ plan (`x-asap-execution: plan`). Its stored inputs read the panes ending at `T - offset`, so `T` must be on the pane grid and those panes stored; its query-time raw ranges are read from Prometheus. An input the plan keeps across evaluations (`MatAtQueryKeptAcrossEvaluations`) is computed from its -pane's raw samples the first time an evaluation reads it, kept in memory by -definition and pane end, and reused by later evaluations; it is evicted like -a stored pane, measured from the newest evaluation. Anything else is forwarded +pane's raw samples when an evaluation reads it and no evaluation has kept +it. It is kept in memory, by definition and pane end, once the pane is +closed by the precompute engine's rule (its end at or before the newest +sample minus `--allowed-lateness-ms`), and later evaluations reuse it; a pane +read before it closes is computed again by later evaluations. Kept panes are +evicted like stored ones, measured from the newest evaluation. Anything else is forwarded to Prometheus (`x-asap-execution: forwarded`); a planned query that could not be answered also says why in `x-asap-forward-reason`. Range queries, SQL and ClickHouse are not supported yet. diff --git a/data_plane/src/engine.rs b/data_plane/src/engine.rs index 282798e6..e1e173c3 100644 --- a/data_plane/src/engine.rs +++ b/data_plane/src/engine.rs @@ -174,6 +174,7 @@ impl Engine { root, time_ms, &self.precompute.store, + self.precompute.watermark_ms(), &raw, )); } diff --git a/data_plane/src/query.rs b/data_plane/src/query.rs index 488a5138..abfb539c 100644 --- a/data_plane/src/query.rs +++ b/data_plane/src/query.rs @@ -7,8 +7,9 @@ //! - kept outputs (`MatAtQueryKeptAcrossEvaluations`): bound to the kept //! pane of their definition ending at `T - offset`. A missing one is built //! from its pane's raw samples, which the caller fetches, and kept for the -//! next evaluations; kept panes are evicted like stored ones, measured -//! from the newest evaluation; +//! next evaluations once it is closed (its end at or before the +//! precompute engine's watermark); kept panes are evicted like stored +//! ones, measured from the newest evaluation; //! - query-time raw ranges, `TimeRange(range) <- TimeShift(offset)* <- Scan`: //! bound to the samples in `(T - offset - range, T - offset]`, which the //! caller fetches, filtered by the plan's `Scan`. @@ -96,7 +97,8 @@ struct Program { /// What happened to the kept panes evaluations read. #[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] pub struct KeptResultStats { - /// Kept panes computed and kept, because no evaluation had kept them. + /// Kept panes computed from raw samples, because no evaluation had kept + /// them; kept if closed. pub built: u64, /// Kept panes read from an earlier evaluation instead of recomputed. pub reused: u64, @@ -188,13 +190,16 @@ impl QueryEngine { /// Runs `root` at `time_ms` over stored panes, kept panes and `raw`, the /// samples of each of its [`RawRange`]s by node. A kept pane that is - /// missing is built from its raw samples and kept; then kept panes no - /// evaluation after the newest one reads are evicted. + /// missing is built from its raw samples, and kept if it is closed: its + /// end is at or before `watermark_ms`, the precompute engine's watermark + /// (`None` before any sample). Then kept panes no evaluation after the + /// newest one reads are evicted. pub fn evaluate( &mut self, root: usize, time_ms: i64, store: &SummaryStore, + watermark_ms: Option, raw: &BTreeMap>, ) -> Result, Unanswered> { let program = self.roots[root] @@ -239,8 +244,12 @@ impl QueryEngine { .kept_build(&input.definition) .run(&self.dag, samples, pane_scope) .map_err(|e| Unanswered::Execution(e.to_string()))?; - self.kept.insert(&input.definition, end_ms, pane.clone()); self.kept_stats.built += 1; + // Kept only once closed, as the precompute engine closes + // panes: a sample may still arrive for an open one. + if watermark_ms.is_some_and(|watermark| end_ms <= watermark) { + self.kept.insert(&input.definition, end_ms, pane.clone()); + } pane } }; diff --git a/data_plane/tests/e2e_examples.rs b/data_plane/tests/e2e_examples.rs index 76b655a8..c5397868 100644 --- a/data_plane/tests/e2e_examples.rs +++ b/data_plane/tests/e2e_examples.rs @@ -98,7 +98,7 @@ fn reference( (range.node, samples) }) .collect(); - sorted(engine.evaluate(root, time_ms, &store, &raw).unwrap()) + sorted(engine.evaluate(root, time_ms, &store, None, &raw).unwrap()) } fn sorted(mut samples: Vec) -> Vec { @@ -451,7 +451,8 @@ async fn example4b() { /// millisecond, where the Planner keeps five 1-minute KLL panes built at /// query time across evaluations. Three evaluations a minute apart match /// the reference; the first reads all five panes' raw samples from -/// Prometheus, each later one only its newest pane. +/// Prometheus, each later one only its newest pane. Samples run to 13 min, +/// so every pane read is closed and kept. #[tokio::test] async fn example4_b3() { let series = [ diff --git a/data_plane/tests/kept.rs b/data_plane/tests/kept.rs index 50107bd3..1c82d46a 100644 --- a/data_plane/tests/kept.rs +++ b/data_plane/tests/kept.rs @@ -67,16 +67,27 @@ fn fetch(ranges: &[RawRange], all: &[RawSample]) -> BTreeMap .collect() } -/// Answers `P99` at `time_ms`, returning the answer and the raw ranges it read. +/// Answers `P99` at `time_ms` once every sample in `all` has arrived (every +/// pane is closed), returning the answer and the raw ranges it read. fn answer( engine: &mut QueryEngine, time_ms: i64, all: &[RawSample], +) -> (Vec, Vec) { + answer_at_watermark(engine, time_ms, Some(30 * MINUTE), all) +} + +/// Answers `P99` at `time_ms` when the watermark is `watermark_ms`. +fn answer_at_watermark( + engine: &mut QueryEngine, + time_ms: i64, + watermark_ms: Option, + all: &[RawSample], ) -> (Vec, Vec) { let store = SummaryStore::new(); let (root, ranges) = engine.raw_ranges(P99, time_ms, &store).unwrap(); let mut result = engine - .evaluate(root, time_ms, &store, &fetch(&ranges, all)) + .evaluate(root, time_ms, &store, watermark_ms, &fetch(&ranges, all)) .unwrap(); result.sort_by(|a, b| a.labels.cmp(&b.labels)); (result, ranges) @@ -192,3 +203,63 @@ fn kept_panes_are_evicted_like_stored_panes() { assert_eq!(ends(&kept), vec![16, 17, 18, 19, 20]); assert_eq!(kept.kept.len(), 5); } + +/// A pane is kept only once it is closed (its end at or before the +/// watermark). A pane read before then is computed for that evaluation but +/// not kept, so a later evaluation computes it again and includes a sample +/// that arrived late, before the pane closed. +#[test] +fn panes_are_kept_only_once_closed() { + let (plan, reference_plan) = plans(); + let dag = &plan.dag.dag; + let sds = Sds::from_dag(dag, &frontier_from_timing(dag).unwrap()).unwrap(); + let definition = sds.kept_definitions.keys().next().unwrap().clone(); + let mut all = samples(); + let mut kept = engine(&plan); + let mut reference = engine(&reference_plan); + let ends = |engine: &QueryEngine| { + (0..=30) + .map(|m| m * MINUTE) + .filter(|end| engine.kept.get(&definition, *end).is_some()) + .map(|end| end / MINUTE) + .collect::>() + }; + + // Watermark at 9.5 min: the pane ending at 10 min is still open. + let (first, ranges) = answer_at_watermark(&mut kept, 10 * MINUTE, Some(570_000), &all); + assert_eq!(ranges.len(), 5); + assert_eq!(first, answer(&mut reference, 10 * MINUTE, &all).0); + assert_eq!(ends(&kept), vec![6, 7, 8, 9], "the open pane is not kept"); + + // A late sample for the open pane arrives before it closes. + let late = RawSample { + timestamp_ms: 10 * MINUTE - 1_000, + value: 1_000.0, + ..all + .iter() + .find(|s| s.labels["service"] == "a") + .unwrap() + .clone() + }; + all.push(late); + let (again, ranges) = answer_at_watermark(&mut kept, 10 * MINUTE, Some(585_000), &all); + assert_eq!( + ranges + .iter() + .map(|r| (r.start_ms, r.end_ms)) + .collect::>(), + vec![(9 * MINUTE, 10 * MINUTE)], + "the open pane is computed again" + ); + assert_eq!(again, answer(&mut reference, 10 * MINUTE, &all).0); + assert_ne!(again, first, "the late sample is included"); + + // Once the watermark reaches 10 min, the pane is kept. + let (_, ranges) = answer_at_watermark(&mut kept, 11 * MINUTE, Some(10 * MINUTE), &all); + assert_eq!(ranges.len(), 2, "the 10- and 11-minute panes: {ranges:?}"); + assert_eq!(ends(&kept), vec![6, 7, 8, 9, 10]); + // Without any sample there is no watermark, and nothing is kept. + let mut fresh = engine(&plan); + answer_at_watermark(&mut fresh, 10 * MINUTE, None, &all); + assert!(fresh.kept.is_empty()); +} diff --git a/data_plane/tests/query.rs b/data_plane/tests/query.rs index 26a49e36..95b7def2 100644 --- a/data_plane/tests/query.rs +++ b/data_plane/tests/query.rs @@ -84,7 +84,13 @@ fn answer(f: &mut Fixture, query: &str, time_ms: i64, all: &[RawSample]) -> Vec< .unwrap(); let mut result = f .query - .evaluate(root, time_ms, &f.precompute.store, &fetch(&ranges, all)) + .evaluate( + root, + time_ms, + &f.precompute.store, + f.precompute.watermark_ms(), + &fetch(&ranges, all), + ) .unwrap(); result.sort_by(|a, b| a.labels.cmp(&b.labels)); result