diff --git a/README.md b/README.md index 8d11d1d4..9ebd9ae3 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,14 @@ 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 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. @@ -61,8 +69,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 +77,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..e1e173c3 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); @@ -172,6 +174,7 @@ impl Engine { root, time_ms, &self.precompute.store, + self.precompute.watermark_ms(), &raw, )); } 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..abfb539c 100644 --- a/data_plane/src/query.rs +++ b/data_plane/src/query.rs @@ -1,9 +1,15 @@ //! 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 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`. @@ -26,7 +32,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 +71,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 +89,31 @@ 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 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, +} + 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 +122,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 +163,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,16 +173,33 @@ 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 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( - &self, + &mut self, root: usize, time_ms: i64, store: &SummaryStore, + watermark_ms: Option, raw: &BTreeMap>, ) -> Result, Unanswered> { let program = self.roots[root] @@ -147,17 +218,73 @@ 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_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 + } + }; + 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 +320,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 +335,7 @@ fn program(dag: &PhysicalASAPDAG, sds: &Sds, root: u64) -> Result Result { raw.push(raw_input( @@ -237,9 +379,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..c5397868 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 @@ -97,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 { @@ -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,42 @@ 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. Samples run to 13 min, +/// so every pane read is closed and kept. +#[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..1c82d46a --- /dev/null +++ b/data_plane/tests/kept.rs @@ -0,0 +1,265 @@ +//! 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` 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, watermark_ms, &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); +} + +/// 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 04af1d49..95b7def2 100644 --- a/data_plane/tests/query.rs +++ b/data_plane/tests/query.rs @@ -77,14 +77,20 @@ 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) .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 @@ -118,9 +124,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 +139,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 +161,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 +179,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 + } + } + } +}