diff --git a/crates/integration-tests/tests/reuse_rules.rs b/crates/integration-tests/tests/reuse_rules.rs index 38001bcaa..1d2cffa72 100644 --- a/crates/integration-tests/tests/reuse_rules.rs +++ b/crates/integration-tests/tests/reuse_rules.rs @@ -112,7 +112,10 @@ fn samples() -> Vec<(&'static str, &'static str, i64, f64)> { rows } -/// Execute `root` at query time, every raw `TimeRange` input fed `samples`. +/// Execute `root` at query time, every raw `TimeRange` input fed the +/// `samples` in its window: `(end - range, end]`, where `end` is the +/// evaluation time moved back by a `TimeShift` directly below (a tumbling +/// pane's). fn execute(root: &Rc) -> Vec { let timed = apply_materialization_timings( root, @@ -120,21 +123,32 @@ fn execute(root: &Rc) -> Vec { &mut TimingMemo::new(), ) .unwrap(); + // Tumbling panes run as their panes, each over its own window. + let timed = asap_types::ir::operator::pane_chain::expand_pane_chains(&timed).unwrap(); let compiled = compile_physical_asap_dag_with_node_ids(&timed).unwrap(); let dag = compiled.dag; let inputs: BTreeMap = dag .nodes .iter() - .filter(|node| { - matches!( - compiled.node_ids.operator_node(node.id).unwrap().operator, - Operator::NonASAP(NonASAPOp::TimeRange { .. }) - ) + .filter_map(|node| { + let Operator::NonASAP(NonASAPOp::TimeRange { range, child, .. }) = + &compiled.node_ids.operator_node(node.id).unwrap().operator + else { + return None; + }; + let back_ms = match child.non_asap() { + Some(NonASAPOp::TimeShift { shift, .. }) => shift.offset_ms, + _ => 0, + }; + let end = EVALUATION_MS - back_ms; + let start = end - range.as_millis() as i64; + Some((node, start, end)) }) - .map(|node| { + .map(|(node, start, end)| { let schema: Arc = Arc::new(node.output_schema.clone()); let rows = samples() .into_iter() + .filter(|&(_, _, ts, _)| start < ts && ts <= end) .map(|(job, series, ts, value)| { let labels = BTreeMap::from([ ("__name__".to_string(), "m".to_string()), diff --git a/crates/logical-asap-optimization/src/asap_aware_cse.rs b/crates/logical-asap-optimization/src/asap_aware_cse.rs index a9ce3c234..6665727fe 100644 --- a/crates/logical-asap-optimization/src/asap_aware_cse.rs +++ b/crates/logical-asap-optimization/src/asap_aware_cse.rs @@ -29,11 +29,16 @@ //! //! Composition sizes each consumer's summary build alone, for its own //! accuracy target. The *members* of a [`CapabilitySharingCandidate`] are summary -//! builds of one candidate that chose the same family (same -//! [`SketchAlgorithm`]) over the same input sub-DAG (window, `WHERE` and -//! time range included), item and weight, grouping and filter, and whose -//! every reader is an estimate the family answers -//! ([`FamilyCapability::answers`]); they differ only in their sizes. +//! builds of one candidate that chose the same family (the same +//! [`SketchAlgorithm`](asap_types::ir::schema::SketchAlgorithm) at any +//! size, or the same per-series exact accumulator) over the same input +//! sub-DAG (window, `WHERE` and time range included), item and weight, +//! grouping and filter, and whose every reader is an estimate the family +//! answers ([`FamilyCapability::answers`]) or, for an exact accumulator, +//! its finalization. They differ in their sizes, or, for per-series builds +//! over one metric's scans without label matchers, in the labels those +//! scans project: the shared build is over the scan with every member's +//! labels, and each other member's readers keep only their own columns. //! [`summary_capability_pass2_candidates`] returns one Pass 2 candidate per //! choice, for each capability sharing candidate, of no sharing or one //! subset of two or more of its members @@ -104,8 +109,7 @@ use asap_types::ir::cse::share_common_sub_dags; use asap_types::ir::operator::operator_properties::{Reduction, Source}; use asap_types::ir::operator::AggIntent; use asap_types::ir::schema::{ - ExactKind, FamilyCapability, FieldDataType, GroupingStrategy, MergeInputRule, SketchAlgorithm, - SketchKind, + ExactKind, FamilyCapability, FieldDataType, GroupingStrategy, MergeInputRule, SketchKind, }; use asap_types::ir::SortKey; use asap_types::ir::{ @@ -354,13 +358,16 @@ impl Sharing<'_> { } /// Summary builds the summary-capability rule can make one build: builds -/// of one family over the same input data that differ only in size, in the -/// order a children-first walk of the roots meets them. +/// of one family over the same input data that differ only in size, or in +/// the labels their per-series scans project, in the order a children-first +/// walk of the roots meets them. #[derive(Debug, Clone)] pub struct CapabilitySharingCandidate { - /// The sketch family every member builds. - pub family: SketchAlgorithm, - /// The summary builds (`SummaryAgg`), at least two, not all of one size. + /// The family the first member builds: every member builds this sketch + /// algorithm at some size, or this exact accumulator. + pub family: FieldDataType, + /// The summary builds (`SummaryAgg`), at least two, not all equal (in + /// size or in the labels their scans project). pub members: Vec>, } @@ -400,7 +407,7 @@ pub fn capability_sharing_candidates(roots: &[QueryRoot]) -> Vec candidate.members.push(Rc::clone(node)), None => candidates.push(CapabilitySharingCandidate { @@ -428,34 +435,59 @@ fn collect_children_first( } /// The family of a per-subpopulation sketch build whose every reader is an -/// estimate the family answers; `None` for any other node. +/// estimate the family answers, or of a per-series exact accumulator whose +/// every reader finalizes it; `None` for any other node. fn shareable_family( node: &OperatorNode, readers: Option<&Vec<&Rc>>, -) -> Option { - let Some(ASAPOp::SummaryAgg { - family: family @ FieldDataType::Sketch(kind, GroupingStrategy::PerSubpopulationInstance), - .. - }) = node.asap() - else { - return None; - }; - let capability = FamilyCapability::of(family); - let answered = |reader: &&Rc| { - matches!( - reader.asap(), - Some(ASAPOp::SummaryEstimate { query, .. }) if capability.answers(query) - ) - }; +) -> Option { let readers = readers?; - readers - .iter() - .all(answered) - .then(|| kind.algorithm().clone()) + match node.asap() { + Some(ASAPOp::SummaryAgg { + family: family @ FieldDataType::Sketch(_, GroupingStrategy::PerSubpopulationInstance), + .. + }) => { + let capability = FamilyCapability::of(family); + readers + .iter() + .all(|reader| { + matches!( + reader.asap(), + Some(ASAPOp::SummaryEstimate { query, .. }) if capability.answers(query) + ) + }) + .then(|| family.clone()) + } + Some(ASAPOp::SummaryAgg { + family: family @ FieldDataType::ExactAggregate(..), + reduction: Reduction::PerEntity, + .. + }) => readers + .iter() + .all(|reader| matches!(reader.asap(), Some(ASAPOp::FinalizeExactAccumulator { .. }))) + .then(|| family.clone()), + _ => None, + } +} + +/// Whether builds of families `a` and `b` can be one build: one sketch +/// algorithm and grouping at any sizes, or one exact accumulator. +fn same_family(a: &FieldDataType, b: &FieldDataType) -> bool { + match (a, b) { + (FieldDataType::Sketch(a, grouping), FieldDataType::Sketch(b, grouping_b)) => { + a.algorithm() == b.algorithm() && grouping == grouping_b + } + _ => a == b, + } } /// Whether two builds of one family read the same data the same way: the -/// same input sub-DAG, item and weight, grouping and filter. +/// same input sub-DAG, item and weight, grouping and filter. Two +/// per-series builds over one metric's samples also do when their scans +/// project different labels ([`same_rows`]): a per-series state is keyed by +/// the series identity, which determines every label, so a build over the +/// scan with the most labels holds every member's states and its readers +/// keep the labels each needs ([`share_summaries`]). fn same_input_data(a: &OperatorNode, b: &OperatorNode) -> bool { match (a.asap(), b.asap()) { ( @@ -476,7 +508,14 @@ fn same_input_data(a: &OperatorNode, b: &OperatorNode) -> bool { .. }), ) => { - (Rc::ptr_eq(child, child_b) || child == child_b) + let same_child = Rc::ptr_eq(child, child_b) + || child == child_b + || (*reduction == Reduction::PerEntity + && filter.is_none() + && child.schema.has_promql_series_identity() + && child_b.schema.has_promql_series_identity() + && same_rows(child, child_b)); + same_child && input == input_b && reduction == reduction_b && grouping == grouping_b @@ -486,6 +525,20 @@ fn same_input_data(a: &OperatorNode, b: &OperatorNode) -> bool { } } +/// The names of the columns of a build's input: for a per-series build, +/// the labels its scan projects, among others. +fn input_column_names(build: &OperatorNode) -> Vec<&str> { + match build.asap() { + Some(ASAPOp::SummaryAgg { child, .. }) => child + .schema + .fields + .iter() + .map(|f| f.name.as_str()) + .collect(), + _ => Vec::new(), + } +} + fn all_equal<'a>(mut nodes: impl Iterator>) -> bool { let Some(first) = nodes.next() else { return true; @@ -594,13 +647,49 @@ fn share_summaries( shared_builds: &[Vec>], ) -> Result>, PerComputationCandidateError> { let mut replaced: HashMap<*const OperatorNode, Rc> = HashMap::new(); + // Every node that reads each build, by the build's address. + let mut nodes = Vec::new(); + let mut seen = std::collections::HashSet::new(); + for node in roots.iter().flat_map(root_operators) { + collect_children_first(node, &mut seen, &mut nodes); + } + let mut readers: HashMap<*const OperatorNode, Vec<&Rc>> = HashMap::new(); + for node in &nodes { + for child in node.children() { + readers.entry(Rc::as_ptr(child)).or_default().push(node); + } + } for members in shared_builds { - let members: Vec<&Rc> = members.iter().collect(); + let mut members: Vec<&Rc> = members.iter().collect(); + // The build is over the input with every member's columns (the + // scan with the most labels); none such, no shared build. + let Some(host) = members.iter().position(|candidate| { + let columns = input_column_names(candidate); + members.iter().all(|member| { + input_column_names(member) + .iter() + .all(|c| columns.contains(c)) + }) + }) else { + return Ok(None); + }; + members.swap(0, host); let Some(build) = sized_for_all(&members)? else { return Ok(None); }; for member in members { - replaced.insert(Rc::as_ptr(member), Rc::clone(&build)); + if member.schema == build.schema { + replaced.insert(Rc::as_ptr(member), Rc::clone(&build)); + continue; + } + // A member over fewer labels: each of its estimates reads the + // shared build and keeps the member's columns, by name. + for reader in readers.get(&Rc::as_ptr(member)).into_iter().flatten() { + let Some(estimate) = relabeled_estimate(reader, &build)? else { + return Ok(None); + }; + replaced.insert(Rc::as_ptr(reader), estimate); + } } } let mut rebuilt: HashMap<*const OperatorNode, Rc> = HashMap::new(); @@ -623,11 +712,64 @@ fn share_summaries( Ok(Some(recompose_guarantees(&out)?)) } +/// `reader`, an estimate or finalization of a per-series build over fewer +/// labels, as the same read of `build` with only `reader`'s columns, by +/// name; `None` for another reader or a missing column. +fn relabeled_estimate( + reader: &OperatorNode, + build: &Rc, +) -> Result>, PerComputationCandidateError> { + use asap_types::ir::ProjectItem; + let operator = match reader.asap() { + Some(ASAPOp::SummaryEstimate { query, .. }) => ASAPOp::SummaryEstimate { + summary_input: Rc::clone(build), + query: query.clone(), + }, + Some(ASAPOp::FinalizeExactAccumulator { .. }) => ASAPOp::FinalizeExactAccumulator { + child: Rc::clone(build), + }, + _ => return Ok(None), + }; + let estimate = Rc::new(OperatorNode::new(Operator::ASAP(operator))?); + let mut cols = Vec::with_capacity(reader.schema.fields.len()); + for field in &reader.schema.fields { + let Some(column) = estimate + .schema + .fields + .iter() + .position(|f| f.name == field.name && f.table == field.table) + else { + return Ok(None); + }; + cols.push(ProjectItem { + alias: Some(field.name.clone()), + expr: ScalarExpr::Column(column), + }); + } + Ok(Some(Rc::new(OperatorNode::new(Operator::NonASAP( + NonASAPOp::Project { + cols, + qualifier: None, + child: estimate, + }, + ))?))) +} + /// One build serving every member: the first member with the members' /// shared size ([`size_for_all_readers`]). fn sized_for_all( members: &[&Rc], ) -> Result>, PerComputationCandidateError> { + // An exact accumulator has one size: the first member serves them all. + if matches!( + members[0].asap(), + Some(ASAPOp::SummaryAgg { + family: FieldDataType::ExactAggregate(..), + .. + }) + ) { + return Ok(Some(Rc::clone(members[0]))); + } let kinds: Vec<&SketchKind> = members .iter() .filter_map(|member| match member.asap() { @@ -1556,6 +1698,7 @@ mod tests { use crate::test_support::lower_promql; use asap_types::ir::properties::AccuracyTarget; use asap_types::ir::schema::schema_support::with_promql_series_identity; + use asap_types::ir::schema::SketchAlgorithm; fn roots(queries: &[&str]) -> Vec { queries @@ -1896,7 +2039,9 @@ mod tests { assert!(own[0].1 < own[1].1); let sharing = capability_sharing_candidates(&roots); assert_eq!(sharing.len(), 1); - assert_eq!(sharing[0].family, SketchAlgorithm::Kll); + assert!( + matches!(&sharing[0].family, FieldDataType::Sketch(kind, _) if kind.algorithm() == &SketchAlgorithm::Kll) + ); let pass2_candidates = summary_capability_pass2_candidates(&roots, &sharing).unwrap(); assert_eq!(pass2_candidates.len(), 1); let [shared_builds] = pass2_candidates[0].shared_builds.as_slice() else { @@ -2013,10 +2158,10 @@ mod tests { SketchAlgorithm::DDSketch, ); assert_eq!(capability_sharing_candidates(&families).len(), 1); - assert_eq!( - capability_sharing_candidates(&families)[0].family, - SketchAlgorithm::DDSketch - ); + assert!(matches!( + &capability_sharing_candidates(&families)[0].family, + FieldDataType::Sketch(kind, _) if kind.algorithm() == &SketchAlgorithm::DDSketch + )); } /// Quantiles over other groupings share no capability build. diff --git a/crates/logical-asap-optimization/src/per_computation_candidates.rs b/crates/logical-asap-optimization/src/per_computation_candidates.rs index da8df64f0..b5653af72 100644 --- a/crates/logical-asap-optimization/src/per_computation_candidates.rs +++ b/crates/logical-asap-optimization/src/per_computation_candidates.rs @@ -1723,7 +1723,8 @@ fn fold_stats(reduction: &Reduction, input: &Rc) -> Option Option { _ => None, } } + // Renaming, reordering or dropping columns keeps every value. Operator::NonASAP(NonASAPOp::Project { cols, child, .. }) - if cols.len() == child.schema.fields.len() - && cols - .iter() - .enumerate() - .all(|(i, item)| item.expr == ScalarExpr::Column(i)) => + if cols + .iter() + .all(|item| matches!(item.expr, ScalarExpr::Column(_))) => { estimates_samples_of(child) } diff --git a/crates/logical-asap-optimization/src/workload_candidates.rs b/crates/logical-asap-optimization/src/workload_candidates.rs index 4f7a09709..8a7777b4f 100644 --- a/crates/logical-asap-optimization/src/workload_candidates.rs +++ b/crates/logical-asap-optimization/src/workload_candidates.rs @@ -256,7 +256,7 @@ fn pass2_candidates( found.identical_expression(&none, roots); } found.window_composition(&none, roots, evaluation_grid_ms); - found.reuse(&none, roots); + found.reuse(&none, roots, evaluation_grid_ms); let candidates = capability_sharing_candidates(roots); let pass2_candidates = match summary_capability_pass2_candidates(roots, &candidates) { Ok(pass2_candidates) => pass2_candidates, @@ -373,11 +373,13 @@ impl Pass2List<'_> { /// The reuse Pass 2 candidates of `roots`: the exact top-k reuse /// rule's, then each rollup Pass 2 candidate followed by its own exact - /// top-k reuse Pass 2 candidates; each followed by its + /// top-k reuse and window-composition Pass 2 candidates (the finer + /// build as tumbling panes, so a deployment can store it and the coarse + /// computation rolls up its merged state); each followed by its /// identical-expression Pass 2 candidates. They are listed eagerly, /// like every Pass 2 rule's: more than [`MAX_REUSE_CANDIDATES`] rollup /// and top-k reuse candidates together list one error. - fn reuse(&mut self, rules: &Rules, roots: &[QueryRoot]) { + fn reuse(&mut self, rules: &Rules, roots: &[QueryRoot], evaluation_grid_ms: &[Option]) { let candidates = rollup_candidates(roots); // Every rollup subset is combined with every top-k subset: cap the // product, not each factor. @@ -412,6 +414,7 @@ impl Pass2List<'_> { self.identical_expression(&rules, &roots); } self.topk_reuse(&rules, &roots); + self.window_composition(&rules, &roots, evaluation_grid_ms); } Err(error) => self.out.push(Pass2::failed(&rules, error)), } @@ -497,6 +500,53 @@ mod tests { .collect() } + /// A rollup Pass 2 candidate is also built over tumbling panes: for a + /// fine and a coarse sum over one window, some Pass 2 candidate answers + /// the coarse sum from the fine one and builds the fine one as panes, so + /// a deployment that keeps no raw samples can store it and roll the + /// coarse sum up from its merged state. + #[test] + fn a_rollup_is_also_built_over_tumbling_panes() { + let roots = [ + "sum by (region, service, job) (sum_over_time(data[100s]))", + "sum by (region) (sum_over_time(data[100s]))", + ] + .iter() + .enumerate() + .map(|(i, q)| { + let root = lower_promql(q, asap_types::ir::properties::AccuracyTarget::exact()); + ( + i, + QueryRoot::Operator(with_promql_series_identity(&root).unwrap()), + ) + }) + .collect(); + let inventory = generate_per_computation_candidates(roots).unwrap(); + let entries: Vec = (0..2) + .map(|_| asap_types::workload::QueryWorkloadEntry { + query: Default::default(), + requirements: Default::default(), + predictability: asap_types::workload::Predictability::Predictable { + known_at: None, + }, + recurrence: asap_types::workload::QueryRecurrence::Repeated( + asap_types::workload::RepeatedDemand::FixedInterval( + asap_types::workload::RepetitionInterval(10_000), + ), + ), + range_evaluation: None, + }) + .collect(); + assert!( + stage1_logical_asap_optimization(&inventory, &entries).any(|candidate| { + candidate.roots.is_ok() + && candidate.pass2.as_ref().is_some_and(|pass2| { + !pass2.rollup_candidates.is_empty() && !pass2.pane_candidates.is_empty() + }) + }) + ); + } + /// Skipping an invalid prefix skips exactly the choices that agree with /// it there, and keeps the others in enumeration order; a prefix of /// length 0 skips every choice. @@ -571,7 +621,7 @@ mod tests { out: Vec::new(), viable: &|_| true, }; - found.reuse(&Rules::default(), &roots); + found.reuse(&Rules::default(), &roots, &[]); let [failed] = found.out.as_slice() else { panic!("one error, {} candidates", found.out.len()) }; diff --git a/crates/planner/tests/runtime_install.rs b/crates/planner/tests/runtime_install.rs index 08f8789a3..adca5d304 100644 --- a/crates/planner/tests/runtime_install.rs +++ b/crates/planner/tests/runtime_install.rs @@ -922,13 +922,15 @@ fn relative_time(from_ms: i64, to_ms: i64) -> Vec { }] } -/// A coarse rollup (`sum by (job)`) built at ingestion from the finer -/// stored sums (`sum by (job, instance)`): both are listed, the rollup -/// after the sums it reads, with the sums' time selection under a -/// definition of its own; built in that order from stored inputs, they -/// answer as the query-time reference. +/// A fine and a coarse sum (`sum by (job, instance)` and `sum by (job)`), +/// both built at ingestion from one stored per-series sum: Stage 1 shares +/// the per-series sums the two queries read (over scans that project other +/// labels). The per-series sum is listed first, then each group sum, which +/// reads only it, with its time selection under a definition of its own; +/// built in that order from stored inputs, they answer as the query-time +/// reference. #[tokio::test] -async fn a_rollup_from_finer_stored_sums_is_listed_and_built() { +async fn group_sums_from_one_stored_per_series_sum_are_listed_and_built() { let workload = exact_workload( &[ "sum by (job) (sum_over_time(m[1m]))", @@ -943,19 +945,29 @@ async fn a_rollup_from_finer_stored_sums_is_listed_and_built() { ) .await; let plan: InstallablePlan = serde_json::from_str(&plan_json).unwrap(); - let [finer, coarse] = plan.stored_outputs.as_slice() else { + let [per_series, sums @ ..] = plan.stored_outputs.as_slice() else { panic!("{:#?}", plan.stored_outputs); }; - assert!(finer.inputs.stored_outputs.is_empty()); - assert_eq!(finer.inputs.raw_sample_windows.len(), 1); - assert_eq!(coarse.inputs.stored_outputs, [finer.node_id]); - assert!(coarse.inputs.raw_sample_windows.is_empty()); - for output in [finer, coarse] { + assert_eq!(sums.len(), 2, "{:#?}", plan.stored_outputs); + assert!(per_series.inputs.stored_outputs.is_empty()); + assert_eq!(per_series.inputs.raw_sample_windows.len(), 1); + for sum in sums { + assert_eq!(sum.inputs.stored_outputs, [per_series.node_id]); + assert!(sum.inputs.raw_sample_windows.is_empty()); + assert_ne!( + sum.coverage.definition_id, + per_series.coverage.definition_id + ); + } + assert_ne!( + sums[0].coverage.definition_id, + sums[1].coverage.definition_id + ); + for output in &plan.stored_outputs { assert_eq!(output.built_at, StoredOutputBuild::IngestionRun); assert_eq!(output.coverage.selection, relative_time(-60_000, 0)); assert_eq!(output.retention_ms, 60_000); } - assert_ne!(coarse.coverage.definition_id, finer.coverage.definition_id); assert!(plan.query_time_raw_windows.is_empty()); let run = run_data_plane( &plan_json, @@ -963,14 +975,15 @@ async fn a_rollup_from_finer_stored_sums_is_listed_and_built() { EVALUATION_MS, &mut SummaryStore::new(), ); - assert_eq!(run.outputs_built, 2); + assert_eq!(run.outputs_built, 3); assert_eq!(run.answers_per_query[0].len(), 2, "one sum per job"); + assert_eq!(run.answers_per_query[1].len(), 6, "one sum per series"); assert_eq!(run.answers_per_query, run.reference_answers_per_query); } /// A maximum across series (`max by (job)`) built at ingestion from the -/// stored per-series maxima: listed after them with their time selection, -/// under a definition of its own. +/// stored per-series maxima, which the per-series query reads too: listed +/// after them with their time selection, under a definition of its own. #[tokio::test] async fn a_summary_across_series_from_per_series_summaries_is_listed_and_built() { let workload = exact_workload( @@ -987,22 +1000,16 @@ async fn a_summary_across_series_from_per_series_summaries_is_listed_and_built() ) .await; let plan: InstallablePlan = serde_json::from_str(&plan_json).unwrap(); - let [per_series, across, other_per_series] = plan.stored_outputs.as_slice() else { + let [per_series, across] = plan.stored_outputs.as_slice() else { panic!("{:#?}", plan.stored_outputs); }; + assert!(per_series.inputs.stored_outputs.is_empty()); assert_eq!(across.inputs.stored_outputs, [per_series.node_id]); assert!(across.inputs.raw_sample_windows.is_empty()); - assert!(other_per_series.inputs.stored_outputs.is_empty()); - for output in [per_series, across, other_per_series] { + for output in [per_series, across] { assert_eq!(output.built_at, StoredOutputBuild::IngestionRun); assert_eq!(output.coverage.selection, relative_time(-60_000, 0)); } - // The second query's per-series maxima read no `job` column, so they - // are another state than the first query's. - assert_ne!( - per_series.coverage.definition_id, - other_per_series.coverage.definition_id - ); assert_ne!( across.coverage.definition_id, per_series.coverage.definition_id @@ -1013,7 +1020,7 @@ async fn a_summary_across_series_from_per_series_summaries_is_listed_and_built() EVALUATION_MS, &mut SummaryStore::new(), ); - assert_eq!(run.outputs_built, 3); + assert_eq!(run.outputs_built, 2); assert_eq!(run.answers_per_query[0].len(), 2, "one maximum per job"); assert_eq!(run.answers_per_query[1].len(), 6, "one maximum per series"); assert_eq!(run.answers_per_query, run.reference_answers_per_query); @@ -1267,6 +1274,11 @@ async fn a_non_negative_gauges_merged_ddsketch_panes_carry_the_relative_bound_th // search builds only the plans that scan nothing at query time. assert!(!selection.selected_candidate_id.is_empty()); assert!(!plan.stored_outputs.is_empty(), "panes built at ingestion"); + assert_eq!( + stored_definitions(&plan, "DDSketch").len(), + 1, + "one DDSketch" + ); assert!( plan.query_time_raw_windows.is_empty(), "no raw samples read" @@ -1492,6 +1504,69 @@ fn grouped_samples() -> Vec { samples } +/// Quantile queries over one metric and window with different `by` labels +/// (ASAPQuery's seed-862 quantile queries) share one per-series DDSketch: +/// planned jointly on a deployment that keeps no raw samples, every stored +/// DDSketch has one definition, and each query answers as its query-time +/// reference. Series of `data` carry `region`, `service` and `job`; values +/// are positive. +#[tokio::test] +async fn quantile_queries_with_different_groupings_share_one_per_series_sketch() { + let plan_json = jointly_planned( + &[ + "avg by (region) (quantile_over_time(0.99, data[100s]))", + "avg by (region, service, job) (quantile_over_time(0.9, data[100s]))", + "avg by (service) (quantile_over_time(0.5, data[100s]))", + ], + r#"{"accuracy": {"accuracy_metric": "relative", "max_error": 0.05, + "failure_probability": 0.01, "accuracy_scope": "per_value"}}"#, + ) + .await; + let plan: InstallablePlan = serde_json::from_str(&plan_json).unwrap(); + let sketches = stored_definitions(&plan, "DDSketch"); + assert_eq!(sketches.len(), 1, "{sketches:?}"); + let run = run_data_plane( + &plan_json, + &grouped_series_samples(), + EVALUATION_MS, + &mut SummaryStore::new(), + ); + assert_eq!(run.answers_per_query[0].len(), 2, "one answer per region"); + assert_eq!(run.answers_per_query[1].len(), 6); + assert_eq!(run.answers_per_query[2].len(), 3, "one answer per service"); + assert_eq!(run.answers_per_query, run.reference_answers_per_query); +} + +/// The six series of `data` the sharing tests read: `region` east/west, +/// `service` api/web/db, `job` batch/online; values positive, one every +/// 10 s over the last two minutes, in time order. +fn grouped_series_samples() -> Vec { + let mut samples = Vec::new(); + for (instance, region, service, job) in [ + ("s1", "east", "api", "online"), + ("s2", "east", "web", "online"), + ("s3", "east", "db", "batch"), + ("s4", "west", "api", "online"), + ("s5", "west", "web", "batch"), + ("s6", "west", "db", "batch"), + ] { + let base: f64 = instance[1..].parse().unwrap(); + for step in (0..12).rev() { + let mut sample = sample( + "data", + instance, + EVALUATION_MS - 5_000 - step * 10_000, + base * 10.0 + ((step * 37) % 11) as f64, + ); + sample.labels.insert("region".into(), region.into()); + sample.labels.insert("service".into(), service.into()); + sample.labels.insert("job".into(), job.into()); + samples.push(sample); + } + } + samples +} + /// The exact answer of `avg by (g) (quantile_over_time(0.99, x[100s]))` at /// `EVALUATION_MS` over [`grouped_samples`]: PromQL's p99 (linear /// interpolation at rank `0.99 × (n − 1)`) of each series' samples in @@ -1970,3 +2045,128 @@ async fn check_compiles_reports_the_executors_error() { let error = plan.check_compiles().unwrap_err(); assert!(error.starts_with("query run: "), "{error}"); } + +/// `queries` every 10 s at `accuracy` (JSON) over six ingesting series of +/// `data`, a non-negative gauge, planned jointly on a deployment that keeps +/// no raw samples: the plan file. +async fn jointly_planned(queries: &[&str], accuracy: &str) -> String { + let entries: Vec = queries + .iter() + .map(|query| { + format!( + r#"{{"query": {query:?}, "demand": {{"fixed_interval": 10000}}, + "requirements": {accuracy}, + "predictability": {{"predictable": {{"known_at": null}}}}}}"# + ) + }) + .collect(); + let workload: PlanningWorkload = serde_json::from_str(&format!( + r#"{{"query_workload": {{"language": "promql", "repeating_queries": [{}]}}, + "data_workload": {{"arrival": {{"ingesting": {{"rows_per_second": 0.6}}}}, + "sources": {{"data": {{"series_count": 6, "sample_interval": 10000, + "metric_type": "gauge", "values_non_negative": true}}}}}}}}"#, + entries.join(",") + )) + .unwrap(); + let capabilities = DeploymentCapabilities { + raw_data_retained: false, + ..DeploymentCapabilities::default() + }; + plan_json_for(workload, &capabilities).await +} + +/// The definition ids of `plan`'s stored outputs whose node's payload +/// mentions `family` (e.g. `"DDSketch"`, `"Sum"`). +fn stored_definitions<'a>(plan: &'a InstallablePlan, family: &str) -> BTreeSet<&'a str> { + plan.stored_outputs + .iter() + .filter(|output| { + let node = plan + .dag + .dag + .nodes + .iter() + .find(|n| n.id == output.node_id) + .unwrap(); + serde_json::to_string(&node.payload) + .unwrap() + .contains(family) + }) + .map(|output| output.coverage.definition_id.as_str()) + .collect() +} + +/// A fine and a coarse sum over one metric and window (ASAPQuery's +/// sum-fine and sum-coarse) store one Sum: both group sums are computed +/// from one stored per-series Sum (Stage 1 shares the per-series sums the +/// two queries' scans read), so every stored Sum has one definition, and +/// both answer as the query-time reference. +#[tokio::test] +async fn fine_and_coarse_sums_store_one_sum() { + let plan_json = jointly_planned( + &[ + "sum by (region, service, job) (sum_over_time(data[100s]))", + "sum by (region) (sum_over_time(data[100s]))", + ], + "{}", + ) + .await; + let plan: InstallablePlan = serde_json::from_str(&plan_json).unwrap(); + let sums = stored_definitions(&plan, "\"Sum\""); + assert_eq!(sums.len(), 1, "{sums:?}"); + let run = run_data_plane( + &plan_json, + &grouped_series_samples(), + EVALUATION_MS, + &mut SummaryStore::new(), + ); + assert_eq!(run.answers_per_query[0].len(), 6); + assert_eq!(run.answers_per_query[1].len(), 2, "one sum per region"); + assert_eq!(run.answers_per_query, run.reference_answers_per_query); +} + +/// Per-series answers planned beside a grouping of the same per-series +/// values (over a scan with other labels) answer as their query-time +/// reference, and the two share one stored per-series state: a top-k of +/// per-series sums keeps each chosen series' labels, and per-series +/// DDSketch quantiles keep every series' labels. +#[tokio::test] +async fn per_series_answers_beside_a_grouping_share_one_state_and_answer_as_the_reference() { + for (queries, family, accuracy, answers) in [ + ( + [ + "topk(2, sum_over_time(data[100s]))", + "sum by (region) (sum_over_time(data[100s]))", + ], + "\"Sum\"", + "{}", + 2, + ), + ( + [ + "quantile_over_time(0.9, data[100s])", + "avg by (region) (quantile_over_time(0.99, data[100s]))", + ], + "DDSketch", + r#"{"accuracy": {"accuracy_metric": "relative", "max_error": 0.05, + "failure_probability": 0.01, "accuracy_scope": "per_value"}}"#, + 6, + ), + ] { + let plan_json = jointly_planned(&queries, accuracy).await; + let plan: InstallablePlan = serde_json::from_str(&plan_json).unwrap(); + assert_eq!(stored_definitions(&plan, family).len(), 1, "{queries:?}"); + let run = run_data_plane( + &plan_json, + &grouped_series_samples(), + EVALUATION_MS, + &mut SummaryStore::new(), + ); + assert_eq!(run.answers_per_query[0].len(), answers, "{queries:?}"); + assert_eq!(run.answers_per_query[1].len(), 2, "{queries:?}: per region"); + assert_eq!( + run.answers_per_query, run.reference_answers_per_query, + "{queries:?}" + ); + } +}