From 5780d836c8aedc3806d3e145ae8123b0dfd467f9 Mon Sep 17 00:00:00 2001 From: zzylol Date: Tue, 20 Oct 2026 19:59:45 +0000 Subject: [PATCH] feat(query): build stored outputs built in a QueryRun on first use A stored output with StoredOutputBuild::QueryRun (materialization MatAtQueryKeptAcrossEvaluations, #509 Example 4's B3) is now executed. The query engine builds its summary state from the pane's raw samples the first time an evaluation needs it, and stores it in the SDS under the same SummaryInstanceKey as stored outputs built in an IngestionRun once the pane is closed. Later evaluations reuse it. - Closure is judged by the ingestion progress snapshotted when the evaluation chose its fetch (bound_raw_sample_windows returns it with the windows and evaluate takes it back), not when the evaluation runs, so a summary state built from an earlier, partial fetch is never stored. - Eviction is by the plan's retention_ms plus one pane, measured from the newest closed pane at or before the watermark, as for IngestionRun outputs. A far-future evaluation no longer stops storing. - One fetch per QueryRun definition covers all its missing panes. - QueryRunStats (built, stored, stored_while_metric_behind, reused) is on /api/v1/status/asap under "query_run". stored_while_metric_behind counts summary states stored while their own metric's newest sample had not reached their end: the watermark is global. - A QueryRun definition whose selector does not build no longer fails the install: its queries are forwarded. SummaryBuild, shared by the precompute and query engines, compiles a definition's build and runs it over one pane's samples. Samples of a metric only QueryRun stored outputs read move the watermark (counted as samples_query_run_only) without being buffered. The backend now declares query_time_results_retained: true, and the startup warning about unbuilt QueryRun outputs is gone. examples/planner-layering-example4-b3.json is a workload for which the Planner selects B3 under the default cost model. --- README.md | 19 +- control_plane/src/lib.rs | 9 +- control_plane/tests/plan_workload.rs | 18 + data_plane/src/engine.rs | 59 ++-- data_plane/src/lib.rs | 6 +- data_plane/src/precompute.rs | 220 +++++++----- data_plane/src/query.rs | 289 ++++++++++++++-- data_plane/src/sds.rs | 4 + data_plane/tests/precompute.rs | 30 +- data_plane/tests/query.rs | 96 ++++-- data_plane/tests/query_run.rs | 381 +++++++++++++++++++++ data_plane/tests/sds.rs | 26 +- examples/planner-layering-example4-b3.json | 44 +++ 13 files changed, 1028 insertions(+), 173 deletions(-) create mode 100644 data_plane/tests/query_run.rs create mode 100644 examples/planner-layering-example4-b3.json diff --git a/README.md b/README.md index 88909a1d..c1df9747 100644 --- a/README.md +++ b/README.md @@ -54,9 +54,19 @@ stopped, which also makes `/-/ready` fail. `/api/v1/status/asap` reports the engine's sample and pane counters as JSON. 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 bound raw-sample windows are read from Prometheus. Anything else is forwarded +plan (`x-asap-execution: plan`) at its evaluation time `T`. Stored outputs +built in an `IngestionRun` are read from the SDS as the panes ending at +`T - offset`, so `T` must be on the pane grid and those panes stored. The +plan's query-time `RawSampleWindow`s are read from Prometheus. A stored output +built in a `QueryRun` (`MatAtQueryKeptAcrossEvaluations`) is built from its +pane's raw samples the first time an evaluation needs it. It is stored in the +SDS under the same key once its pane is closed, that is, once its end is at or +before the watermark at the time the evaluation chose its fetch. Later +evaluations reuse it. It is evicted like an `IngestionRun` output: the plan's +`retention_ms` plus one pane, measured from the newest closed pane at or +before the watermark. `/api/v1/status/asap` reports `query_run` counters, +including summary states stored while their own metric lagged behind the +global watermark. Anything else is forwarded to Prometheus with its end-to-end headers, such as `Authorization` and `X-Scope-OrgID` (`x-asap-execution: forwarded`); a planned query that could not be answered also says why in `x-asap-forward-reason`. Range queries, SQL and @@ -75,8 +85,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 stored outputs built in a `QueryRun` -(`MatAtQueryKeptAcrossEvaluations`). For example +in Prometheus, and stored outputs built in a `QueryRun` (`MatAtQueryKeptAcrossEvaluations`). 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 diff --git a/control_plane/src/lib.rs b/control_plane/src/lib.rs index ba1ee7a5..5e575d53 100644 --- a/control_plane/src/lib.rs +++ b/control_plane/src/lib.rs @@ -82,14 +82,15 @@ pub fn load_plan(path: &Path) -> Result { } /// What this backend's data plane can execute and keep: every summary the -/// Planner executor builds, raw samples kept in Prometheus, and no -/// stored outputs built in a `QueryRun` (`MatAtQueryKeptAcrossEvaluations`): the SDS describes such -/// outputs, but nothing builds them until the query engine does. +/// Planner executor builds, raw samples kept in Prometheus, and stored +/// outputs built in a `QueryRun` (`MatAtQueryKeptAcrossEvaluations`): the +/// query engine builds their summary states on first use and stores them in +/// the SDS once their coverage is closed. 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 f4855511..1d592a63 100644 --- a/control_plane/tests/plan_workload.rs +++ b/control_plane/tests/plan_workload.rs @@ -88,3 +88,21 @@ fn load_rejects_a_plan_without_a_query_per_root() { )); std::fs::remove_file(path).unwrap(); } + +/// The backend retains query-time results (`query_time_results_retained`), +/// so for a few series sampled every millisecond the Planner selects #509 +/// Example 4's B3: five 1-minute KLL panes, stored outputs built in a +/// `QueryRun` (`MatAtQueryKeptAcrossEvaluations`). +#[test] +fn selects_stored_outputs_built_in_a_query_run() { + let inputs = DeploymentInputs::default(); + assert!(inputs.deployment_capabilities.query_time_results_retained); + let plan = plan_workload(&example("example4-b3"), &inputs).unwrap(); + let query_run: Vec<_> = plan + .stored_outputs + .iter() + .filter(|o| o.built_at == asap_planner::StoredOutputBuild::QueryRun) + .collect(); + assert_eq!(query_run.len(), 5, "five one-minute summaries"); + assert!(query_run.iter().all(|o| o.pane.width_ms == 60_000)); +} diff --git a/data_plane/src/engine.rs b/data_plane/src/engine.rs index df882c5a..c9df45c0 100644 --- a/data_plane/src/engine.rs +++ b/data_plane/src/engine.rs @@ -12,7 +12,9 @@ use asap_planner::InstallablePlan; use tokio::sync::oneshot; use crate::precompute::{Precompute, PrecomputeOptions, PrecomputeStats, RawSample}; -use crate::query::{BoundRawSampleWindow, QueryEngine, ResultSample, Unanswered}; +use crate::query::{ + IngestionProgress, Prepared, QueryEngine, QueryRunStats, ResultSample, Unanswered, +}; use crate::sds::Sds; /// How the engine thread ingests samples. @@ -42,11 +44,12 @@ enum Command { BoundRawSampleWindows { query: String, time_ms: i64, - reply: Reply<(usize, Vec)>, + reply: Reply, }, Evaluate { root: usize, time_ms: i64, + progress: IngestionProgress, raw: BTreeMap>, reply: Reply>, }, @@ -75,6 +78,8 @@ pub struct EngineHandle { running: Arc, /// The counters as of the last command the engine ran. stats: Arc>, + /// The stored output built in a `QueryRun` counters as of the last evaluation. + query_run_stats: Arc>, } /// Marks the engine stopped when its thread exits, panics included. @@ -96,9 +101,11 @@ impl EngineHandle { let running = Arc::new(AtomicBool::new(true)); let queued_samples = Arc::new(AtomicU64::new(0)); let stats = Arc::new(Mutex::new(PrecomputeStats::default())); + let query_run_stats = Arc::new(Mutex::new(QueryRunStats::default())); let shared = Shared { queued_samples: queued_samples.clone(), stats: stats.clone(), + query_run_stats: query_run_stats.clone(), }; let guard = RunningGuard(running.clone()); std::thread::Builder::new() @@ -122,6 +129,7 @@ impl EngineHandle { max_queued_samples: options.max_queued_samples, running, stats, + query_run_stats, }; Ok((handle, queries)) } @@ -158,7 +166,7 @@ impl EngineHandle { &self, query: String, time_ms: i64, - ) -> Result<(usize, Vec), Unanswered> { + ) -> Result { self.ask(|reply| Command::BoundRawSampleWindows { query, time_ms, @@ -167,22 +175,32 @@ impl EngineHandle { .await } - /// Runs a root over stored panes and `raw`; see [`QueryEngine::evaluate`]. + /// Runs a prepared root over stored summaries and `raw`; see + /// [`QueryEngine::evaluate`]. pub async fn evaluate( &self, - root: usize, + prepared: Prepared, time_ms: i64, raw: BTreeMap>, ) -> Result, Unanswered> { self.ask(|reply| Command::Evaluate { - root, + root: prepared.root, time_ms, + progress: prepared.progress, raw, reply, }) .await } + /// The stored output built in a `QueryRun` counters as of the last evaluation. + pub fn query_run_stats(&self) -> QueryRunStats { + *self + .query_run_stats + .lock() + .unwrap_or_else(|e| e.into_inner()) + } + async fn ask(&self, command: impl FnOnce(Reply) -> Command) -> Result { let stopped = || Unanswered::Execution("the engine has stopped".into()); let (reply, answer) = oneshot::channel(); @@ -232,6 +250,7 @@ fn admit(queued: u64, count: u64, max_queued: u64) -> Result { struct Shared { queued_samples: Arc, stats: Arc>, + query_run_stats: Arc>, } struct Engine { @@ -249,18 +268,6 @@ impl Engine { outputs = sds.outputs.len(), "stored summaries" ); - let query_run = sds - .definitions - .values() - .filter(|d| d.built_at == asap_planner::StoredOutputBuild::QueryRun) - .count(); - if query_run > 0 { - tracing::warn!( - definitions = query_run, - "the plan has stored outputs built in a QueryRun, which this data plane does not build yet; \ - queries that read them are forwarded" - ); - } let queries = plan.queries.iter().map(|q| q.query.clone()).collect(); let query = QueryEngine::new(&plan, &sds); let engine = Self { @@ -289,20 +296,30 @@ impl Engine { &query, time_ms, &self.precompute.store, + self.precompute.progress(), )); } Command::Evaluate { root, time_ms, + progress, raw, reply, } => { - let _ = reply.send(self.query.evaluate( + let answer = self.query.evaluate( root, time_ms, - &self.precompute.store, + &mut self.precompute.store, + &progress, &raw, - )); + ); + // Published before the reply, so a caller that awaited + // the evaluation reads its counters. + *shared + .query_run_stats + .lock() + .unwrap_or_else(|e| e.into_inner()) = self.query.query_run_stats; + let _ = reply.send(answer); } Command::Stop => break, } diff --git a/data_plane/src/lib.rs b/data_plane/src/lib.rs index b65ac699..455fd9f0 100644 --- a/data_plane/src/lib.rs +++ b/data_plane/src/lib.rs @@ -176,6 +176,7 @@ async fn status(State(state): State>) -> Response { "engine_running": state.engine.is_running(), "samples_queued": state.engine.queued_samples(), "precompute": state.engine.stats(), + "query_run": state.engine.query_run_stats(), }); ( [(header::CONTENT_TYPE, "application/json")], @@ -249,10 +250,11 @@ async fn answer( .duration_since(std::time::UNIX_EPOCH) .map_or(0, |d| d.as_millis() as i64) }); - let (root, ranges) = state + let prepared = state .engine .bound_raw_sample_windows(query.query, time_ms) .await?; + let ranges = &prepared.windows; let mut fetch_headers = forwarded_headers(headers); // The raw read is a GET without the caller's body. fetch_headers.remove(header::CONTENT_TYPE); @@ -267,7 +269,7 @@ async fn answer( .map_err(|e| Unanswered::Execution(format!("fetching raw samples: {e}")))? .into_iter() .collect(); - let samples = state.engine.evaluate(root, time_ms, raw).await?; + let samples = state.engine.evaluate(prepared, time_ms, raw).await?; Ok(prometheus::vector_response(time_ms, &samples)) } diff --git a/data_plane/src/precompute.rs b/data_plane/src/precompute.rs index 3cdedce2..19d6c475 100644 --- a/data_plane/src/precompute.rs +++ b/data_plane/src/precompute.rs @@ -29,6 +29,8 @@ use asap_executor::{RawSampleSelector, Sample}; use futures::executor::block_on; use futures::StreamExt; +use asap_planner::StoredOutputBuild; + use crate::sds::{ DefinitionId, Sds, Selection, SummaryDefinition, SummaryInstanceKey, SummaryStore, }; @@ -68,6 +70,10 @@ pub struct PrecomputeStats { pub samples_dropped_future: u64, /// Samples of a metric no stored summary reads. pub samples_unrouted: u64, + /// Samples of a metric only stored outputs built in a `QueryRun` read: they move the + /// watermark, which says when such a summary is closed, but are not + /// buffered (the query engine reads them from Prometheus). + pub samples_query_run_only: u64, /// Panes built and stored. pub panes_built: u64, /// Panes whose build failed; they are skipped, not retried. @@ -95,13 +101,84 @@ impl Default for PrecomputeOptions { } } -struct Builder { - definition: SummaryDefinition, - /// Turns a pane's samples into the build's input: the `Scan`'s label - /// matchers, then the pane's `TimeRange` rows. +/// How one definition's summaries are built from raw samples. The +/// precompute engine builds stored outputs built in an `IngestionRun` with it, and the query +/// engine `QueryRun` ones. +pub struct SummaryBuild { + pub definition: SummaryDefinition, + /// Turns a coverage's samples into the build's input: the `Scan`'s label + /// matchers, then the coverage's `TimeRange` rows. selector: RawSampleSelector, raw_schema: SchemaRef, build: CompiledPhysicalDAG, +} + +impl SummaryBuild { + /// Compiles `definition`'s build from its `TimeRange` up to its builder. + /// A definition built from other stored outputs is refused until the + /// engine can build it once its inputs are sealed. + pub fn new( + dag: &PhysicalASAPDAG, + definition: &SummaryDefinition, + ) -> Result { + let error = |message: String| PrecomputeError::Definition(definition.id.clone(), message); + if !definition.stored_inputs.is_empty() { + return Err(error( + "built from other stored outputs, which the data plane cannot build yet".into(), + )); + } + if definition.pane_ms <= 0 { + return Err(error(format!("pane width {} ms", definition.pane_ms))); + } + let raw_schema = Arc::new(output_schema(dag, definition.raw_input()).clone()); + // `window_batch` is called with each coverage's end, so the window's + // offset from evaluation time is not used. + let selector = RawSampleSelector::new(dag, &definition.window)?; + let build = compile( + dag, + BTreeMap::from([( + definition.raw_input(), + InputContract::bounded(raw_schema.clone()), + )]), + &[definition.builder], + )?; + Ok(Self { + definition: definition.clone(), + selector, + raw_schema, + build, + }) + } + + /// The metric whose samples the build reads. + pub fn metric(&self) -> &str { + self.selector.metric() + } + + /// Selects the coverage ending at `end_ms` from `samples` with the plan's + /// `Scan` matchers, then runs the build over those rows under `scope`. + pub fn run( + &self, + end_ms: i64, + samples: &[RawSample], + scope: Scope, + ) -> Result { + let selected = self + .selector + .window_batch(end_ms, &executor_samples(samples))?; + let input: Source<'_> = + Box::new(Operator::source(self.raw_schema.clone(), vec![selected])?); + let build = self + .build + .instantiate(BTreeMap::from([(self.definition.raw_input(), input)]))?; + let rows = run_root(&build, self.definition.builder, scope)?; + let schema = Arc::new(self.definition.output_schema.clone()); + Ok(Batch::try_new(schema, rows)?) + } +} + +struct Builder { + summary: SummaryBuild, /// Samples of open panes, by pane end. open: BTreeMap>, /// End of the first pane seen, which is skipped. `None` until then. @@ -121,6 +198,9 @@ enum Outcome { pub struct Precompute { builders: Vec, + /// Metrics read by stored outputs built in a `QueryRun`, with each one's newest routed + /// sample time. + query_run_metrics: BTreeMap>, allowed_lateness_ms: i64, max_future_ms: i64, newest_sample_ms: Option, @@ -147,43 +227,35 @@ impl Precompute { .ingestion_build_order() .map_err(|e| PrecomputeError::Plan(e.to_string()))?; for id in &order { - let definition = &sds.definitions[id]; - let error = - |message: String| PrecomputeError::Definition(definition.id.clone(), message); - if !definition.stored_inputs.is_empty() { - return Err(error( - "built from other stored outputs, which the precompute engine cannot build yet" - .into(), - )); - } - if definition.pane_ms <= 0 { - return Err(error(format!("pane width {} ms", definition.pane_ms))); - } - let raw_schema = Arc::new(output_schema(&dag, definition.raw_input()).clone()); - // `window_batch` is called with each pane's end, so the window's - // offset from evaluation time is not used. - let selector = RawSampleSelector::new(&dag, &definition.window)?; - let build = compile( - &dag, - BTreeMap::from([( - definition.raw_input(), - InputContract::bounded(raw_schema.clone()), - )]), - &[definition.builder], - )?; builders.push(Builder { - definition: definition.clone(), - selector, - raw_schema, - build, + summary: SummaryBuild::new(&dag, &sds.definitions[id])?, open: BTreeMap::new(), first_pane_end: None, built_through: None, }); } + // Samples of a metric only stored outputs built in a `QueryRun` read move the + // watermark, which says when such a summary's coverage is closed. + // A `QueryRun` definition whose selector does not build is not + // tracked here; the query engine forwards the queries that read it. + let mut query_run_metrics = BTreeMap::new(); + for definition in sds.definitions.values() { + if definition.built_at == StoredOutputBuild::QueryRun { + match RawSampleSelector::new(&dag, &definition.window) { + Ok(selector) => { + query_run_metrics.insert(selector.metric().to_string(), None); + } + Err(error) => tracing::warn!( + definition = ?definition.id, %error, + "stored output built in a `QueryRun` cannot be built; its queries are forwarded" + ), + } + } + } let millis = |value: u64| i64::try_from(value).unwrap_or(i64::MAX); Ok(Self { builders, + query_run_metrics, allowed_lateness_ms: millis(options.allowed_lateness_ms), max_future_ms: millis(options.max_future_ms), newest_sample_ms: None, @@ -193,6 +265,20 @@ impl Precompute { }) } + /// What ingestion has seen, for the query engine's store decisions: the + /// watermark and the newest sample of each metric a `QueryRun` stored + /// output reads. + pub fn progress(&self) -> crate::query::IngestionProgress { + crate::query::IngestionProgress { + watermark_ms: self.watermark_ms(), + newest_sample_ms: self + .query_run_metrics + .iter() + .filter_map(|(metric, newest)| Some((metric.clone(), (*newest)?))) + .collect(), + } + } + /// The watermark: no sample older than this is expected any more. It is /// the newest routed sample, capped at the wall clock, minus the allowed /// lateness. @@ -220,10 +306,12 @@ impl Precompute { let latest_allowed_ms = now_ms.saturating_add(self.max_future_ms); for sample in samples { let metric = sample.labels.get("__name__").map(String::as_str); - if !self - .builders - .iter() - .any(|b| metric == Some(b.selector.metric())) + let query_run = metric.is_some_and(|m| self.query_run_metrics.contains_key(m)); + if !query_run + && !self + .builders + .iter() + .any(|b| metric == Some(b.summary.metric())) { self.stats.samples_unrouted += 1; continue; @@ -234,10 +322,10 @@ impl Precompute { } let mut outcome: Option = None; for builder in &mut self.builders { - if metric != Some(builder.selector.metric()) { + if metric != Some(builder.summary.metric()) { continue; } - let this = match pane_end(sample.timestamp_ms, builder.definition.pane_ms) { + let this = match pane_end(sample.timestamp_ms, builder.summary.definition.pane_ms) { None => Outcome::Late, Some(end) => { let first = *builder.first_pane_end.get_or_insert(end); @@ -255,15 +343,14 @@ impl Precompute { outcome = outcome.max(Some(this)); } match outcome { - None => self.stats.samples_unrouted += 1, - Some(outcome) => { - match outcome { - Outcome::Buffered => self.stats.samples_accepted += 1, - Outcome::FirstPane => self.stats.samples_skipped_first_pane += 1, - Outcome::Late => self.stats.samples_dropped_late += 1, - } - self.newest_sample_ms = self.newest_sample_ms.max(Some(sample.timestamp_ms)); - } + None => self.stats.samples_query_run_only += 1, + Some(Outcome::Buffered) => self.stats.samples_accepted += 1, + Some(Outcome::FirstPane) => self.stats.samples_skipped_first_pane += 1, + Some(Outcome::Late) => self.stats.samples_dropped_late += 1, + } + self.newest_sample_ms = self.newest_sample_ms.max(Some(sample.timestamp_ms)); + if let Some(newest) = metric.and_then(|m| self.query_run_metrics.get_mut(m)) { + *newest = (*newest).max(Some(sample.timestamp_ms)); } } self.close_panes(); @@ -277,8 +364,8 @@ impl Precompute { let Some(built) = builder.built_through else { continue; }; - let pane_ms = builder.definition.pane_ms; - let id = &builder.definition.id; + let pane_ms = builder.summary.definition.pane_ms; + let id = &builder.summary.definition.id; // The newest evaluation stored panes can serve is at the end of // the newest closed pane. Panes are kept for one extra pane // width beyond the plan's retention, so an evaluation one pane @@ -286,6 +373,7 @@ impl Precompute { // skip them. let newest_closed_end = watermark.div_euclid(pane_ms).saturating_mul(pane_ms); let oldest_kept_end = builder + .summary .definition .oldest_kept_end_ms(newest_closed_end.saturating_sub(pane_ms)); let Some(mut end) = built.checked_add(pane_ms) else { @@ -304,7 +392,12 @@ impl Precompute { // a pane without samples is stored as empty, not missing. while end <= watermark { let samples = builder.open.remove(&end).unwrap_or_default(); - match build_pane(builder, end, &samples) { + let scope = Scope::Ingestion { + window_start_ms: end - pane_ms, + window_end_ms: end, + revision: 1, + }; + match builder.summary.run(end, &samples, scope) { Ok(state) => { self.store.insert( SummaryInstanceKey::new(id.clone(), Selection::ending_at(end)), @@ -341,33 +434,6 @@ pub fn pane_end(timestamp_ms: i64, pane_ms: i64) -> Option { .checked_mul(pane_ms) } -/// Selects the pane's rows from `samples` with the plan's `Scan` matchers, -/// then runs the definition's build over them, for the pane ending at `end`. -fn build_pane( - builder: &Builder, - end: i64, - samples: &[RawSample], -) -> Result { - let scope = Scope::Ingestion { - window_start_ms: end - builder.definition.pane_ms, - window_end_ms: end, - revision: 1, - }; - let selected = builder - .selector - .window_batch(end, &executor_samples(samples))?; - let input: Source<'_> = Box::new(Operator::source( - builder.raw_schema.clone(), - vec![selected], - )?); - let build = builder - .build - .instantiate(BTreeMap::from([(builder.definition.raw_input(), input)]))?; - let rows = run_root(&build, builder.definition.builder, scope)?; - let schema = Arc::new(builder.definition.output_schema.clone()); - Ok(Batch::try_new(schema, rows)?) -} - /// `samples` as the executor's [`Sample`]s. The selector takes owned label /// maps, so each sample's labels are copied. pub fn executor_samples(samples: &[RawSample]) -> Vec { diff --git a/data_plane/src/query.rs b/data_plane/src/query.rs index d73012a8..3416a5d7 100644 --- a/data_plane/src/query.rs +++ b/data_plane/src/query.rs @@ -1,12 +1,20 @@ //! 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: -//! - stored outputs (the SDS frontier): bound to the pane of their definition -//! ending at `T - offset`, which must be stored; -//! - query-time bound raw-sample windows, `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`. +//! Each root is compiled once with three kinds of inputs: +//! - stored outputs built in an ingestion run: bound to their definition's +//! summary ending at `T - offset`, which must be stored; +//! - stored outputs built in a query run (`StoredOutputBuild::QueryRun`): +//! bound to their definition's summary ending at `T - offset`. A missing +//! one is built from its coverage's raw samples, which the caller fetches, +//! and stored in the SDS for later evaluations once its pane is closed +//! (its end at or before the watermark when the evaluation chose its +//! fetch); these summary states are evicted by the plan's `retention_ms` +//! plus one pane, measured from the newest closed pane at or before the +//! watermark, as for stored outputs built in an `IngestionRun`; +//! - the plan's query-time `RawSampleWindow`s: bound to the samples in +//! `(T - offset - width, T - offset]`, which the caller fetches, selected +//! by the window's `RawSampleSelector`. //! //! A query the engine cannot answer returns [`Unanswered`]; the caller then //! forwards it to Prometheus. @@ -23,10 +31,10 @@ use asap_executor::planner::ir::{NonASAPOp, Operator as IrOperator, ScalarExpr}; use asap_executor::runtime::Scope; use asap_executor::values::{Batch, SchemaRef, Value}; use asap_executor::{series_labels_of_output_row, RawSampleSelector}; -use asap_planner::InstallablePlan; +use asap_planner::{InstallablePlan, StoredOutputBuild}; -use crate::precompute::{executor_samples, run_root, RawSample}; -use crate::sds::{DefinitionId, Sds, Selection, SummaryStore}; +use crate::precompute::{executor_samples, run_root, RawSample, SummaryBuild}; +use crate::sds::{DefinitionId, Sds, Selection, SummaryInstanceKey, SummaryStore}; /// A bound raw-sample window: a `RawSampleWindow` of the plan bound to one /// evaluation time, i.e. `metric`'s raw samples in `(start_ms, end_ms]` @@ -112,14 +120,74 @@ struct RawInput { offset_ms: i64, } +/// A stored output built in a `QueryRun` (`MatAtQueryKeptAcrossEvaluations`). +struct QueryRunInput { + node: u64, + definition: DefinitionId, + offset_ms: i64, + /// The output's own `TimeRange`, which its coverage's samples feed. + raw_input: u64, +} + struct Program { compiled: CompiledPhysicalDAG, stored: Vec, + query_run: Vec, + /// For each `QueryRun` definition the root reads, the node its one + /// fetch is keyed by: the lowest `TimeRange` among its outputs. + query_run_fetch_node: BTreeMap, raw: Vec, output: Schema, } +/// What happened to the stored outputs built in a `QueryRun` evaluations read. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, serde::Serialize)] +pub struct QueryRunStats { + /// Summary states built from raw samples because the SDS had none. + pub built: u64, + /// Of those, the ones stored in the SDS: their coverage was closed. + pub stored: u64, + /// Of those stored, the ones whose own metric's newest sample had not + /// passed the coverage's end: the watermark is global, so a metric + /// lagging behind others can have a summary state stored before all its samples + /// arrived. A per-metric watermark is the follow-up. + pub stored_while_metric_behind: u64, + /// Summaries read from an earlier evaluation instead of rebuilt. + pub reused: u64, +} + +/// What ingestion had seen when an evaluation decided what to fetch. The +/// store decision uses this snapshot, not the state when the evaluation +/// runs: samples ingested in between were not in the fetched data. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct IngestionProgress { + /// The precompute engine's watermark; `None` before any sample. + pub watermark_ms: Option, + /// The newest sample time of each metric stored outputs built in a `QueryRun` read. + pub newest_sample_ms: BTreeMap, +} + +/// One raw read an evaluation needs, prepared by [`QueryEngine::bound_raw_sample_windows`]. +#[derive(Debug, Clone)] +pub struct Prepared { + /// The root answering the query. + pub root: usize, + /// The bound raw-sample windows to fetch. + pub windows: Vec, + /// What ingestion had seen when `ranges` were chosen. + pub progress: IngestionProgress, +} + pub struct QueryEngine { + /// How many summary states of stored outputs built in a `QueryRun` + /// evaluations built, stored and reused. + pub query_run_stats: QueryRunStats, + /// How each `QueryRun` definition's summaries are built, or why they + /// cannot be. + query_run_builds: BTreeMap>, + /// The label matchers of each `QueryRun` definition's `Scan`, pushed into + /// its fetch. + query_run_matchers: BTreeMap>, /// Each query, with its compiled root or why it cannot run. roots: Vec<(String, Result)>, } @@ -132,30 +200,56 @@ impl QueryEngine { .iter() .map(|w| (w.time_range_node_id as u64, w)) .collect(); + let query_run_builds: BTreeMap<_, _> = sds + .definitions + .values() + .filter(|d| d.built_at == StoredOutputBuild::QueryRun) + .map(|d| { + let build = SummaryBuild::new(dag, d).map_err(|e| e.to_string()); + (d.id.clone(), build) + }) + .collect(); let roots = plan .queries .iter() .map(|q| { - let program = program(dag, sds, &windows, q.root_node_id as u64); + let program = program(dag, sds, &query_run_builds, &windows, q.root_node_id as u64); (q.query.trim().to_string(), program) }) .collect(); - Self { roots } + let query_run_matchers = query_run_builds + .keys() + .map(|id| { + let scan = sds.definitions[id].window.scan_node_id as u64; + (id.clone(), label_matchers(dag, scan)) + }) + .collect(); + Self { + query_run_stats: QueryRunStats::default(), + query_run_builds, + query_run_matchers, + 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 windows, and for each `QueryRun` definition one range + /// covering every summary it reads that is not stored. Fails if the + /// query is not planned, a time does not fit, or an ingestion-built + /// summary it reads is missing. `progress` is what ingestion has seen + /// now; pass the returned copy to [`Self::evaluate`]. pub fn bound_raw_sample_windows( &self, query: &str, time_ms: i64, store: &SummaryStore, - ) -> Result<(usize, Vec), Unanswered> { + progress: IngestionProgress, + ) -> Result { let (root, program) = self.program(query)?; for input in &program.stored { pane(input, time_ms, store)?; } - let ranges = program + let mut ranges: Vec = program .raw .iter() .map(|input| { @@ -169,16 +263,53 @@ impl QueryEngine { }) }) .collect::>()?; - Ok((root, ranges)) + // One fetch per `QueryRun` definition, covering all its missing + // summaries. + let mut missing: BTreeMap<&DefinitionId, (i64, i64)> = BTreeMap::new(); + for input in &program.query_run { + let build = self.query_run_build(&input.definition); + let (start_ms, end_ms) = window(time_ms, input.offset_ms, build.definition.pane_ms)?; + if store + .get(&input.definition, Selection::ending_at(end_ms)) + .is_none() + { + let span = missing + .entry(&input.definition) + .or_insert((start_ms, end_ms)); + *span = (span.0.min(start_ms), span.1.max(end_ms)); + } + } + for (definition, (start_ms, end_ms)) in missing { + let build = self.query_run_build(definition); + ranges.push(BoundRawSampleWindow { + node: program.query_run_fetch_node[definition], + metric: build.metric().to_string(), + matchers: self.query_run_matchers[definition].clone(), + start_ms, + end_ms, + }); + } + Ok(Prepared { + root, + windows: ranges, + progress, + }) } - /// Runs `root` at `time_ms` over stored panes and `raw`, the samples of - /// each of its [`BoundRawSampleWindow`]s by node. + /// Runs `root` at `time_ms` over stored summaries and `raw`, the samples + /// of each of its [`BoundRawSampleWindow`]s by node. A stored output built in a `QueryRun` not stored + /// is built from its raw samples and stored in `store`, under the same + /// key as those built in an `IngestionRun`, if its coverage was closed in + /// `progress` (the snapshot [`Self::bound_raw_sample_windows`] chose the fetch by). + /// Then stored outputs built in a `QueryRun` older than their definition's retention, + /// measured from the newest closed pane at or before the watermark, are + /// evicted. pub fn evaluate( - &self, + &mut self, root: usize, time_ms: i64, - store: &SummaryStore, + store: &mut SummaryStore, + progress: &IngestionProgress, raw: &BTreeMap>, ) -> Result, Unanswered> { let program = self.roots[root] @@ -197,6 +328,70 @@ impl QueryEngine { Operator::source(pane.schema().clone(), vec![pane.clone()]).map_err(failed)?; sources.insert(input.node, Box::new(source)); } + for input in &program.query_run { + let build = self.query_run_builds[&input.definition] + .as_ref() + .expect("compiled roots read only `QueryRun` definitions that build"); + let (_, end_ms) = window(time_ms, input.offset_ms, build.definition.pane_ms)?; + let selection = Selection::ending_at(end_ms); + let summary = match store.get(&input.definition, selection) { + Some(summary) => { + self.query_run_stats.reused += 1; + summary.clone() + } + None => { + // `bound_raw_sample_windows` asked for these samples unless the + // summary was stored then and evicted since. + let samples = raw + .get(&program.query_run_fetch_node[&input.definition]) + .ok_or_else(|| { + Unanswered::MissingPane(format!( + "stored output built in a `QueryRun` ending at {end_ms} of {:?} was evicted", + input.definition + )) + })?; + // Built as at its coverage's end, so every evaluation + // that reads it reads the same summary. + let build_scope = Scope::Query { + evaluation_time_ms: end_ms, + revision: 1, + }; + let summary = build + .run(end_ms, samples, build_scope) + .map_err(|e| Unanswered::Execution(e.to_string()))?; + self.query_run_stats.built += 1; + // Stored only once closed, as the precompute engine closes + // panes: a sample may still arrive for an open one. + if progress.watermark_ms.is_some_and(|w| end_ms <= w) { + self.query_run_stats.stored += 1; + let metric_newest = progress.newest_sample_ms.get(build.metric()); + if metric_newest.is_none_or(|newest| *newest < end_ms) { + self.query_run_stats.stored_while_metric_behind += 1; + } + store.insert( + SummaryInstanceKey::new(input.definition.clone(), selection), + summary.clone(), + ); + } + summary + } + }; + let source = + Operator::source(summary.schema().clone(), vec![summary]).map_err(failed)?; + sources.insert(input.node, Box::new(source)); + } + if let Some(watermark) = progress.watermark_ms { + for build in self.query_run_builds.values().flatten() { + let definition = &build.definition; + let pane_ms = definition.pane_ms; + // As for stored outputs built in an `IngestionRun`: measured from the newest + // closed coverage, one pane of margin. + let newest_closed_end = watermark.div_euclid(pane_ms).saturating_mul(pane_ms); + let oldest = + definition.oldest_kept_end_ms(newest_closed_end.saturating_sub(pane_ms)); + store.evict_before(&definition.id, Selection::ending_at(oldest)); + } + } for input in &program.raw { let samples = raw.get(&input.node).map(Vec::as_slice).unwrap_or_default(); let rows = input @@ -214,6 +409,13 @@ impl QueryEngine { result_samples(&program.output, rows) } + /// The build of a `QueryRun` definition a compiled root reads. + fn query_run_build(&self, definition: &DefinitionId) -> &SummaryBuild { + self.query_run_builds[definition] + .as_ref() + .expect("compiled roots read only `QueryRun` definitions that build") + } + fn program(&self, query: &str) -> Result<(usize, &Program), Unanswered> { let query = query.trim(); let root = self @@ -265,6 +467,7 @@ fn pane<'a>( fn program( dag: &PhysicalASAPDAG, sds: &Sds, + query_run_builds: &BTreeMap>, windows: &HashMap, root: u64, ) -> Result { @@ -276,7 +479,7 @@ fn program( .or_default() .push(edge.producer as u64); } - let (mut stored, mut raw) = (Vec::new(), Vec::new()); + let (mut stored, mut query_run, mut raw) = (Vec::new(), Vec::new(), Vec::new()); let mut pending = vec![root]; let mut seen = BTreeSet::new(); while let Some(id) = pending.pop() { @@ -284,12 +487,25 @@ fn program( continue; } if let Some((definition, binding)) = sds.outputs.get(&id) { - stored.push(StoredInput { - node: id, - definition: definition.clone(), - offset_ms: binding.offset_ms, - pane_ms: sds.definitions[definition].pane_ms, - }); + match sds.definitions[definition].built_at { + StoredOutputBuild::IngestionRun => stored.push(StoredInput { + node: id, + definition: definition.clone(), + offset_ms: binding.offset_ms, + pane_ms: sds.definitions[definition].pane_ms, + }), + StoredOutputBuild::QueryRun => { + query_run_builds[definition] + .as_ref() + .map_err(Clone::clone)?; + query_run.push(QueryRunInput { + node: id, + definition: definition.clone(), + offset_ms: binding.offset_ms, + raw_input: binding.raw_input, + }); + } + } continue; } match &nodes[&id].payload { @@ -316,17 +532,30 @@ fn program( } } let mut inputs = BTreeMap::new(); - for input in &stored { - let schema = Arc::new(nodes[&input.node].output_schema.clone()); - inputs.insert(input.node, InputContract::bounded(schema)); + for node in stored + .iter() + .map(|i| i.node) + .chain(query_run.iter().map(|i| i.node)) + { + let schema = Arc::new(nodes[&node].output_schema.clone()); + inputs.insert(node, InputContract::bounded(schema)); } for input in &raw { inputs.insert(input.node, InputContract::bounded(input.schema.clone())); } let compiled = compile(dag, inputs, &[root]).map_err(|e| e.to_string())?; + let mut query_run_fetch_node: BTreeMap = BTreeMap::new(); + for input in &query_run { + let node = query_run_fetch_node + .entry(input.definition.clone()) + .or_insert(input.raw_input); + *node = (*node).min(input.raw_input); + } Ok(Program { compiled, stored, + query_run, + query_run_fetch_node, raw, output: nodes[&root].output_schema.clone(), }) diff --git a/data_plane/src/sds.rs b/data_plane/src/sds.rs index 184f5f2c..299c7528 100644 --- a/data_plane/src/sds.rs +++ b/data_plane/src/sds.rs @@ -86,6 +86,9 @@ impl SummaryDefinition { #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct PaneBinding { pub offset_ms: i64, + /// The output's own `TimeRange` node: where its coverage's raw samples + /// feed when a stored output built in a `QueryRun` is built for it. + pub raw_input: u64, } /// The stored summaries of one plan. @@ -154,6 +157,7 @@ impl Sds { definition.retention_ms = definition.retention_ms.max(retention_ms); let binding = PaneBinding { offset_ms: output.pane.end_offset_ms, + raw_input: output.pane.time_range_node_id as u64, }; outputs.insert(node, (id, binding)); } diff --git a/data_plane/tests/precompute.rs b/data_plane/tests/precompute.rs index 72cb9269..004655fc 100644 --- a/data_plane/tests/precompute.rs +++ b/data_plane/tests/precompute.rs @@ -432,7 +432,7 @@ fn does_not_build_query_run_outputs() { let (mut engine, _) = precompute(plan); engine.ingest(&samples("latency_ms", &["a"], 0, 45 * MINUTE)); assert_eq!(engine.stats.panes_built, 0); - assert_eq!(engine.stats.samples_unrouted, 180); + assert_eq!(engine.stats.samples_query_run_only, 180); } /// Retention is measured from the newest evaluation the stored panes can @@ -473,3 +473,31 @@ fn keeps_one_pane_of_margin_beyond_the_retention() { // the one more the evaluation at 280 reads. assert_eq!(engine.store.len(), 7); } + +/// Samples of a metric only stored outputs built in a `QueryRun` read (B3) are not buffered +/// but move the watermark, which says when those summaries are closed. +#[test] +fn query_run_metrics_move_the_watermark() { + let (mut engine, _) = precompute_with(common::plan("example4-b3", true), 0); + assert_eq!(engine.watermark_ms(), None); + engine.ingest(&samples("latency_ms", &["a"], 0, 3 * MINUTE)); + assert_eq!(engine.stats.samples_query_run_only, 12); + assert_eq!(engine.stats.samples_accepted, 0); + assert_eq!(engine.watermark_ms(), Some(3 * MINUTE - 15_000)); +} + +/// Samples of a metric only `QueryRun` stored outputs read get the same future bound and +/// wall-clock cap as routed samples: one far ahead is dropped, and one +/// ahead of the clock but within the bound moves the watermark only to the +/// clock. +#[test] +fn query_run_metrics_are_bounded_like_routed_samples() { + let (mut engine, _) = precompute_with(common::plan("example4-b3", true), 0); + let now = 10 * MINUTE; + engine.ingest_at(&[sample("latency_ms", "a", 30 * DAY, 1.0)], now); + assert_eq!(engine.stats.samples_dropped_future, 1); + assert_eq!(engine.watermark_ms(), None); + engine.ingest_at(&[sample("latency_ms", "a", now + 2 * MINUTE, 1.0)], now); + assert_eq!(engine.stats.samples_query_run_only, 1); + assert_eq!(engine.watermark_ms(), Some(now)); +} diff --git a/data_plane/tests/query.rs b/data_plane/tests/query.rs index ae9b698e..2e12f204 100644 --- a/data_plane/tests/query.rs +++ b/data_plane/tests/query.rs @@ -81,14 +81,21 @@ fn fetch(ranges: &[BoundRawSampleWindow], all: &[RawSample]) -> BTreeMap Vec { - let (root, ranges) = f +fn answer(f: &mut Fixture, query: &str, time_ms: i64, all: &[RawSample]) -> Vec { + let prepared = f .query - .bound_raw_sample_windows(query, time_ms, &f.precompute.store) + .bound_raw_sample_windows(query, time_ms, &f.precompute.store, f.precompute.progress()) .unwrap(); + let raw = fetch(&prepared.windows, all); let mut result = f .query - .evaluate(root, time_ms, &f.precompute.store, &fetch(&ranges, all)) + .evaluate( + prepared.root, + time_ms, + &mut f.precompute.store, + &prepared.progress, + &raw, + ) .unwrap(); result.sort_by(|a, b| a.labels.cmp(&b.labels)); result @@ -112,23 +119,40 @@ fn stored_panes_answer_like_a_single_build() { assert!( panes .query - .bound_raw_sample_windows(P99, 70 * MINUTE, &panes.precompute.store) + .bound_raw_sample_windows( + P99, + 70 * MINUTE, + &panes.precompute.store, + Default::default() + ) .is_err(), "no panes before ingestion" ); panes.precompute.ingest(&all); - let (_, ranges) = panes + let ranges = panes .query - .bound_raw_sample_windows(P99, 70 * MINUTE, &panes.precompute.store) - .unwrap(); + .bound_raw_sample_windows( + P99, + 70 * MINUTE, + &panes.precompute.store, + Default::default(), + ) + .unwrap() + .windows; 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 (_, ranges) = single + let mut single = fixture(common::plan("example4b", true)); + let ranges = single .query - .bound_raw_sample_windows(P99, 70 * MINUTE, &single.precompute.store) - .unwrap(); + .bound_raw_sample_windows( + P99, + 70 * MINUTE, + &single.precompute.store, + Default::default(), + ) + .unwrap() + .windows; assert_eq!( ranges .iter() @@ -137,7 +161,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); @@ -159,9 +183,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, @@ -177,7 +201,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, @@ -196,11 +220,13 @@ fn reports_what_it_cannot_answer() { let mut f = fixture(common::plan("example4b", false)); let store = &f.precompute.store; assert!(matches!( - f.query.bound_raw_sample_windows("up", 70 * MINUTE, store), + f.query + .bound_raw_sample_windows("up", 70 * MINUTE, store, Default::default()), Err(Unanswered::NotPlanned(_)) )); assert!(matches!( - f.query.bound_raw_sample_windows(P99, 70 * MINUTE, store), + f.query + .bound_raw_sample_windows(P99, 70 * MINUTE, store, Default::default()), Err(Unanswered::MissingPane(_)) )); let all = samples(&[("latency_ms", &[])], MINUTE, 75 * MINUTE, |_, t| t as f64); @@ -208,16 +234,17 @@ fn reports_what_it_cannot_answer() { let store = &f.precompute.store; assert!(f .query - .bound_raw_sample_windows(P99, 70 * MINUTE, store) + .bound_raw_sample_windows(P99, 70 * MINUTE, store, Default::default()) .is_ok()); assert!(matches!( f.query - .bound_raw_sample_windows(P99, 70 * MINUTE + 1, store), + .bound_raw_sample_windows(P99, 70 * MINUTE + 1, store, Default::default()), Err(Unanswered::MissingPane(_)) )); // The pane ending at 10 min was skipped as possibly partial. assert!(matches!( - f.query.bound_raw_sample_windows(P99, 60 * MINUTE, store), + f.query + .bound_raw_sample_windows(P99, 60 * MINUTE, store, Default::default()), Err(Unanswered::MissingPane(_)) )); } @@ -230,20 +257,23 @@ fn extreme_evaluation_times_are_not_answered() { for time_ms in [i64::MIN, i64::MIN + 1] { assert!(matches!( f.query - .bound_raw_sample_windows(P99, time_ms, &f.precompute.store), + .bound_raw_sample_windows(P99, time_ms, &f.precompute.store, Default::default()), Err(Unanswered::Unsupported(_)) )); } // The window ending at `i64::MAX` fits; binding it does not panic. assert!(f .query - .bound_raw_sample_windows(P99, i64::MAX, &f.precompute.store) + .bound_raw_sample_windows(P99, i64::MAX, &f.precompute.store, Default::default()) .is_ok()); let panes = fixture(common::plan("example4b", false)); assert!(matches!( - panes - .query - .bound_raw_sample_windows(P99, i64::MIN, &panes.precompute.store), + panes.query.bound_raw_sample_windows( + P99, + i64::MIN, + &panes.precompute.store, + Default::default() + ), Err(Unanswered::Unsupported(_)) )); } @@ -276,12 +306,12 @@ fn examples_answer_like_the_query_time_reference() { let reference = plan.query_time_reference().unwrap(); let mut installed = fixture(plan.clone()); installed.precompute.ingest(&all); - let by_reference = fixture(reference); + let mut by_reference = fixture(reference); for query in plan.queries.iter().map(|q| q.query.as_str()) { - let expected = answer(&by_reference, query, 70 * MINUTE, &all); + let expected = answer(&mut by_reference, query, 70 * MINUTE, &all); assert!(!expected.is_empty(), "{example} raw={raw} {query}"); assert_eq!( - answer(&installed, query, 70 * MINUTE, &all), + answer(&mut installed, query, 70 * MINUTE, &all), expected, "{example} raw={raw} {query}" ); @@ -300,14 +330,16 @@ fn bound_raw_sample_windows_carry_the_scans_label_matchers() { ); let plan = control_plane::plan_workload(&workload, &common::inputs(true)).unwrap(); let f = fixture(plan); - let (_, ranges) = f + let ranges = f .query .bound_raw_sample_windows( "quantile_over_time(0.99, latency_ms{job=\"a\"}[60m])", 70 * MINUTE, &f.precompute.store, + Default::default(), ) - .unwrap(); + .unwrap() + .windows; assert_eq!(ranges.len(), 1); assert_eq!( ranges[0].matchers, diff --git a/data_plane/tests/query_run.rs b/data_plane/tests/query_run.rs new file mode 100644 index 00000000..85ececb0 --- /dev/null +++ b/data_plane/tests/query_run.rs @@ -0,0 +1,381 @@ +//! Stored outputs built in a `QueryRun` (`MatAtQueryKeptAcrossEvaluations`): +//! #509 Example 4's B3, five 1-minute KLL panes built at query time and +//! stored in the SDS 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::values::Batch; +use asap_planner::{InstallablePlan, StoredOutputBuild}; +use data_plane::engine::{EngineHandle, EngineOptions}; +use data_plane::precompute::RawSample; +use data_plane::query::{ + BoundRawSampleWindow, IngestionProgress, QueryEngine, QueryRunStats, ResultSample, +}; +use data_plane::sds::{DefinitionId, Sds, Selection, SummaryStore}; + +const MINUTE: i64 = 60_000; +const P99: &str = "quantile_over_time(0.99, latency_ms[5m])"; + +/// A query engine and the SDS store its stored outputs built in a `QueryRun` go to. +struct Fixture { + engine: QueryEngine, + store: SummaryStore, + definition: DefinitionId, +} + +fn fixture(plan: &InstallablePlan) -> Fixture { + let sds = Sds::from_plan(plan).unwrap(); + assert!( + sds.ingestion_build_order().unwrap().is_empty(), + "nothing is built at ingestion" + ); + let definition = sds + .definitions + .values() + .find(|d| d.built_at == StoredOutputBuild::QueryRun) + .map(|d| d.id.clone()) + .unwrap_or_else(|| DefinitionId(String::new())); + Fixture { + engine: QueryEngine::new(plan, &sds), + store: SummaryStore::new(), + definition, + } +} + +impl Fixture { + /// The ends, in minutes, of the stored stored outputs built in a `QueryRun`. + fn ends(&self) -> Vec { + (0..=30) + .map(|m| m * MINUTE) + .filter(|end| { + self.store + .get(&self.definition, Selection::ending_at(*end)) + .is_some() + }) + .map(|end| end / MINUTE) + .collect() + } +} + +/// 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(|job| { + Arc::new(BTreeMap::from([ + ("__name__".to_string(), "latency_ms".to_string()), + ("job".to_string(), job.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: &[BoundRawSampleWindow], 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` with ingestion's watermark at `watermark_ms` +/// (every metric's newest sample there too), returning the answer and the +/// bound raw-sample windows it read. +fn answer_at_watermark( + k: &mut Fixture, + time_ms: i64, + watermark_ms: Option, + all: &[RawSample], +) -> (Vec, Vec) { + let progress = IngestionProgress { + watermark_ms, + newest_sample_ms: watermark_ms + .map(|w| BTreeMap::from([("latency_ms".to_string(), w)])) + .unwrap_or_default(), + }; + let prepared = k + .engine + .bound_raw_sample_windows(P99, time_ms, &k.store, progress) + .unwrap(); + let ranges = prepared.windows.clone(); + let mut result = k + .engine + .evaluate( + prepared.root, + time_ms, + &mut k.store, + &prepared.progress, + &fetch(&ranges, all), + ) + .unwrap(); + result.sort_by(|a, b| a.labels.cmp(&b.labels)); + (result, ranges) +} + +/// Answers `P99` at `time_ms` once every sample through `time_ms` has +/// arrived: the watermark is `time_ms`. +fn answer( + k: &mut Fixture, + time_ms: i64, + all: &[RawSample], +) -> (Vec, Vec) { + answer_at_watermark(k, time_ms, Some(time_ms), all) +} + +fn spans(ranges: &[BoundRawSampleWindow]) -> Vec<(i64, i64)> { + ranges + .iter() + .map(|r| (r.start_ms / MINUTE, r.end_ms / MINUTE)) + .collect() +} + +/// The installed B3 plan and the Planner's reference for it: the same +/// selection with nothing materialized. +fn plans() -> (InstallablePlan, InstallablePlan) { + let plan = common::plan("example4-b3", true); + let reference = plan.query_time_reference().unwrap(); + (plan, reference) +} + +/// The first evaluation builds all five summaries from one fetch and +/// answers like the reference; the next one, a minute later, fetches only +/// the newest minute, builds one summary, reuses four, and still answers +/// like the reference. +#[test] +fn query_run_outputs_answer_like_the_reference_and_are_reused() { + let (plan, reference_plan) = plans(); + let all = samples(); + let mut k = fixture(&plan); + let mut reference = fixture(&reference_plan); + + let (first, ranges) = answer(&mut k, 10 * MINUTE, &all); + assert_eq!( + spans(&ranges), + vec![(5, 10)], + "one fetch for five summaries" + ); + assert_eq!(first, answer(&mut reference, 10 * MINUTE, &all).0); + assert_eq!(first.len(), 2); + assert_eq!( + k.engine.query_run_stats, + QueryRunStats { + built: 5, + stored: 5, + stored_while_metric_behind: 0, + reused: 0 + } + ); + + let (second, ranges) = answer(&mut k, 11 * MINUTE, &all); + assert_eq!(spans(&ranges), vec![(10, 11)], "only the newest summary"); + assert_eq!(second, answer(&mut reference, 11 * MINUTE, &all).0); + assert_ne!(second, first, "the answer moves with the window"); + assert_eq!( + ( + k.engine.query_run_stats.built, + k.engine.query_run_stats.reused + ), + (6, 4) + ); + + // The same evaluation again builds nothing. + let (again, ranges) = answer(&mut k, 11 * MINUTE, &all); + assert!(ranges.is_empty()); + assert_eq!(again, second); + assert_eq!( + ( + k.engine.query_run_stats.built, + k.engine.query_run_stats.reused + ), + (6, 9) + ); + assert_eq!( + reference.engine.query_run_stats, + QueryRunStats::default(), + "the reference stores nothing" + ); +} + +/// Stored stored outputs built in a `QueryRun` are evicted by the plan's retention (5 min, +/// the oldest summary a 5-minute window reads), plus one pane, measured +/// from the newest closed coverage at or before the watermark, as for +/// stored outputs built in an `IngestionRun`. A far-future evaluation does not move that +/// point: its summary states are built but not stored, and storing resumes. +#[test] +fn query_run_outputs_are_evicted_by_the_plans_retention() { + let (plan, _) = plans(); + let retention = plan.stored_outputs.iter().map(|o| o.retention_ms).max(); + assert_eq!(retention, Some(5 * MINUTE as u64)); + let all = samples(); + let mut k = fixture(&plan); + answer(&mut k, 10 * MINUTE, &all); + assert_eq!(k.ends(), vec![6, 7, 8, 9, 10]); + answer(&mut k, 11 * MINUTE, &all); + assert_eq!(k.ends(), vec![6, 7, 8, 9, 10, 11]); + answer(&mut k, 12 * MINUTE, &all); + assert_eq!(k.ends(), vec![7, 8, 9, 10, 11, 12]); + // An evaluation far ahead of the watermark is answered, but its + // summaries' coverages are not closed, so nothing is stored or evicted. + let (far, ranges) = answer_at_watermark(&mut k, 20 * MINUTE, Some(12 * MINUTE), &all); + assert_eq!(far.len(), 2); + assert_eq!(spans(&ranges), vec![(15, 20)]); + assert_eq!(k.ends(), vec![7, 8, 9, 10, 11, 12]); + // Keeping resumes with the watermark. + answer(&mut k, 13 * MINUTE, &all); + assert_eq!(k.ends(), vec![8, 9, 10, 11, 12, 13]); +} + +/// A summary state is stored only once its coverage is closed (its end at or +/// before the watermark). One read before then is built for that +/// evaluation but not stored, so a later evaluation builds it again and +/// includes a sample that arrived late. +#[test] +fn query_run_outputs_are_stored_only_once_closed() { + let (plan, reference_plan) = plans(); + let mut all = samples(); + let mut k = fixture(&plan); + let mut reference = fixture(&reference_plan); + + // Watermark at 9.5 min: the summary ending at 10 min is still open. + let (first, ranges) = answer_at_watermark(&mut k, 10 * MINUTE, Some(570_000), &all); + assert_eq!(spans(&ranges), vec![(5, 10)]); + assert_eq!(first, answer(&mut reference, 10 * MINUTE, &all).0); + assert_eq!( + k.ends(), + vec![6, 7, 8, 9], + "the open summary state is not stored" + ); + + // A late sample for the open coverage arrives before it closes. + let late = RawSample { + timestamp_ms: 10 * MINUTE - 1_000, + value: 1_000.0, + ..all.iter().find(|s| s.labels["job"] == "a").unwrap().clone() + }; + all.push(late); + let (again, ranges) = answer_at_watermark(&mut k, 10 * MINUTE, Some(585_000), &all); + assert_eq!( + spans(&ranges), + vec![(9, 10)], + "the open summary is built 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 summary state is stored. + answer_at_watermark(&mut k, 10 * MINUTE, Some(10 * MINUTE), &all); + assert_eq!(k.ends(), vec![6, 7, 8, 9, 10]); + // Without any sample there is no watermark, and nothing is stored. + let mut fresh = fixture(&plan); + answer_at_watermark(&mut fresh, 10 * MINUTE, None, &all); + assert!(fresh.store.is_empty()); +} + +/// The store decision uses the watermark when the evaluation chose what to +/// fetch, not when it runs: ingestion that closes a coverage after the +/// fetch was chosen does not make the partial summary built from that +/// fetch permanent. +#[tokio::test] +async fn stores_by_the_watermark_the_fetch_was_chosen_at() { + let (plan, _) = plans(); + let options = EngineOptions { + precompute: data_plane::precompute::PrecomputeOptions { + allowed_lateness_ms: 0, + ..Default::default() + }, + ..EngineOptions::default() + }; + let (engine, _) = EngineHandle::spawn(serde_json::to_string(&plan).unwrap(), options).unwrap(); + let all = samples(); + let through = |t: i64| -> Vec { + all.iter() + .filter(|s| s.timestamp_ms <= t) + .cloned() + .collect() + }; + let settle = |engine: &EngineHandle| { + let engine = engine.clone(); + async move { + while engine.queued_samples() > 0 { + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + } + }; + engine.try_ingest(through(570_000)).unwrap(); + settle(&engine).await; + let prepared = engine + .bound_raw_sample_windows(P99.into(), 10 * MINUTE) + .await + .unwrap(); + assert_eq!(prepared.progress.watermark_ms, Some(570_000)); + // The fetch sees only what had arrived; then ingestion closes 10 min. + let raw = fetch(&prepared.windows, &through(570_000)); + engine.try_ingest(through(11 * MINUTE)).unwrap(); + settle(&engine).await; + engine.evaluate(prepared, 10 * MINUTE, raw).await.unwrap(); + assert_eq!(engine.query_run_stats().stored, 4, "6..9 stored, 10 not"); + // So the next evaluation builds the summary ending at 10 min again. + let prepared = engine + .bound_raw_sample_windows(P99.into(), 10 * MINUTE) + .await + .unwrap(); + assert_eq!(spans(&prepared.windows), vec![(9, 10)]); +} + +/// A summary state stored while its own metric's newest sample has not reached its +/// end (the global watermark moved on another metric) is counted. +#[test] +fn counts_outputs_stored_while_their_metric_is_behind() { + let (plan, _) = plans(); + let all = samples(); + let mut k = fixture(&plan); + let progress = IngestionProgress { + watermark_ms: Some(10 * MINUTE), + newest_sample_ms: BTreeMap::from([("latency_ms".to_string(), 8 * MINUTE)]), + }; + let prepared = k + .engine + .bound_raw_sample_windows(P99, 10 * MINUTE, &k.store, progress) + .unwrap(); + let raw = fetch(&prepared.windows, &all); + k.engine + .evaluate( + prepared.root, + 10 * MINUTE, + &mut k.store, + &prepared.progress, + &raw, + ) + .unwrap(); + assert_eq!(k.engine.query_run_stats.stored, 5); + assert_eq!( + k.engine.query_run_stats.stored_while_metric_behind, 2, + "9 and 10" + ); +} diff --git a/data_plane/tests/sds.rs b/data_plane/tests/sds.rs index 53019fd4..f4663089 100644 --- a/data_plane/tests/sds.rs +++ b/data_plane/tests/sds.rs @@ -116,7 +116,7 @@ fn identity_depends_on_the_source() { ); } -/// Outputs built in a query run and kept are stored summaries too: their +/// Stored outputs built in a `QueryRun` are stored summaries too: their /// definition says so, and no ingestion run builds them. #[test] fn query_run_outputs_have_query_run_definitions() { @@ -251,3 +251,27 @@ fn build_order_refuses_ingestion_reading_query_run() { }) ); } + +/// B3's five 1-minute KLL panes, stored outputs built in a `QueryRun` +/// (`MatAtQueryKeptAcrossEvaluations`), are one `QueryRun` definition, each output reading the +/// summary `i * 1min` before evaluation. Nothing is built at ingestion. +#[test] +fn query_run_outputs_share_one_definition() { + let sds = Sds::from_plan(&plan("example4-b3", true)).unwrap(); + assert!(sds.ingestion_build_order().unwrap().is_empty()); + let mut query_run = sds + .definitions + .values() + .filter(|d| d.built_at == StoredOutputBuild::QueryRun); + let definition = query_run.next().unwrap(); + assert!(query_run.next().is_none()); + assert_eq!(definition.pane_ms, 60_000); + let mut offsets: Vec = sds + .outputs + .values() + .filter(|(id, _)| id == &definition.id) + .map(|(_, binding)| binding.offset_ms) + .collect(); + offsets.sort(); + assert_eq!(offsets, (0..5).map(|i| i * 60_000).collect::>()); +} 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 + } + } + } +}