diff --git a/crates/asap-aware-mapping/src/pass/major.rs b/crates/asap-aware-mapping/src/pass/major.rs index bc3de3d1..1206bda8 100644 --- a/crates/asap-aware-mapping/src/pass/major.rs +++ b/crates/asap-aware-mapping/src/pass/major.rs @@ -68,10 +68,11 @@ impl OptimizationPass for MajorPass { // Each reference is planned against the entry's accuracy target, so a // combination of approximate values (e.g. a difference of two // quantiles) keeps its summary candidates. The combined guarantee is - // not composed in this pass (it is effectively `None`): composition is - // designed in #669 and implemented in #575, Stage 3's accuracy model - // checks it, and in the MVP a rank-interval guarantee counts as - // `None`. The same holds for scalar references inside Project/Filter + // not composed (it is `None`, #669), so the vector form's candidate is + // kept but selected automatically only with evidence about its output: + // a root guarantee that meets the target, or the deployment's + // empirical accuracy model (plan selection, #576). In the MVP a rank-interval + // guarantee counts as `None`. The same holds for scalar references inside Project/Filter // expressions, which are planned as interior sites. for ((index, _), refs) in workload.scalar_roots().iter().zip(&scalar_refs) { for node in refs { diff --git a/crates/asap-aware-mapping/src/per_computation_candidates.rs b/crates/asap-aware-mapping/src/per_computation_candidates.rs index 2afe4603..1833e8c4 100644 --- a/crates/asap-aware-mapping/src/per_computation_candidates.rs +++ b/crates/asap-aware-mapping/src/per_computation_candidates.rs @@ -24,17 +24,71 @@ //! The input must be a logical DAG: ASAP operators, execution timing or a //! materialization are rejected. No cost or accuracy model is used; ranking //! and discarding valid candidates belong to Stage 3. +//! +//! # Whole-workload candidates (logical ASAP-aware optimization, end of Pass 1) +//! +//! [`compose_logical_candidate`] is Algorithm 1's +//! `SUBSTITUTE(d, s_1 ↦ c_1, …)`: it picks one per-computation candidate per +//! target (`choice[i]` indexes `targets[i].choices.candidates`) and builds one +//! whole-workload candidate DAG. An aggregate whose chosen candidates are all +//! `Exact` is kept. Otherwise each of its measures is substituted on its own +//! over the aggregate's input: +//! +//! - `ExactAccumulator`: `SummaryAgg` with the exact family → +//! `FinalizeExactAccumulator`. +//! - `Summary`: `SummaryAgg` → `SummaryEstimate`. +//! - `AvgAsSumCount`: the finalized `sum` and `count` accumulators divided: +//! a PromQL `BinaryOp`, or for SQL rows a join on the group keys (NULL keys +//! match) and a SQL division, NULL where the count is 0. +//! - `Exact` (or a measure that is no target): a single-measure `Aggregate`. +//! +//! The substituted output keeps the aggregate's column names. The measures +//! of a multi-measure aggregate are joined back on the group keys (NULL keys +//! match, as grouping treats them). A measure filter becomes +//! `SummaryAgg.filter`, so a group with no matching row stays. A +//! `BelowAggregate` `HAVING` becomes a `Filter` below the state, with its key +//! columns renumbered to the input's, and the consumer `Filter` it came from +//! is dropped; it is not a `SummaryAgg.filter`, which would keep the dropped +//! groups. An `AboveOutput` `HAVING` stays a `Filter` above the output. +//! +//! Summaries are sized here, each for its own measure's accuracy requirement. +//! Sizing belongs after Pass 2, where a shared summary takes the strictest +//! requirement among its consumers; Pass 2 sharing is not implemented yet, so +//! this is that post-Pass-2 sizing step with no sharing. +//! +//! Every node of the candidate gets a theoretical guarantee: exact over +//! exact inputs; a `SummaryEstimate` over exact input the sized family's +//! bound where that bound answers the estimate (not for Top-K, where a heap +//! sketch bounds item frequencies, not membership; not for a Count-Min +//! family over weights not proven non-negative); and `None` otherwise and for +//! anything computed from an approximate value. Composing guarantees through +//! operators is #669; plan selection's accuracy model decides a candidate without +//! one. use std::collections::{HashMap, HashSet}; use std::rc::Rc; -use asap_types::ir::operator_properties::Reduction; -use asap_types::ir::{NonASAPOp, OperatorNode, Predicate, QueryRoot, SchemaDerivationError}; -use asap_types::post_asap::{ExactKind, ExactParams, Materialization, SketchAlgorithm}; -use asap_types::pre_asap::{resolve_column_ref, AggIntent, ColumnRef}; +use asap_types::ir::operator_properties::{BinaryOpKind, JoinKind, Reduction}; +use asap_types::ir::{ + ASAPOp, BinaryOperator, ExprSemantics, NonASAPOp, Operator, OperatorNode, OperatorResultKind, + Predicate, ProjectItem, QueryRoot, ScalarExpr, SchemaDerivationError, +}; +use asap_types::post_asap::guarantee::ResultGuarantee; +use asap_types::post_asap::{ + EntityIdentity, ExactKind, ExactParams, FieldDataType, GroupingStrategy, Materialization, + NonNegativeWeightProof, SketchAlgorithm, SketchKind, SketchStatistic, SummaryInputExpr, + SummaryUpdate, WeightDomain, +}; +use asap_types::pre_asap::expr_ir::ArithmeticOpKind; +use asap_types::pre_asap::{ + resolve_column_ref, AggIntent, ColumnRef, CompareOpKind, DataType, ScalarValue, Schema, +}; use asap_types::types::AccuracyTarget; use thiserror::Error; -use crate::replacement::{accuracy_budget, accuracy_target, summary_candidates}; +use crate::accuracy::{AccuracyModel, DefaultAccuracyModel}; +use crate::replacement::{ + accuracy_budget, accuracy_target, default_size_params, summary_candidates, +}; /// One Pass 1 candidate for a computation (one measure of an aggregate). /// @@ -109,7 +163,7 @@ impl ComputationChoices { pub enum HavingPlacement { /// The predicate reads only group keys and every consumer of the /// aggregate applies it, so it is pushed below the aggregate as a row - /// predicate, like a measure filter. + /// predicate (a `Filter` on its input, which drops the group). BelowAggregate, /// The predicate reads aggregate values, so it stays a `Filter` above the /// finalize/estimate output; approximate candidates of this target carry @@ -170,6 +224,10 @@ pub enum PerComputationCandidateError { root: Option, measure: Option, }, + #[error("choice must name one listed candidate per target")] + InvalidChoice, + #[error("unsupported substitution in logical ASAP-aware optimization: {0}")] + Unsupported(&'static str), } /// Enumerate exact, rewrite and summary choices of one measure in stable catalog @@ -386,6 +444,790 @@ pub fn generate_per_computation_candidates( Ok(PerComputationCandidates { roots, targets }) } +/// Build one whole-workload candidate (end of Pass 1): `choice[i]` indexes +/// `inventory.targets[i].choices.candidates`. An aggregate with a chosen +/// candidate other than `Exact` is replaced measure by measure: an exact +/// accumulator by `SummaryAgg` → `FinalizeExactAccumulator`, a summary by +/// `SummaryAgg` → `SummaryEstimate`, `avg` by its `sum` and `count` +/// accumulators divided, and any other measure by its own single-measure +/// aggregate; several measures are joined back on the group keys. A summary +/// is sized here for its measure's accuracy requirement; Pass 1 leaves it +/// unsized. A measure filter becomes `SummaryAgg.filter`; `HAVING` is +/// applied as its target records. A sub-DAG shared between roots stays shared. +/// Every node carries the guarantee [`composed_guarantee`] gives it. +pub fn compose_logical_candidate( + inventory: &PerComputationCandidates, + choice: &[usize], +) -> Result, PerComputationCandidateError> { + if choice.len() != inventory.targets.len() { + return Err(PerComputationCandidateError::InvalidChoice); + } + let mut chosen: Chosen = HashMap::new(); + for (target, &index) in inventory.targets.iter().zip(choice) { + let candidate = target + .choices + .candidates + .get(index) + .ok_or(PerComputationCandidateError::InvalidChoice)?; + chosen + .entry(Rc::as_ptr(&target.target)) + .or_default() + .push((target, candidate)); + } + // Consumer `Filter`s whose key-only `HAVING` moves below a substituted + // aggregate: each is replaced by its input. + let pushed_below: HashSet<_> = chosen + .values() + .filter(|targets| substituted(targets)) + .flat_map(|targets| &targets[0].0.having) + .filter(|entry| entry.placement == HavingPlacement::BelowAggregate) + .filter_map(|entry| entry.filter.as_ref().map(Rc::as_ptr)) + .collect(); + let (mut memo, mut guaranteed) = (HashMap::new(), HashMap::new()); + let mut rewrite = |node: &Rc| { + let node = rewrite(node, &chosen, &pushed_below, &mut memo)?; + with_guarantees(&node, &mut guaranteed) + }; + inventory + .roots + .iter() + .map(|(id, root)| { + let root = match root { + QueryRoot::Operator(node) => QueryRoot::Operator(rewrite(node)?), + QueryRoot::Scalar(expr) => { + let mut rewritten = HashMap::new(); + for node in expr.operator_refs() { + rewritten.insert(Rc::as_ptr(node), rewrite(node)?); + } + QueryRoot::Scalar( + expr.map_operator_refs(&mut |node| rewritten[&Rc::as_ptr(node)].clone()), + ) + } + }; + // A substitution this module cannot express is an error, never an + // invalid DAG. + root.validate_structure()?; + Ok((id.clone(), root)) + }) + .collect() +} + +/// Each aggregate's targets with their chosen candidates. +type Chosen<'a> = + HashMap<*const OperatorNode, Vec<(&'a ComputationTarget, &'a ComputationCandidate)>>; +type Memo = HashMap<*const OperatorNode, Rc>; + +/// Whether any measure of the aggregate is substituted by something other than +/// its plain exact computation. +fn substituted(targets: &[(&ComputationTarget, &ComputationCandidate)]) -> bool { + targets + .iter() + .any(|(_, candidate)| **candidate != ComputationCandidate::Exact) +} + +fn rewrite( + node: &Rc, + chosen: &Chosen<'_>, + pushed_below: &HashSet<*const OperatorNode>, + memo: &mut Memo, +) -> Result, PerComputationCandidateError> { + if let Some(done) = memo.get(&Rc::as_ptr(node)) { + return Ok(done.clone()); + } + for child in node.children() { + rewrite(child, chosen, pushed_below, memo)?; + } + let changed = node + .children() + .iter() + .any(|child| !Rc::ptr_eq(child, &memo[&Rc::as_ptr(child)])); + let rebuilt = match (chosen.get(&Rc::as_ptr(node)), node.non_asap()) { + (Some(targets), _) if substituted(targets) => substitute(node, targets, memo)?, + (_, Some(NonASAPOp::Filter { child, .. })) if pushed_below.contains(&Rc::as_ptr(node)) => { + memo[&Rc::as_ptr(child)].clone() + } + _ if changed => Rc::new(node.with_new_children(|child| memo[&Rc::as_ptr(child)].clone())?), + _ => node.clone(), + }; + memo.insert(Rc::as_ptr(node), rebuilt.clone()); + Ok(rebuilt) +} + +/// `node` and everything below it, each node with its [`composed_guarantee`]. +fn with_guarantees( + node: &Rc, + memo: &mut Memo, +) -> Result, PerComputationCandidateError> { + if let Some(done) = memo.get(&Rc::as_ptr(node)) { + return Ok(done.clone()); + } + for child in node.children() { + with_guarantees(child, memo)?; + } + let mut rebuilt = node.with_new_children(|child| memo[&Rc::as_ptr(child)].clone())?; + rebuilt.guarantee = composed_guarantee(&rebuilt); + let rebuilt = Rc::new(rebuilt); + memo.insert(Rc::as_ptr(node), rebuilt.clone()); + Ok(rebuilt) +} + +/// The theoretical guarantee of a composed node, given its children's: exact +/// over exact inputs; a summary estimate over exact input has its sized +/// family's bound when [`estimate_has_bound`]. Anything computed from an +/// approximate value (arithmetic, a comparison, a value `HAVING` or any other +/// predicate, at any depth) gets `None`: composition through operators is +/// not defined yet (#669), and plan selection's accuracy model decides. A +/// projection of columns only keeps its child's guarantee. A summary state +/// carries none itself. +fn composed_guarantee(node: &OperatorNode) -> Option { + let exact = |node: &OperatorNode| { + node.guarantee + .as_ref() + .is_some_and(ResultGuarantee::is_exact) + }; + // A summary state's input rows and any subquery in its filter. + let exact_inputs = |state: &OperatorNode| state.children().into_iter().all(|c| exact(c)); + match &node.operator { + Operator::ASAP(ASAPOp::FinalizeExactAccumulator { child }) => match child.asap() { + Some(ASAPOp::SummaryAgg { + family: FieldDataType::ExactAggregate(..), + .. + }) if exact_inputs(child) => { + Some(ResultGuarantee::exact("exact accumulator over exact input")) + } + _ => None, + }, + Operator::ASAP(ASAPOp::SummaryEstimate { + summary_input, + query, + }) => match summary_input.asap() { + Some(ASAPOp::SummaryAgg { + family, + input: update, + .. + }) if exact_inputs(summary_input) && estimate_has_bound(family, update, query) => { + DefaultAccuracyModel.local_guarantee(family, query) + } + _ => None, + }, + Operator::ASAP(_) => None, + // Renaming or reordering columns keeps the child's guarantee. + Operator::NonASAP(NonASAPOp::Project { cols, child, .. }) + if cols + .iter() + .all(|item| matches!(item.expr, ScalarExpr::Column(_))) => + { + child.guarantee.clone() + } + Operator::NonASAP(_) => node + .children() + .into_iter() + .all(|child| exact(child)) + .then(|| ResultGuarantee::exact("exact over exact inputs")), + } +} + +/// Whether the default accuracy model's bound for `family` answers `query` +/// over `update`. A heap sketch's bound is per item frequency, not Top-K +/// membership; a Count-Min bound assumes non-negative weights. +fn estimate_has_bound( + family: &FieldDataType, + update: &SummaryUpdate, + query: &SketchStatistic, +) -> bool { + let count_min = matches!( + family, + FieldDataType::Sketch(kind, _) + if matches!(kind.algorithm(), SketchAlgorithm::Cms | SketchAlgorithm::CmsWithHeap) + ); + !matches!(query, SketchStatistic::TopK { .. }) + && !(count_min && update.weight_domain == WeightDomain::UnknownOrSigned) +} + +/// Replace an aggregate whose measures are `targets`' (or exact, for a +/// measure that is no target) by its substitution, with the same output +/// columns. +fn substitute( + target: &OperatorNode, + targets: &[(&ComputationTarget, &ComputationCandidate)], + memo: &Memo, +) -> Result, PerComputationCandidateError> { + let Some(NonASAPOp::Aggregate { + child, + reduction, + measures, + filters, + .. + }) = target.non_asap() + else { + return Err(PerComputationCandidateError::Unsupported( + "target is not an aggregate", + )); + }; + // A key-only `HAVING` is a row predicate over the input: rows of a + // dropped group never reach the state. + let having = &targets[0].0.having; + // Operator refs in a predicate (a subquery) are this aggregate's + // children, already rewritten. + let rewritten = |pred: &Predicate| { + Predicate( + pred.0 + .map_operator_refs(&mut |node| memo[&Rc::as_ptr(node)].clone()), + ) + }; + let mut input = memo[&Rc::as_ptr(child)].clone(); + for entry in having { + if entry.placement == HavingPlacement::BelowAggregate { + let keys = reduction.group_keys().map_or(&[][..], |by| by.keys()); + let pred = Predicate(entry.predicate.0.map_columns(&mut |column| keys[column])); + input = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Filter { + pred, + child: input, + }))?; + } + } + let parts = measures + .iter() + .enumerate() + .map(|(measure, intent)| { + let candidate = targets + .iter() + .find(|(target, _)| target.measure == measure) + .map_or(&ComputationCandidate::Exact, |(_, candidate)| *candidate); + let filter = filters.get(measure).and_then(Option::as_ref).map(rewritten); + substitute_measure(target, intent, candidate, filter, &input) + }) + .collect::, _>>()?; + let mut output = match <[_; 1]>::try_from(parts) { + Ok([part]) => with_names_of(target, part)?, + Err(parts) => join_measures(target, parts)?, + }; + for entry in having { + if entry.placement == HavingPlacement::AboveOutput && entry.filter.is_none() { + output = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Filter { + pred: rewritten(&entry.predicate), + child: output, + }))?; + } + } + Ok(output) +} + +/// `output` with `target`'s column names: an estimate or accumulator names +/// its value after the operation (`quantile`, `sum`), not the aggregate's +/// output column. +fn with_names_of( + target: &OperatorNode, + output: Rc, +) -> Result, PerComputationCandidateError> { + let names = |node: &OperatorNode| -> Vec { + node.schema + .fields + .iter() + .map(|field| field.name.clone()) + .collect() + }; + if names(&output) == names(target) { + return Ok(output); + } + if output.schema.fields.len() != target.schema.fields.len() { + return Err(PerComputationCandidateError::Unsupported( + "substituted output with other columns than the aggregate", + )); + } + let cols = names(target) + .into_iter() + .enumerate() + .map(|(column, name)| ProjectItem { + alias: Some(name), + expr: ScalarExpr::Column(column), + }) + .collect(); + Ok(OperatorNode::new_shared(Operator::NonASAP( + NonASAPOp::Project { + cols, + qualifier: None, + child: output, + }, + ))?) +} + +/// One measure of `target` over `input`: its group keys, then its value. +fn substitute_measure( + target: &OperatorNode, + intent: &AggIntent, + candidate: &ComputationCandidate, + filter: Option, + input: &Rc, +) -> Result, PerComputationCandidateError> { + let Some(NonASAPOp::Aggregate { reduction, .. }) = target.non_asap() else { + unreachable!("substitute checked the aggregate") + }; + let state = |family: FieldDataType, intent: &AggIntent, read: SummaryInputExpr| { + Ok::<_, PerComputationCandidateError>(OperatorNode::new_shared(Operator::ASAP( + ASAPOp::SummaryAgg { + child: input.clone(), + input: summary_update(intent, &family, reduction, &input.schema, read)?, + family, + reduction: reduction.clone(), + grouping: GroupingStrategy::default(), + filter: filter.clone(), + }, + ))?) + }; + let finalized = |kind: ExactKind, params: ExactParams, intent: &AggIntent, read| { + let state = state(FieldDataType::ExactAggregate(kind, params), intent, read)?; + Ok::<_, PerComputationCandidateError>(OperatorNode::new_shared(Operator::ASAP( + ASAPOp::FinalizeExactAccumulator { child: state }, + ))?) + }; + Ok(match candidate { + ComputationCandidate::Exact => { + OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Aggregate { + child: input.clone(), + reduction: reduction.clone(), + measures: vec![intent.clone()], + output_names: vec![], + filters: vec![filter.clone()], + having: None, + }))? + } + ComputationCandidate::ExactAccumulator { kind, params } => { + let read = measure_input(intent, input, filter.as_ref())?; + finalized(kind.clone(), params.clone(), intent, read)? + } + ComputationCandidate::Summary(algorithm) => { + let family = + FieldDataType::Sketch(sized(algorithm, intent), GroupingStrategy::default()); + OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryEstimate { + summary_input: state( + family, + intent, + measure_input(intent, input, filter.as_ref())?, + )?, + query: statistic(intent)?, + }))? + } + ComputationCandidate::AvgAsSumCount => { + let AggIntent::Avg { col } = intent else { + return Err(PerComputationCandidateError::Unsupported( + "sum/count of a non-avg", + )); + }; + let sum_intent = AggIntent::Sum { col: *col }; + // The count counts the averaged column, as `avg` does: it is + // non-null (a nullable input prunes this candidate), so it counts + // the rows the sum reads. + let read = measure_input(&sum_intent, input, filter.as_ref())?; + let mut sum = finalized(ExactKind::Sum, ExactParams::Sum, &sum_intent, read.clone())?; + let count = finalized( + ExactKind::Count, + ExactParams::Count, + &AggIntent::Count { + accuracy: AccuracyTarget::Exact, + }, + read, + )?; + if input.result_kind == OperatorResultKind::Relation { + return sql_avg(reduction, filter.is_some(), sum, count); + } + // `avg` is Float64 whatever the summed type, as `Avg`'s output is. + if sum.schema.fields.last().map(|field| &field.dtype) + != Some(&FieldDataType::Plain(DataType::Float64)) + { + let Reduction::Reduce(by) = reduction else { + return Err(PerComputationCandidateError::Unsupported( + "non-Float64 per-series avg", + )); + }; + let mut cols = key_items(by.len()); + cols.push(ProjectItem { + alias: None, + expr: ScalarExpr::Cast { + expr: Box::new(ScalarExpr::Column(by.len())), + to: DataType::Float64, + try_cast: false, + }, + }); + sum = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Project { + cols, + qualifier: None, + child: sum, + }))?; + } + OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::BinaryOp { + operator: BinaryOperator { + checked_relative_division: false, + // A per-series window may hold no finite sum to divide. + checked_finite_division: matches!(reduction, Reduction::PerEntity), + kind: BinaryOpKind::Arithmetic(ArithmeticOpKind::Div), + vector_match: None, + }, + return_bool: false, + lhs: sum, + rhs: count, + }))? + } + }) +} + +/// SQL `avg` from its finalized `sum` and `count`: joined on the group keys +/// (NULL keys match), then `sum / count` as Float64. A group with no row +/// passing the measure filter, or an empty global input, counts 0 and its +/// `avg` is NULL. +fn sql_avg( + reduction: &Reduction, + filtered: bool, + sum: Rc, + count: Rc, +) -> Result, PerComputationCandidateError> { + let Reduction::Reduce(by) = reduction else { + return Err(PerComputationCandidateError::Unsupported( + "SQL avg without `by` keys", + )); + }; + if by.is_without() { + return Err(PerComputationCandidateError::Unsupported( + "SQL avg without `by` keys", + )); + } + let keys = by.len(); + let float = |column| { + Box::new(ScalarExpr::Cast { + expr: Box::new(ScalarExpr::Column(column)), + to: DataType::Float64, + try_cast: false, + }) + }; + let count_column = keys + 1 + keys; + let mut avg = ScalarExpr::Arithmetic { + op: ArithmeticOpKind::Div, + left: float(keys), + right: float(count_column), + semantics: ExprSemantics::Sql, + }; + if filtered || keys == 0 { + avg = ScalarExpr::Case { + operand: None, + branches: vec![( + ScalarExpr::Compare { + left: Box::new(ScalarExpr::Column(count_column)), + op: CompareOpKind::Eq, + right: Box::new(ScalarExpr::Literal(ScalarValue::Int64(0))), + semantics: ExprSemantics::Sql, + }, + ScalarExpr::Literal(ScalarValue::Null), + )], + else_expr: Some(Box::new(avg)), + }; + } + let mut cols = key_items(keys); + cols.push(ProjectItem { + alias: None, + expr: avg, + }); + Ok(OperatorNode::new_shared(Operator::NonASAP( + NonASAPOp::Project { + cols, + qualifier: None, + child: join_on_keys(keys, sum, count)?, + }, + ))?) +} + +/// `left` joined with `right` on their leading `keys` columns; NULL keys +/// match each other, as grouping treats them. +fn join_on_keys( + keys: usize, + left: Rc, + right: Rc, +) -> Result, PerComputationCandidateError> { + let width = left.schema.fields.len(); + let column = |id| Box::new(ScalarExpr::Column(id)); + let matches = (0..keys) + .map(|key| { + let equal = ScalarExpr::Compare { + left: column(key), + op: CompareOpKind::Eq, + right: column(width + key), + semantics: ExprSemantics::Sql, + }; + if right.schema.fields[key].nullable { + ScalarExpr::BoolOr(vec![ + equal, + ScalarExpr::BoolAnd(vec![ + ScalarExpr::IsNull(column(key)), + ScalarExpr::IsNull(column(width + key)), + ]), + ]) + } else { + equal + } + }) + .collect::>(); + let (kind, pred) = match matches.len() { + 0 => ( + JoinKind::Cross, + ScalarExpr::Literal(ScalarValue::Boolean(true)), + ), + _ => (JoinKind::Inner, ScalarExpr::BoolAnd(matches)), + }; + Ok(OperatorNode::new_shared(Operator::NonASAP( + NonASAPOp::Join { + kind, + pred: Predicate(pred), + left, + right, + }, + ))?) +} + +/// Join single-measure `parts` (each: group keys, then one value) on their +/// group keys and project `target`'s output columns. Every part has the same +/// groups, so an inner join keeps them all; NULL keys match each other, as +/// grouping treats them. +fn join_measures( + target: &OperatorNode, + parts: Vec>, +) -> Result, PerComputationCandidateError> { + let Some(NonASAPOp::Aggregate { + reduction: Reduction::Reduce(by), + .. + }) = target.non_asap() + else { + return Err(PerComputationCandidateError::Unsupported( + "multi-measure aggregate without `by` keys", + )); + }; + if by.is_without() { + return Err(PerComputationCandidateError::Unsupported( + "multi-measure aggregate without `by` keys", + )); + } + let keys = by.len(); + let width = keys + 1; + let mut parts = parts.into_iter(); + let mut joined = parts.next().expect("an aggregate has measures"); + for part in parts { + joined = join_on_keys(keys, joined, part)?; + } + let mut cols = key_items(keys); + cols.extend( + (0..target.schema.fields.len() - keys).map(|measure| ProjectItem { + alias: None, + expr: ScalarExpr::Column(measure * width + keys), + }), + ); + for (item, field) in cols.iter_mut().zip(&target.schema.fields) { + item.alias = Some(field.name.clone()); + } + Ok(OperatorNode::new_shared(Operator::NonASAP( + NonASAPOp::Project { + cols, + qualifier: None, + child: joined, + }, + ))?) +} + +fn key_items(keys: usize) -> Vec { + (0..keys) + .map(|key| ProjectItem { + alias: None, + expr: ScalarExpr::Column(key), + }) + .collect() +} + +/// `algorithm` sized for `intent`'s accuracy requirement. +fn sized(algorithm: &SketchAlgorithm, intent: &AggIntent) -> SketchKind { + let (epsilon, delta) = + accuracy_budget(accuracy_target(intent).unwrap_or(&AccuracyTarget::Exact)); + SketchKind::new( + algorithm.clone(), + default_size_params(algorithm.clone(), intent, epsilon, delta), + ) +} + +/// What a measure's state reads from each row of `input`, resolvable there. +/// A SQL `count` reads [`counted_column`]; any other measure, a PromQL count +/// included, reads its input column (the sample value when it names none). +fn measure_input( + intent: &AggIntent, + input: &OperatorNode, + filter: Option<&Predicate>, +) -> Result { + let schema = &input.schema; + let read = match intent { + AggIntent::Count { .. } if input.result_kind == OperatorResultKind::Relation => { + SummaryInputExpr::Column(counted_column(schema, filter)?) + } + _ => crate::replacement::summarised_input(intent, schema).map_err(|_| { + PerComputationCandidateError::Unsupported("input column outside child schema") + })?, + }; + if let SummaryInputExpr::Column(column) = &read { + resolve_column_ref(column, schema).map_err(|_| { + PerComputationCandidateError::Unsupported("summary input column does not resolve") + })?; + } + Ok(read) +} + +/// The column a SQL `count` state updates on. The runtime skips NULL updates, +/// so it is a column non-null on every counted row: one the measure filter +/// proves non-null (`count(col)` lowers to the filter `col IS NOT NULL`), else +/// a non-nullable one, Float64 first (the runtime's numeric update). +fn counted_column( + schema: &Schema, + filter: Option<&Predicate>, +) -> Result { + fn not_null(expr: &ScalarExpr) -> Vec { + match expr { + ScalarExpr::IsNotNull(inner) => match **inner { + ScalarExpr::Column(column) => vec![column], + _ => vec![], + }, + ScalarExpr::BoolAnd(conjuncts) => conjuncts.iter().flat_map(not_null).collect(), + _ => vec![], + } + } + let non_nullable = |float: bool| { + schema + .fields + .iter() + .enumerate() + .filter_map(move |(id, field)| { + (!field.nullable + && (!float || field.dtype == FieldDataType::Plain(DataType::Float64))) + .then_some(id) + }) + }; + let reference = |id: usize| schema.fields.get(id).map(crate::replacement::column_ref); + filter + .map_or_else(Vec::new, |filter| not_null(&filter.0)) + .into_iter() + .chain(non_nullable(true)) + .chain(non_nullable(false)) + .find_map(|id| { + reference(id).filter(|column| resolve_column_ref(column, schema).ok() == Some(id)) + }) + .ok_or(PerComputationCandidateError::Unsupported( + "SQL count over rows with no column known to be non-null", + )) +} + +/// What each input row contributes, given what the measure reads +/// ([`measure_input`]), following the legacy realization rules: heap sketches +/// rank series identities, frequency sketches count values, and every other +/// family reads the measure's input. +fn summary_update( + intent: &AggIntent, + family: &FieldDataType, + reduction: &Reduction, + child: &Schema, + weight: SummaryInputExpr, +) -> Result { + let algorithm = match family { + FieldDataType::Sketch(kind, _) => Some(kind.algorithm()), + _ => None, + }; + Ok(match (intent, algorithm) { + (AggIntent::TopK { .. }, Some(_)) => { + if reduction.group_keys().is_some_and(|keys| keys.is_without()) { + return Err(PerComputationCandidateError::Unsupported( + "Top-K partitions given by `without`", + )); + } + let groups: Vec = reduction + .group_keys() + .into_iter() + .flat_map(|keys| keys.iter().copied()) + .collect(); + let item = if child.has_promql_series_identity() { + // Rows that carry the full series identity rank it as a + // column, the item form the runtime builds keyed summaries + // from. + SummaryInputExpr::Column(ColumnRef::Named( + asap_types::pre_asap::schema::PROMQL_SERIES_IDENTITY.into(), + )) + } else if child.closed { + // A closed PromQL child (an aggregate's output, say) lists + // every label: the item is its labels outside the partition, + // as the legacy current-series and counter rules rank them. + // SQL rows carry no sample value to rank by. + let value = resolve_column_ref(&ColumnRef::SampleValue, child).map_err(|_| { + PerComputationCandidateError::Unsupported( + "Top-K item identity for closed schemas without a sample value", + ) + })?; + let items: Vec<_> = child + .fields + .iter() + .enumerate() + .filter(|(index, _)| *index != value && !groups.contains(index)) + .map(|(_, field)| { + SummaryInputExpr::Column(crate::replacement::column_ref(field)) + }) + .collect(); + if items.is_empty() { + return Err(PerComputationCandidateError::Unsupported( + "Top-K item has no label outside its partition", + )); + } + SummaryInputExpr::Tuple(items) + } else { + let excluding = groups + .iter() + .filter_map(|&index| child.fields.get(index)) + .map(crate::replacement::column_ref) + .collect(); + SummaryInputExpr::EntityIdentity(EntityIdentity::PromqlLabelSet { excluding }) + }; + SummaryUpdate { + item: Some(item), + weight: SummaryInputExpr::Column(ColumnRef::SampleValue), + // Not proven non-negative; selection decides whether CMS is admissible. + weight_domain: WeightDomain::UnknownOrSigned, + } + } + (_, Some(SketchAlgorithm::UnivMon)) + | (AggIntent::Count { .. }, Some(SketchAlgorithm::Cms | SketchAlgorithm::CountSketch)) => { + SummaryUpdate { + item: Some(weight), + weight: SummaryInputExpr::Constant(1.0), + weight_domain: WeightDomain::NonNegative { + proof: NonNegativeWeightProof::UnitCount, + }, + } + } + _ => SummaryUpdate { + item: None, + weight, + weight_domain: WeightDomain::UnknownOrSigned, + }, + }) +} + +fn statistic(intent: &AggIntent) -> Result { + Ok(match intent { + AggIntent::Quantile { q, .. } => SketchStatistic::Quantile { q: *q }, + AggIntent::Cardinality { .. } => SketchStatistic::Cardinality, + AggIntent::FrequencyL2 { .. } => SketchStatistic::FrequencyL2, + AggIntent::FrequencyEntropy { .. } => SketchStatistic::FrequencyEntropy, + AggIntent::TopK { k, .. } => SketchStatistic::TopK { k: *k }, + AggIntent::Count { .. } => SketchStatistic::PointCount { + key: ColumnRef::SampleValue, + value: None, + }, + _ => { + return Err(PerComputationCandidateError::Unsupported( + "sketch evaluation", + )) + } + }) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/asap-aware-mapping/src/replacement.rs b/crates/asap-aware-mapping/src/replacement.rs index 394a1d07..9c26d8d9 100644 --- a/crates/asap-aware-mapping/src/replacement.rs +++ b/crates/asap-aware-mapping/src/replacement.rs @@ -532,10 +532,9 @@ pub struct ReplacementSubDAG { } impl ReplacementSubDAG { - /// Whether this summary still needs accuracy/domain evidence before it can - /// be treated as certified. A missing guarantee on any summary candidate - /// (a sub-DAG whose root is an ASAP operator) is unknown; a kept - /// pre-ASAP sub-DAG carries an explicit exact guarantee. + /// Whether this summary still needs accuracy/domain evidence about its + /// output before it can be selected automatically; it stays a candidate. + /// A kept pre-ASAP sub-DAG carries an explicit exact guarantee. pub fn has_missing_accuracy_evidence(&self) -> bool { matches!( &self.replacement, @@ -2089,7 +2088,7 @@ fn realize_binary( return Ok(None); }; let (op, vector_match) = (&operator.kind, &operator.vector_match); - if !matches!(op, BinaryOpKind::Arithmetic(_)) || vector_match.is_some() { + if !is_supported_exact_binary(root) { return Ok(None); } @@ -2379,12 +2378,16 @@ fn finalize_exact_accumulator( )) } +/// Arithmetic and comparisons over summary operands keep their candidates; +/// over approximate operands their combined guarantee is not composed (#669). fn is_supported_exact_binary(root: &OperatorNode) -> bool { matches!( root.non_asap(), Some(NonASAPOp::BinaryOp { operator: BinaryOperator { - kind: BinaryOpKind::Arithmetic(_), + kind: BinaryOpKind::Arithmetic(_) + | BinaryOpKind::Compare(_) + | BinaryOpKind::CompareBool(_), vector_match: None, .. }, @@ -2455,6 +2458,13 @@ fn ddsketch_quantile_alpha(node: &OperatorNode) -> Option { } } +/// Whether nothing shows that the candidate's output meets its target. Only +/// the output's own guarantee (the composed one; realization already checked +/// it against the target) counts here; certified inputs do not, since a +/// combination of them (a difference, a ratio) can be unbounded. Such a +/// candidate is kept, not pruned: composition through operators (#669) or the +/// deployment's empirical output-level accuracy model (plan selection, #576) may +/// supply the evidence later. fn has_missing_accuracy_evidence(node: &OperatorNode) -> bool { node.guarantee .as_ref() @@ -3828,7 +3838,7 @@ fn summarised_column(intent: &AggIntent, child_schema: &Schema) -> ColumnRef { } } -fn column_ref(column: &Field) -> ColumnRef { +pub(crate) fn column_ref(column: &Field) -> ColumnRef { match &column.table { Some(t) => ColumnRef::Qualified { table: t.clone(), @@ -3846,7 +3856,7 @@ fn column_ref(column: &Field) -> ColumnRef { /// A tuple leg outside the child schema is an error rather than /// [`summarised_column`]'s sample-value fallback: a leg has no sample-value /// reading, and silently dropping one would under-count. -fn summarised_input( +pub(crate) fn summarised_input( intent: &AggIntent, child_schema: &Schema, ) -> Result { diff --git a/crates/asap-aware-mapping/tests/per_computation_candidates.rs b/crates/asap-aware-mapping/tests/per_computation_candidates.rs index fe3bde19..2892bc3a 100644 --- a/crates/asap-aware-mapping/tests/per_computation_candidates.rs +++ b/crates/asap-aware-mapping/tests/per_computation_candidates.rs @@ -850,3 +850,291 @@ fn without_and_per_entity_having_stays_above() { ); } } + +/// A workload candidate replaces only chosen targets, derives the summary's +/// coverage from its subtree, and keeps unchosen plans as they are, now +/// with an exact guarantee. +#[test] +fn composed_candidate_replaces_chosen_target_with_summary_evaluation() { + use asap_aware_mapping::per_computation_candidates::compose_logical_candidate; + let producer = aggregate(AggIntent::Cardinality { + cols: vec![0], + accuracy: approximate(), + }); + let inventory = + generate_per_computation_candidates(vec![(0, QueryRoot::Operator(producer.clone()))]) + .unwrap(); + let exact = compose_logical_candidate(&inventory, &[0]).unwrap(); + let QueryRoot::Operator(kept) = &exact[0].1 else { + panic!("operator root expected") + }; + assert_eq!(kept.operator.kind_name(), producer.operator.kind_name()); + assert_eq!(kept.schema, producer.schema); + assert!(kept.guarantee.as_ref().is_some_and(|g| g.is_exact())); + + let hll = compose_logical_candidate(&inventory, &[1]).unwrap(); + let QueryRoot::Operator(estimate) = &hll[0].1 else { + panic!("operator root expected") + }; + let Some(ASAPOp::SummaryEstimate { summary_input, .. }) = estimate.asap() else { + panic!("summary evaluation expected") + }; + // The state takes every row of its source: one unrestricted selection box. + let coverage = summary_input.coverage().unwrap(); + assert!(OperatorNode::reachable(&coverage.definition).iter().any(|node| matches!( + node.non_asap(), + Some(NonASAPOp::Scan { source: Source::Table { table_ref }, .. }) if table_ref == "flows" + ))); + assert_eq!(coverage.selection.len(), 1); + assert!( + coverage.selection[0].columns.is_empty() && coverage.selection[0].relative_time.is_none() + ); + hll[0].1.validate_structure().unwrap(); + assert!(compose_logical_candidate(&inventory, &[99]).is_err()); +} + +/// The one root of the workload candidate composed from `roots` and `choice`. +fn composed(roots: Vec>, choice: &[usize]) -> Vec> { + use asap_aware_mapping::per_computation_candidates::compose_logical_candidate; + let inventory = generate_per_computation_candidates( + roots + .into_iter() + .enumerate() + .map(|(id, node)| (id, QueryRoot::Operator(node))) + .collect(), + ) + .unwrap(); + compose_logical_candidate(&inventory, choice) + .unwrap() + .into_iter() + .map(|(_, root)| { + root.validate_structure().unwrap(); + match root { + QueryRoot::Operator(node) => node, + QueryRoot::Scalar(_) => panic!("operator root expected"), + } + }) + .collect() +} +/// Output columns as (name, type, nullable). +fn columns(node: &OperatorNode) -> Vec<(String, FieldDataType, bool)> { + node.schema + .fields + .iter() + .map(|field| (field.name.clone(), field.dtype.clone(), field.nullable)) + .collect() +} +fn summary_agg(node: &OperatorNode) -> &ASAPOp { + match node.asap() { + Some(ASAPOp::SummaryEstimate { summary_input, .. }) => summary_agg(summary_input), + Some(ASAPOp::FinalizeExactAccumulator { child }) => summary_agg(child), + Some(agg @ ASAPOp::SummaryAgg { .. }) => agg, + _ => panic!("summary evaluation expected, got {node:#?}"), + } +} + +/// A summary is sized when the workload candidate is composed, for its +/// measure's accuracy requirement. +#[test] +fn composed_summary_is_sized_for_its_measure() { + use asap_aware_mapping::replacement::default_size_params; + let intent = AggIntent::Count { + accuracy: approximate(), + }; + let [root] = <[_; 1]>::try_from(composed(vec![aggregate(intent.clone())], &[1])).unwrap(); + let ASAPOp::SummaryAgg { + family: FieldDataType::Sketch(kind, _), + .. + } = summary_agg(&root) + else { + panic!("sketch state expected") + }; + assert_eq!(*kind.algorithm(), SketchAlgorithm::Cms); + assert_eq!( + kind.params(), + &default_size_params(SketchAlgorithm::Cms, &intent, 0.05, 0.01) + ); +} + +/// A measure filter becomes the state's filter; a key-only `HAVING` applied +/// by every consumer moves below the state; a value `HAVING` stays a `Filter` +/// above the estimate. +#[test] +fn composed_filters_and_having() { + let cms = 1; + let filtered = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Aggregate { + child: scan(false), + reduction: Reduction::by(vec![0]), + measures: vec![AggIntent::Count { + accuracy: approximate(), + }], + output_names: vec![], + filters: vec![Some(always())], + having: None, + })) + .unwrap(); + let [root] = <[_; 1]>::try_from(composed(vec![filtered.clone()], &[cms])).unwrap(); + assert!( + matches!(summary_agg(&root), ASAPOp::SummaryAgg { filter: Some(f), child, .. } + if *f == always() && child.non_asap().is_some_and(|op| matches!(op, NonASAPOp::Scan { .. }))) + ); + assert_eq!(columns(&root)[0], columns(&filtered)[0]); + + let key = equals(0, ScalarValue::Utf8("a".into())); + let [root] = <[_; 1]>::try_from(composed( + vec![having_filter(grouped_count(None), key.clone())], + &[cms], + )) + .unwrap(); + let ASAPOp::SummaryAgg { child, filter, .. } = summary_agg(&root) else { + unreachable!() + }; + assert_eq!(*filter, None); + assert!(matches!(child.non_asap(), Some(NonASAPOp::Filter { pred, .. }) if *pred == key)); + + let value = || equals(1, ScalarValue::Int64(10)); + for root in composed( + vec![ + having_filter(grouped_count(None), value()), + grouped_count(Some(value())), + ], + &[cms, cms], + ) { + let Some(NonASAPOp::Filter { pred, child }) = root.non_asap() else { + panic!("value HAVING above the estimate expected, got {root:#?}") + }; + assert_eq!(*pred, value()); + assert!(matches!(child.asap(), Some(ASAPOp::SummaryEstimate { .. }))); + } +} + +/// SQL `avg` joins its finalized `sum` and `count` accumulator states on the +/// group keys and divides them. +#[test] +fn composed_avg_divides_sum_and_count_states() { + let avg = aggregate_over( + scan(false), + vec![AggIntent::Avg { col: Some(1) }], + vec![], + None, + ); + let [root] = <[_; 1]>::try_from(composed(vec![avg.clone()], &[0])).unwrap(); + // The quotient is renamed to the aggregate's column. + assert_eq!(columns(&root), columns(&avg)); + let join = OperatorNode::reachable(&root) + .into_iter() + .find(|node| matches!(node.non_asap(), Some(NonASAPOp::Join { .. }))) + .unwrap_or_else(|| panic!("sum joined with count expected, got {root:#?}")); + let Some(NonASAPOp::Join { left, right, .. }) = join.non_asap() else { + unreachable!() + }; + for (side, kind) in [(left, ExactKind::Sum), (right, ExactKind::Count)] { + assert!(matches!( + summary_agg(side), + ASAPOp::SummaryAgg { family: FieldDataType::ExactAggregate(found, _), .. } if *found == kind + )); + } +} + +/// Each measure of a multi-measure aggregate is substituted on its own and the +/// measures are joined back on the group keys, with the aggregate's columns. +#[test] +fn composed_multi_measure_aggregate_joins_its_measures() { + for reduction in [Reduction::by(vec![0]), Reduction::by(vec![])] { + let node = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Aggregate { + child: scan(false), + reduction, + measures: vec![ + AggIntent::Sum { col: Some(1) }, + AggIntent::Count { + accuracy: approximate(), + }, + AggIntent::StdDev { + col: Some(1), + population: true, + }, + ], + output_names: vec![], + filters: vec![], + having: None, + })) + .unwrap(); + let [root] = <[_; 1]>::try_from(composed(vec![node.clone()], &[0, 1])).unwrap(); + assert!(matches!(root.non_asap(), Some(NonASAPOp::Project { .. }))); + let reached = OperatorNode::reachable(&root); + let joins = reached + .iter() + .filter(|node| matches!(node.non_asap(), Some(NonASAPOp::Join { .. }))) + .count(); + assert_eq!(joins, 2); + assert!(reached + .iter() + .any(|node| matches!(node.asap(), Some(ASAPOp::SummaryEstimate { .. })))); + let names = |node: &OperatorNode| -> Vec<_> { + columns(node) + .into_iter() + .map(|(name, dtype, _)| (name, dtype)) + .collect() + }; + assert_eq!(names(&root), names(&node)); + } +} + +/// A summary estimate carries its sized family's bound; an exact accumulator +/// is exact; anything computed from an approximate value, a value `HAVING` +/// included, has no guarantee. +#[test] +fn composed_guarantees() { + let value = equals(1, ScalarValue::Int64(10)); + let roots = composed( + vec![ + aggregate(AggIntent::Cardinality { + cols: vec![0], + accuracy: approximate(), + }), + aggregate(AggIntent::Sum { col: Some(1) }), + having_filter(grouped_count(None), value), + ], + &[1, 0, 1], + ); + let guarantee = |node: &OperatorNode| node.guarantee.clone(); + let hll = guarantee(&roots[0]).expect("HLL bound"); + assert!(!hll.is_exact()); + assert!(guarantee(&roots[1]).is_some_and(|g| g.is_exact())); + let Some(NonASAPOp::Filter { child, .. }) = roots[2].non_asap() else { + panic!("value HAVING above the estimate expected") + }; + assert!(guarantee(child).is_some()); + assert!(guarantee(&roots[2]).is_none()); +} + +/// The aggregate's own value `having` stays a `Filter` above the output, and +/// a subquery in it reads the subquery's substitution. +#[test] +fn own_having_subquery_is_substituted() { + use asap_aware_mapping::per_computation_candidates::compose_logical_candidate; + let inner = aggregate(AggIntent::Count { + accuracy: approximate(), + }); + let outer = grouped_count(Some(above_scalar(1, inner.clone()))); + let inventory = + generate_per_computation_candidates(vec![(0, QueryRoot::Operator(outer.clone()))]).unwrap(); + assert_eq!(inventory.targets.len(), 2); + let cms = 1; + let composed = compose_logical_candidate(&inventory, &[cms, cms]).unwrap(); + let QueryRoot::Operator(root) = &composed[0].1 else { + panic!("operator root expected") + }; + let Some(NonASAPOp::Filter { pred, child }) = root.non_asap() else { + panic!("own HAVING above the estimate expected, got {root:#?}") + }; + assert!(matches!(child.asap(), Some(ASAPOp::SummaryEstimate { .. }))); + let [subquery] = <[_; 1]>::try_from(pred.0.operator_refs()).unwrap(); + assert!( + matches!(subquery.asap(), Some(ASAPOp::SummaryEstimate { .. })), + "{subquery:#?}" + ); + assert!(OperatorNode::reachable(root) + .iter() + .all(|node| !matches!(node.non_asap(), Some(NonASAPOp::Aggregate { .. })))); +} diff --git a/crates/devtools/src/bin/stage_pipeline.rs b/crates/devtools/src/bin/stage_pipeline.rs new file mode 100644 index 00000000..23fe0d06 --- /dev/null +++ b/crates/devtools/src/bin/stage_pipeline.rs @@ -0,0 +1,340 @@ +// cargo run -p asap-devtools --bin stage_pipeline -- \ +// --example planner-layering-1 --out planner-layering-example1.json +// cargo run -p asap-devtools --bin stage_pipeline -- \ +// --promql "topk by (job) (10, rate(x[1m]))" --epsilon 0.01 --delta 0.001 --out run.json +// +// Writes an `asap-stage-pipeline/v1` document (tools/dag-viewer) with the +// first two planner stages only: +// - language_specific_frontends: the frontends' workload DAG, one root per query; +// - logical_asap_optimization: Pass 1 whole-workload candidates, one per +// choice of a per-computation candidate for every target (the Cartesian +// product), in enumeration order and capped by `--max-candidates` +// (default 64). A candidate that cannot be composed carries `error` +// instead of `dag`. +// Physical ASAP-aware optimization and plan selection, and therefore cost, +// are absent. +// +// `--promql` may repeat. `--epsilon`/`--delta` apply to every `--promql` +// query; without them the queries are exact. `--interval-ms` is the source +// cadence PromQL needs (default 15000). + +use std::rc::Rc; + +use asap_aware_mapping::per_computation_candidates::{ + compose_logical_candidate, generate_per_computation_candidates, ComputationCandidate, + PerComputationCandidates, +}; +use asap_types::ir::flat::{flatten, FlatDag}; +use asap_types::ir::{OperatorNode, QueryRoot}; +use asap_types::post_asap::SketchAlgorithm; +use asap_types::types::AccuracyTarget; +use asap_types::workload::{ + AccuracyRequirement, BatchEntry, DataArrival, DataDistribution, DataWorkload, DurationMs, + Evidence, EvidenceSource, LatencyRequirement, PlanningWorkload, Predictability, Query, + QueryLanguage, QueryRecurrence, QueryRequirements, QueryTimeScope, QueryWorkload, Rate, + RepeatedDemand, RepeatingEntry, RepetitionInterval, TimeSelection, +}; +use serde_json::{json, Value}; + +const USAGE: &str = "usage: stage_pipeline (--example planner-layering-1 | --promql ... \ +[--epsilon --delta ] [--interval-ms ]) [--max-candidates ] --out "; + +fn main() { + if let Err(message) = run(std::env::args().skip(1).collect()) { + eprintln!("stage_pipeline: {message}\n{USAGE}"); + std::process::exit(2); + } +} + +fn run(args: Vec) -> Result<(), String> { + let mut example = None; + let mut queries = Vec::new(); + let (mut epsilon, mut delta, mut interval_ms) = (None, None, 15_000u64); + let mut max_candidates = 64usize; + let mut out = None; + let mut args = args.into_iter(); + while let Some(flag) = args.next() { + let mut value = || args.next().ok_or(format!("{flag} needs a value")); + let number = |v: String| v.parse::().map_err(|e| format!("{v}: {e}")); + match flag.as_str() { + "--example" => example = Some(value()?), + "--promql" => queries.push(value()?), + "--epsilon" => epsilon = Some(number(value()?)?), + "--delta" => delta = Some(number(value()?)?), + "--interval-ms" => interval_ms = value()?.parse().map_err(|e| format!("{e}"))?, + "--max-candidates" => max_candidates = value()?.parse().map_err(|e| format!("{e}"))?, + "--out" => out = Some(value()?), + other => return Err(format!("unknown argument {other}")), + } + } + let workload = match (example.as_deref(), queries.is_empty()) { + (Some("planner-layering-1"), true) => planner_layering_example1(), + (Some(other), true) => return Err(format!("unknown example {other}")), + (None, false) => { + let accuracy = match (epsilon, delta) { + (None, None) => AccuracyTarget::Exact, + (Some(epsilon), None) => AccuracyTarget::Epsilon(epsilon), + (Some(epsilon), Some(delta)) => AccuracyTarget::EpsilonDelta { epsilon, delta }, + (None, Some(_)) => return Err("--delta needs --epsilon".into()), + }; + promql_batch(&queries, accuracy, interval_ms) + } + _ => return Err("give exactly one of --example or --promql".into()), + }; + let out = out.ok_or("--out is required")?; + let document = stage_pipeline(&workload, max_candidates)?; + let text = serde_json::to_string_pretty(&document).map_err(|e| e.to_string())? + "\n"; + std::fs::write(&out, text).map_err(|e| format!("{out}: {e}")) +} + +fn stage_pipeline(workload: &PlanningWorkload, max_candidates: usize) -> Result { + let roots = asap_frontend_promql::lower_promql_query_workload(workload, 0) + .map_err(|e| format!("lowering: {e}"))?; + let frontends_dag = export(&roots)?; + let inventory = generate_per_computation_candidates(roots.into_iter().enumerate().collect()) + .map_err(|e| format!("Pass 1: {e}"))?; + let combinations = inventory + .targets + .iter() + .map(|target| target.choices.candidates.len()) + .product::(); + let owners = target_owners(&inventory); + let mut candidates = Vec::new(); + let mut choice = vec![0; inventory.targets.len()]; + for index in 0..combinations.min(max_candidates) { + let mut candidate = json!({ + "id": format!("L{}", index + 1), + "label": label(&inventory, &owners, &choice), + }); + // A choice that composition cannot substitute yet is listed with its + // error rather than ending the run. + match compose_logical_candidate(&inventory, &choice) { + Ok(roots) => { + let roots: Vec<_> = roots.into_iter().map(|(_, root)| root).collect(); + candidate["dag"] = json!(export(&roots)?); + } + Err(error) => candidate["error"] = json!(error.to_string()), + } + candidates.push(candidate); + // Mixed-radix increment: the last target varies fastest. + for (digit, target) in choice.iter_mut().zip(&inventory.targets).rev() { + *digit += 1; + if *digit < target.choices.candidates.len() { + break; + } + *digit = 0; + } + } + Ok(json!({ + "format": "asap-stage-pipeline/v1", + "workload": { "queries": workload_queries(workload) }, + "language_specific_frontends": { "dag": frontends_dag }, + "logical_asap_optimization": { + "combinations": combinations, + "capped": combinations > max_candidates, + "candidates": candidates, + }, + })) +} + +/// A logical DAG as a flat node list: node `i` is `nodes[i]`, and children are +/// node indices inside each operator. +fn export(roots: &[QueryRoot]) -> Result { + for root in roots { + root.validate_structure() + .map_err(|e| format!("validation: {e}"))?; + } + Ok(flatten(roots).0) +} + +/// The first query whose plan reaches each target (enumeration order). +fn target_owners(inventory: &PerComputationCandidates) -> Vec { + inventory + .targets + .iter() + .map(|target| { + inventory + .roots + .iter() + .position(|(_, root)| { + let operators = match root { + QueryRoot::Operator(node) => vec![node], + QueryRoot::Scalar(expr) => expr.operator_refs(), + }; + operators.into_iter().any(|node| { + OperatorNode::reachable(node) + .iter() + .any(|n| Rc::ptr_eq(n, &target.target)) + }) + }) + .expect("every target is reached from a root") + }) + .collect() +} + +/// E.g. "Q1 exact · Q2 CMS+heap"; exact accumulators are listed in parentheses. +fn label( + inventory: &PerComputationCandidates, + owners: &[usize], + choice: &[usize], +) -> String { + (0..inventory.roots.len()) + .map(|query| { + let mut sketches = Vec::new(); + let mut accumulators = Vec::new(); + for (target, (&owner, &index)) in + inventory.targets.iter().zip(owners.iter().zip(choice)) + { + if owner != query { + continue; + } + match &target.choices.candidates[index] { + ComputationCandidate::Exact => {} + ComputationCandidate::ExactAccumulator { kind, .. } => { + accumulators.push(format!("{kind:?} acc")) + } + ComputationCandidate::AvgAsSumCount => { + accumulators.push("Sum/Count acc".into()) + } + ComputationCandidate::Summary(algorithm) => sketches.push(match algorithm { + SketchAlgorithm::CmsWithHeap => "CMS+heap".to_string(), + SketchAlgorithm::CountSketchWithHeap => "CountSketch+heap".to_string(), + other => format!("{other:?}"), + }), + } + } + let mut text = format!("Q{} ", query + 1); + text += &if sketches.is_empty() { + "exact".to_string() + } else { + sketches.join(" + ") + }; + if !accumulators.is_empty() { + text += &format!(" ({})", accumulators.join(", ")); + } + text + }) + .collect::>() + .join(" · ") +} + +fn workload_queries(workload: &PlanningWorkload) -> Vec { + workload + .query_workload + .entries() + .enumerate() + .map(|(index, entry)| { + let accuracy = match entry.requirements.accuracy.target() { + AccuracyTarget::Exact => json!("exact"), + AccuracyTarget::Epsilon(epsilon) => json!({ "epsilon": epsilon }), + AccuracyTarget::EpsilonDelta { epsilon, delta } => { + json!({ "epsilon": epsilon, "delta": delta }) + } + }; + let mut requirements = json!({ "accuracy": accuracy }); + if let LatencyRequirement::ExplicitMaxMs(ms) = entry.requirements.response_latency { + requirements["latency_ms"] = json!(ms); + } + if let QueryRecurrence::Repeated(RepeatedDemand::FixedInterval(interval)) = + &entry.recurrence + { + requirements["repeat_interval_ms"] = json!(interval.0); + } + json!({ + "id": format!("q{}", index + 1), + "language": "promql", + "text": entry.query.0, + "requirements": requirements, + }) + }) + .collect() +} + +fn declared(value: T) -> Evidence { + Evidence { + value: Some(value), + source: EvidenceSource::Declared, + ..Default::default() + } +} + +fn promql_batch( + queries: &[String], + accuracy: AccuracyTarget, + interval_ms: u64, +) -> PlanningWorkload { + PlanningWorkload { + query_workload: QueryWorkload { + language: QueryLanguage::PromQL, + query_batch: Some( + queries + .iter() + .map(|query| BatchEntry { + query: Query(query.clone()), + requirements: QueryRequirements { + accuracy: AccuracyRequirement::Explicit(accuracy.clone()), + ..Default::default() + }, + predictability: Default::default(), + invocations: 1, + execute_at: None, + time_selection: Default::default(), + }) + .collect(), + ), + repeating_queries: None, + }, + data_workload: Some(DataWorkload { + data_ingestion_interval: declared(DurationMs(interval_ms)), + ..Default::default() + }), + } +} + +/// #509 Example 1 (docs/design_docs/proposals/planner-layering.md) over its +/// shared data workload; matches the acceptance spec's test workload. +fn planner_layering_example1() -> PlanningWorkload { + let panel = |query: &str, accuracy, response_latency| RepeatingEntry { + query: Query(query.into()), + demand: RepeatedDemand::FixedInterval(RepetitionInterval(10_000)), + requirements: QueryRequirements { + accuracy: AccuracyRequirement::Explicit(accuracy), + response_latency, + }, + predictability: Predictability::Predictable { known_at: None }, + time_selection: TimeSelection { + scope: QueryTimeScope::RealTime, + lookback: Some(DurationMs(60_000)), + as_of: None, + }, + }; + PlanningWorkload { + query_workload: QueryWorkload { + language: QueryLanguage::PromQL, + query_batch: None, + repeating_queries: Some(vec![ + panel( + "sum by (job) (rate(http_requests_total[1m]))", + AccuracyTarget::Exact, + LatencyRequirement::Unspecified, + ), + panel( + "topk by (job) (10, sum_over_time(http_requests_total[1m]))", + AccuracyTarget::EpsilonDelta { + epsilon: 0.01, + delta: 0.001, + }, + LatencyRequirement::ExplicitMaxMs(100.0), + ), + ]), + }, + data_workload: Some(DataWorkload { + arrival: DataArrival::ContinuouslyIngesting, + data_ingestion_interval: declared(DurationMs(15_000)), + ingestion_volume: Evidence::default(), + ingestion_rate: declared(Rate(1_000_000.0 / 15.0)), + input_cardinality: declared(1_000_000), + distribution: declared(DataDistribution::Zipf), + }), + } +} diff --git a/crates/devtools/tests/stage_pipeline.rs b/crates/devtools/tests/stage_pipeline.rs new file mode 100644 index 00000000..ef5a688b --- /dev/null +++ b/crates/devtools/tests/stage_pipeline.rs @@ -0,0 +1,127 @@ +//! `stage_pipeline` writes a valid viewer document for #509 Example 1: the +//! language-specific frontends' DAG and the logical ASAP-aware optimization +//! candidates. +use std::collections::HashSet; +use std::path::PathBuf; +use std::process::Command; + +use asap_types::ir::flat::FlatDag; +use asap_types::ir::QueryRoot; +use serde_json::Value; + +const COMMITTED: &str = "../../tools/dag-viewer/examples/planner-layering-example1.json"; + +fn generate(extra: &[&str]) -> Value { + let out = std::env::temp_dir().join(format!( + "stage_pipeline_{}_{}.json", + std::process::id(), + extra.len() + )); + let status = Command::new(env!("CARGO_BIN_EXE_stage_pipeline")) + .args(["--example", "planner-layering-1", "--out"]) + .arg(&out) + .args(extra) + .status() + .unwrap(); + assert!(status.success()); + let document = serde_json::from_str(&std::fs::read_to_string(&out).unwrap()).unwrap(); + std::fs::remove_file(out).unwrap(); + document +} + +fn validated_dag(value: &Value, queries: usize) { + let dag: FlatDag = serde_json::from_value(value.clone()).unwrap(); + assert_eq!(dag.roots.len(), queries); + // Children come before their parents. + for (id, node) in dag.nodes.iter().enumerate() { + assert!(node + .operator + .children() + .into_iter() + .all(|child| *child < id)); + } + for root in &dag.roots { + if let QueryRoot::Operator(id) = root { + assert!(*id < dag.nodes.len()); + } + } +} + +/// Every DAG is a well-formed flat DAG with one root per query; ids are unique; +/// only the frontends and logical ASAP-aware optimization are present; the +/// committed fixture is current. +#[test] +fn example1_document_is_valid_and_committed_fixture_is_current() { + let document = generate(&[]); + let committed: Value = serde_json::from_str( + &std::fs::read_to_string(PathBuf::from(env!("CARGO_MANIFEST_DIR")).join(COMMITTED)) + .unwrap(), + ) + .unwrap(); + assert_eq!(document, committed, "regenerate the committed fixture"); + + assert_eq!(document["format"], "asap-stage-pipeline/v1"); + let queries = document["workload"]["queries"].as_array().unwrap().len(); + assert_eq!(queries, 2); + assert!(document.get("physical_asap_optimization").is_none()); + assert!(document.get("plan_selection").is_none()); + validated_dag(&document["language_specific_frontends"]["dag"], queries); + + let logical = &document["logical_asap_optimization"]; + let candidates = logical["candidates"].as_array().unwrap(); + assert_eq!(logical["capped"], false); + assert_eq!(logical["combinations"], candidates.len()); + let mut ids = HashSet::new(); + for candidate in candidates { + assert!(ids.insert(candidate["id"].as_str().unwrap())); + validated_dag(&candidate["dag"], queries); + } +} + +/// The Cartesian product is cut at `--max-candidates` and says so. +#[test] +fn candidate_cap_is_recorded() { + let document = generate(&["--max-candidates", "2"]); + let logical = &document["logical_asap_optimization"]; + assert_eq!(logical["capped"], true); + assert_eq!(logical["candidates"].as_array().unwrap().len(), 2); + assert!(logical["combinations"].as_u64().unwrap() > 2); +} + +/// A choice composition cannot substitute is listed with its error, and the +/// other candidates are still written. +#[test] +fn uncomposable_candidate_is_recorded_with_its_error() { + let out = + std::env::temp_dir().join(format!("stage_pipeline_{}_error.json", std::process::id())); + let status = Command::new(env!("CARGO_BIN_EXE_stage_pipeline")) + .args([ + "--promql", + // Top-K partitioned by every label of its input has no item. + "topk by (job) (3, sum by (job) (rate(m[5m])))", + "--epsilon", + "0.01", + "--delta", + "0.001", + "--out", + ]) + .arg(&out) + .status() + .unwrap(); + assert!(status.success()); + let document: Value = serde_json::from_str(&std::fs::read_to_string(&out).unwrap()).unwrap(); + std::fs::remove_file(out).unwrap(); + let candidates = document["logical_asap_optimization"]["candidates"] + .as_array() + .unwrap(); + let (composed, failed): (Vec<_>, Vec<_>) = candidates + .iter() + .partition(|candidate| candidate.get("dag").is_some()); + assert!(!composed.is_empty() && !failed.is_empty()); + for candidate in failed { + assert!(candidate["error"].as_str().unwrap().contains("Top-K")); + } + for candidate in composed { + validated_dag(&candidate["dag"], 1); + } +} diff --git a/crates/integration-tests/tests/logical_asap_optimization.rs b/crates/integration-tests/tests/logical_asap_optimization.rs new file mode 100644 index 00000000..cef00403 --- /dev/null +++ b/crates/integration-tests/tests/logical_asap_optimization.rs @@ -0,0 +1,728 @@ +//! Frontends → Pass 1 → composed candidate logical ASAP DAGs: every candidate of +//! a query is a valid DAG with the query's output columns (names, types and +//! nullability). + +use std::collections::BTreeMap; +use std::rc::Rc; +use std::sync::Arc; + +use asap_aware_mapping::per_computation_candidates::{ + compose_logical_candidate, generate_per_computation_candidates, ComputationCandidate, +}; +use asap_frontend_sql::{lower_sql, SqlCatalog}; +use asap_physical_operators::{ + physical_planner::{compile, InputContract, Source}, + runtime::{Limits, RunContext, Scope}, + sources::{DataSources, MemorySource}, + values::{Batch, Value}, +}; +use asap_types::ir::physical_export::PhysicalASAPOperatorPayload; +use asap_types::ir::{ASAPOp, NonASAPOp, OperatorNode, QueryRoot, ScalarExpr}; +use asap_types::post_asap::{FieldDataType, SketchAlgorithm, SummaryInputExpr, SummaryUpdate}; +use asap_types::pre_asap::schema::{DataType, Field, Schema}; +use asap_types::pre_asap::{ColumnRef, CompareOpKind, ScalarValue}; +use asap_types::types::AccuracyTarget; +use asap_types::workload::{ + AccuracyRequirement, BatchEntry, DataWorkload, DurationMs, Evidence, PlanningWorkload, Query, + QueryLanguage, QueryRequirements, QueryWorkload, +}; +use futures::StreamExt; + +mod physical_common; + +fn accuracy() -> AccuracyTarget { + AccuracyTarget::EpsilonDelta { + epsilon: 0.01, + delta: 0.01, + } +} + +async fn sql(query: &str) -> QueryRoot { + let catalog = SqlCatalog::new().with_table( + "flows", + Schema::new(vec![ + Field::plain("src", DataType::Utf8, false), + Field::plain("dst", DataType::Utf8, false), + Field::plain("bytes", DataType::Float64, false), + ]), + ); + let root = lower_sql(query, &catalog, accuracy()) + .await + .unwrap_or_else(|e| panic!("lower failed for {query:?}: {e}")); + QueryRoot::Operator(root) +} + +fn promql(query: &str) -> QueryRoot { + let workload = PlanningWorkload { + query_workload: QueryWorkload { + language: QueryLanguage::PromQL, + query_batch: Some(vec![BatchEntry { + query: Query(query.into()), + requirements: QueryRequirements { + accuracy: AccuracyRequirement::Explicit(accuracy()), + ..Default::default() + }, + predictability: Default::default(), + invocations: 1, + execute_at: None, + time_selection: Default::default(), + }]), + repeating_queries: None, + }, + data_workload: Some(DataWorkload { + data_ingestion_interval: Evidence { + value: Some(DurationMs(1000)), + ..Default::default() + }, + ..Default::default() + }), + }; + asap_frontend_promql::lower_promql_query_workload(&workload, 0) + .unwrap_or_else(|e| panic!("lower failed for {query:?}: {e}")) + .remove(0) +} + +fn output_columns(root: &QueryRoot) -> Vec<(String, String, bool)> { + match root { + QueryRoot::Operator(node) => node + .schema + .fields + .iter() + .map(|field| { + ( + field.name.clone(), + format!("{:?}", field.dtype), + field.nullable, + ) + }) + .collect(), + QueryRoot::Scalar(_) => vec![], + } +} + +/// Compose every candidate of `root`; returns how many there were. +fn compose_all(label: &str, root: QueryRoot) -> usize { + let expected = output_columns(&root); + let inventory = generate_per_computation_candidates(vec![(0, root)]).unwrap(); + let radix: Vec<_> = inventory + .targets + .iter() + .map(|target| target.choices.candidates.len()) + .collect(); + let total: usize = radix.iter().product(); + let mut choice = vec![0; radix.len()]; + for _ in 0..total { + let composed = compose_logical_candidate(&inventory, &choice) + .unwrap_or_else(|e| panic!("{label} {choice:?}: {e}")); + let root = &composed[0].1; + root.validate_structure() + .unwrap_or_else(|e| panic!("{label} {choice:?}: {e}")); + assert_eq!(output_columns(root), expected, "{label} {choice:?}"); + for (digit, radix) in choice.iter_mut().zip(&radix).rev() { + *digit += 1; + if *digit < *radix { + break; + } + *digit = 0; + } + } + total +} + +#[tokio::test] +async fn sql_candidates_compose() { + for query in [ + "SELECT src, count(*) FROM flows GROUP BY src HAVING src = 'a'", + "SELECT src, count(*) FROM flows GROUP BY src HAVING count(*) > 10", + "SELECT src, avg(bytes), count(*), approx_distinct(dst) FROM flows GROUP BY src", + "SELECT count(*) FILTER (WHERE bytes > 10), approx_percentile_cont(bytes, 0.99) FROM flows", + "SELECT avg(bytes) FROM flows", + "SELECT sum(flows.bytes) FROM flows", + "SELECT src, approx_percentile_cont(bytes, 0.99) FILTER (WHERE bytes > 10) FROM flows GROUP BY src", + "SELECT approx_percentile_cont(bytes, 0.99) FILTER (WHERE bytes > 10) FROM flows", + ] { + assert!(compose_all(query, sql(query).await) >= 1, "{query}"); + } +} + +#[test] +fn promql_candidates_compose() { + for query in [ + "avg by (job) (m)", + "avg_over_time(m[5m])", + "quantile(0.99, m) - quantile(0.5, m)", + "scalar(quantile(0.99, m)) - scalar(quantile(0.5, m))", + "topk by (job) (10, sum_over_time(m[1m]))", + "topk(3, sum by (job) (rate(m[5m])))", + "count by (job) (m)", + "count_over_time(m[5m])", + "quantile_over_time(0.99, m[5m])", + ] { + assert!(compose_all(query, promql(query)) >= 1, "{query}"); + } +} + +/// The summary estimates of `root`'s candidates that vary one target's +/// choice, every other target taking its first. +fn composed_estimates(root: QueryRoot) -> Vec> { + let inventory = generate_per_computation_candidates(vec![(0, root)]).unwrap(); + let targets = inventory.targets.len(); + let choices: Vec<_> = inventory + .targets + .iter() + .enumerate() + .flat_map(|(target, computation)| { + (1..computation.choices.candidates.len()).map(move |index| { + let mut choice = vec![0; targets]; + choice[target] = index; + choice + }) + }) + .collect(); + choices + .into_iter() + .flat_map(|choice| { + let composed = compose_logical_candidate(&inventory, &choice).unwrap(); + let QueryRoot::Operator(node) = &composed[0].1 else { + panic!("operator root expected") + }; + OperatorNode::reachable(node) + }) + .filter(|node| matches!(node.asap(), Some(ASAPOp::SummaryEstimate { .. }))) + .collect() +} + +fn update_of(estimate: &OperatorNode) -> (SketchAlgorithm, SummaryUpdate) { + let Some(ASAPOp::SummaryEstimate { summary_input, .. }) = estimate.asap() else { + unreachable!() + }; + let Some(ASAPOp::SummaryAgg { + family: FieldDataType::Sketch(kind, _), + input, + .. + }) = summary_input.asap() + else { + panic!("sketch state expected") + }; + (kind.algorithm().clone(), input.clone()) +} + +/// A heap sketch's bound is per item frequency, not Top-K membership, and +/// its weights are not proven non-negative: a Top-K estimate gets no +/// guarantee. A unit-count CMS estimate keeps its bound. +#[test] +fn topk_estimates_carry_no_frequency_bound() { + let topk = composed_estimates(promql("topk by (job) (10, sum_over_time(m[1m]))")); + let algorithms: Vec<_> = topk.iter().map(|node| update_of(node).0).collect(); + assert_eq!( + algorithms, + [ + SketchAlgorithm::CmsWithHeap, + SketchAlgorithm::CountSketchWithHeap + ] + ); + assert!(topk.iter().all(|node| node.guarantee.is_none())); + + let count = composed_estimates(promql("count by (job) (m)")); + let cms = count + .iter() + .find(|node| update_of(node).0 == SketchAlgorithm::Cms) + .expect("CMS candidate"); + assert!(cms.guarantee.as_ref().is_some_and(|g| !g.is_exact())); +} + +/// Top-K over a closed PromQL child without series identity ranks the +/// child's labels outside the partition. +#[test] +fn topk_over_an_aggregate_ranks_its_labels() { + let estimates = composed_estimates(promql("topk(3, sum by (job) (rate(m[5m])))")); + assert_eq!(estimates.len(), 2); + for estimate in &estimates { + assert_eq!( + update_of(estimate).1.item, + Some(SummaryInputExpr::Tuple(vec![SummaryInputExpr::Column( + ColumnRef::Named("job".into()) + )])) + ); + } +} + +/// A subquery inside a measure filter is substituted too: each choice for +/// the inner count gives its own DAG, and an approximate inner count clears +/// the outer measure's guarantee. +#[tokio::test] +async fn subquery_in_a_measure_filter_is_substituted() { + let query = "SELECT src, count(*) FILTER (WHERE bytes > (SELECT count(*) FROM flows)) \ + FROM flows GROUP BY src"; + let inventory = generate_per_computation_candidates(vec![(0, sql(query).await)]).unwrap(); + assert_eq!(inventory.targets.len(), 2); + let inner = inventory + .targets + .iter() + .position(|target| { + target + .target + .non_asap() + .and_then(|op| match op { + NonASAPOp::Aggregate { reduction, .. } => reduction.group_keys(), + _ => None, + }) + .is_some_and(|by| by.keys().is_empty()) + }) + .expect("inner count"); + let candidates = &inventory.targets[inner].choices.candidates; + assert!(candidates.len() > 1); + let (mut seen, mut approximate_inner) = (Vec::new(), false); + for (index, candidate) in candidates.iter().enumerate() { + let mut choice = vec![0; 2]; + choice[inner] = index; + let composed = compose_logical_candidate(&inventory, &choice).unwrap(); + let QueryRoot::Operator(root) = &composed[0].1 else { + panic!("operator root expected") + }; + let shown = format!("{root:?}"); + assert!(!seen.contains(&shown), "{candidate:?} composes as another"); + seen.push(shown); + let finalize = OperatorNode::reachable(root) + .into_iter() + .find(|node| { + matches!( + node.asap(), + Some(ASAPOp::FinalizeExactAccumulator { child }) + if matches!(child.asap(), Some(ASAPOp::SummaryAgg { filter: Some(_), .. })) + ) + }) + .expect("filtered outer count"); + let Some(ASAPOp::FinalizeExactAccumulator { child }) = finalize.asap() else { + unreachable!() + }; + let exact = |node: &OperatorNode| node.guarantee.as_ref().is_some_and(|g| g.is_exact()); + // The state's children: its input rows, then the subquery. + let inputs_exact = child.children().into_iter().all(|node| exact(node)); + assert_eq!(exact(&finalize), inputs_exact, "{candidate:?}"); + approximate_inner |= !inputs_exact; + } + assert!(approximate_inner); +} + +/// Run the logical DAG `root` over `rows` of its one scanned table, each row +/// given by column name. Rows come back shown and sorted. +fn execute(root: &Rc, rows: &[Vec<(&str, Value)>]) -> Vec> { + let dag = physical_common::compile_physical_asap_dag(root).unwrap(); + let scan = dag + .nodes + .iter() + .find(|node| { + matches!( + &node.payload, + PhysicalASAPOperatorPayload::NonASAP(NonASAPOp::Scan { .. }) + ) + }) + .expect("one scan"); + let PhysicalASAPOperatorPayload::NonASAP(NonASAPOp::Scan { source, .. }) = &scan.payload else { + unreachable!() + }; + let schema = Arc::new(scan.output_schema.clone()); + let plan = compile( + &dag, + BTreeMap::from([(scan.id as u64, InputContract::bounded(schema.clone()))]), + &[dag.roots[0] as u64], + ) + .unwrap(); + let rows = rows + .iter() + .map(|row| { + schema + .fields + .iter() + .map(|field| { + row.iter() + .find(|(name, _)| *name == field.name) + .map(|(_, value)| value.clone()) + .expect("value for every column") + }) + .collect() + }) + .collect(); + let expression = OperatorNode::reachable(root) + .into_iter() + .find(|node| matches!(node.non_asap(), Some(NonASAPOp::Scan { source: s, .. }) if s == source)) + .unwrap(); + let mut sources = DataSources::default(); + sources + .register( + source.clone(), + Arc::new( + MemorySource::new( + schema.clone(), + vec![Batch::try_new(schema.clone(), rows).unwrap()], + ) + .unwrap(), + ), + ) + .unwrap(); + let bound = plan + .instantiate(BTreeMap::from([( + scan.id as u64, + Box::new(sources.bind(&expression).unwrap()) as Source<'_>, + )])) + .unwrap(); + let context = RunContext::new( + Scope::Query { + evaluation_time_ms: 0, + revision: 1, + }, + Limits::default(), + ) + .unwrap(); + let show = |value: &Value| match value { + Value::Null => "NULL".to_string(), + Value::Bool(value) => value.to_string(), + Value::Float64(value) => value.to_string(), + Value::Utf8(value) => format!("{value:?}"), + other => format!("{other:?}"), + }; + let mut out: Vec> = futures::executor::block_on(async { + let mut stream = bound.execute(plan.roots(), context).unwrap().remove(0); + let mut rows = Vec::new(); + while let Some(batch) = stream.next().await { + rows.extend( + batch + .unwrap() + .rows() + .iter() + .map(|row| row.iter().map(show).collect()), + ); + } + rows + }); + out.sort(); + out +} + +/// SQL `avg` from its `sum` and `count` states, composed for the query. +async fn avg_as_sum_and_count( + catalog: &SqlCatalog, + query: &str, +) -> (Rc, Rc) { + let exact = lower_sql(query, catalog, accuracy()).await.unwrap(); + let inventory = + generate_per_computation_candidates(vec![(0, QueryRoot::Operator(exact.clone()))]).unwrap(); + let choice: Vec<_> = inventory + .targets + .iter() + .map(|target| { + target + .choices + .candidates + .iter() + .position(|c| *c == ComputationCandidate::AvgAsSumCount) + .expect("avg as sum and count") + }) + .collect(); + let composed = + compose_logical_candidate(&inventory, &choice).unwrap_or_else(|e| panic!("{query}: {e}")); + let QueryRoot::Operator(root) = &composed[0].1 else { + panic!("operator root expected") + }; + assert!( + OperatorNode::reachable(root) + .iter() + .any(|node| matches!(node.non_asap(), Some(NonASAPOp::Join { .. }))), + "{query}: sum joined with count expected" + ); + assert_eq!( + output_columns(&QueryRoot::Operator(root.clone())), + output_columns(&QueryRoot::Operator(exact.clone())), + "{query}" + ); + (exact, root.clone()) +} + +fn avg_table() -> SqlCatalog { + SqlCatalog::new().with_table( + "t", + Schema::new(vec![ + Field::plain("k", DataType::Utf8, true), + Field::plain("ok", DataType::Bool, false), + Field::plain("value", DataType::Float64, false), + ]), + ) +} + +/// SQL `avg` as `sum` and `count` states executes as the exact aggregate: +/// NULL and '' groups stay apart, a non-Utf8 key works, and an empty global +/// input averages to NULL. +#[tokio::test] +async fn sql_avg_from_sum_and_count_executes_as_avg() { + let catalog = avg_table(); + let row = |k: Option<&str>, ok: bool, value: f64| { + vec![ + ("k", k.map_or(Value::Null, |k| Value::Utf8(k.into()))), + ("ok", Value::Bool(ok)), + ("value", Value::Float64(value)), + ] + }; + let rows = vec![ + row(None, true, 2.), + row(None, true, 4.), + row(Some(""), false, 10.), + row(Some("a"), false, 200.), + row(Some("a"), true, 1.), + ]; + let rows_of = |rows: &[&[&str]]| -> Vec> { + rows.iter() + .map(|row| row.iter().map(|cell| cell.to_string()).collect()) + .collect() + }; + for (query, input, want) in [ + ( + "SELECT k, avg(value) FROM t GROUP BY k", + rows.clone(), + rows_of(&[&["\"\"", "10"], &["\"a\"", "100.5"], &["NULL", "3"]]), + ), + ( + "SELECT ok, avg(value) FROM t GROUP BY ok", + rows.clone(), + rows_of(&[&["false", "105"], &["true", "2.3333333333333335"]]), + ), + ( + "SELECT avg(value) FROM t", + rows.clone(), + rows_of(&[&["43.4"]]), + ), + ("SELECT avg(value) FROM t", vec![], rows_of(&[&["NULL"]])), + ] { + let (exact, root) = avg_as_sum_and_count(&catalog, query).await; + assert_eq!(execute(&exact, &input), want, "{query}"); + assert_eq!(execute(&root, &input), want, "{query}"); + } +} + +/// A filtered SQL `avg` is NULL where no row passes the filter: its quotient +/// is NULL when the count is 0. The runtime executes no filtered state yet, +/// so the shape is checked. +#[tokio::test] +async fn filtered_sql_avg_is_null_without_matching_rows() { + let catalog = avg_table(); + for query in [ + "SELECT k, avg(value) FILTER (WHERE value > 100) FROM t GROUP BY k", + "SELECT avg(value) FILTER (WHERE value > 100) FROM t", + ] { + let (_, root) = avg_as_sum_and_count(&catalog, query).await; + assert!(root.schema.fields.last().unwrap().nullable, "{query}"); + let null_when_no_count = OperatorNode::reachable(&root).iter().any(|node| { + let Some(NonASAPOp::Project { cols, child, .. }) = node.non_asap() else { + return false; + }; + let count = child.schema.fields.len() - 1; + matches!( + &cols.last().unwrap().expr, + ScalarExpr::Case { branches, .. } if matches!( + &branches[..], + [(ScalarExpr::Compare { left, op: CompareOpKind::Eq, right, .. }, + ScalarExpr::Literal(ScalarValue::Null))] + if **left == ScalarExpr::Column(count) + && **right == ScalarExpr::Literal(ScalarValue::Int64(0)) + ) + ) + }); + assert!(null_when_no_count, "{query}: {root:#?}"); + } +} + +/// PromQL `avg` divides its `sum` and `count` vectors with a `BinaryOp`. +#[test] +fn promql_avg_divides_sum_and_count_vectors() { + let inventory = + generate_per_computation_candidates(vec![(0, promql("avg by (job) (m)"))]).unwrap(); + assert_eq!( + inventory.targets[0].choices.candidates[0], + ComputationCandidate::AvgAsSumCount + ); + let composed = compose_logical_candidate(&inventory, &[0]).unwrap(); + let QueryRoot::Operator(root) = &composed[0].1 else { + panic!("operator root expected") + }; + let kinds: Vec<_> = OperatorNode::reachable(root) + .iter() + .map(|node| node.operator.kind_name()) + .collect(); + assert!( + kinds.contains(&"BinaryOp") && !kinds.contains(&"Join"), + "{kinds:?}" + ); +} + +/// `t(k Utf8 NULL, )` as a catalog. +fn table(columns: &[(&str, bool)]) -> SqlCatalog { + let mut fields = vec![Field::plain("k", DataType::Utf8, true)]; + fields.extend( + columns + .iter() + .map(|(name, nullable)| Field::plain(*name, DataType::Float64, *nullable)), + ); + SqlCatalog::new().with_table("t", Schema::new(fields)) +} + +/// Every candidate of `query` over `catalog`, composed, by candidate. +async fn every_candidate( + catalog: &SqlCatalog, + query: &str, +) -> Vec<( + ComputationCandidate, + Result< + Vec<(i32, QueryRoot)>, + asap_aware_mapping::per_computation_candidates::PerComputationCandidateError, + >, +)> { + let root = lower_sql(query, catalog, accuracy()).await.unwrap(); + let inventory = + generate_per_computation_candidates(vec![(0, QueryRoot::Operator(root))]).unwrap(); + assert_eq!(inventory.targets.len(), 1, "{query}"); + inventory.targets[0] + .choices + .candidates + .iter() + .enumerate() + .map(|(index, candidate)| { + ( + candidate.clone(), + compose_logical_candidate(&inventory, &[index]), + ) + }) + .collect() +} + +/// The update of every `count` state (exact or sketch) below `root`. +fn count_updates(root: &QueryRoot) -> Vec { + let QueryRoot::Operator(root) = root else { + panic!("operator root expected") + }; + OperatorNode::reachable(root) + .iter() + .filter_map(|node| match node.asap() { + Some(ASAPOp::SummaryAgg { family, input, .. }) + if !matches!( + family, + FieldDataType::ExactAggregate(asap_types::post_asap::ExactKind::Sum, _) + ) => + { + Some(input.clone()) + } + _ => None, + }) + .collect() +} + +/// Whether `update` reads only the column `name` of `t`. +fn reads_only(update: &SummaryUpdate, name: &str) -> bool { + let column = SummaryInputExpr::Column(ColumnRef::Qualified { + table: "t".into(), + name: name.into(), + }); + match &update.item { + Some(item) => *item == column && update.weight == SummaryInputExpr::Constant(1.0), + None => update.weight == column, + } +} + +/// `avg(c)`'s count counts `c`, not a nullable `value` column: the result is +/// the exact `avg`. +#[tokio::test] +async fn sql_avg_counts_its_own_input() { + let catalog = table(&[("value", true), ("c", false)]); + let query = "SELECT k, avg(c) FROM t GROUP BY k"; + let (exact, root) = avg_as_sum_and_count(&catalog, query).await; + let row = |k: &str, value: Option, c: f64| { + vec![ + ("k", Value::Utf8(k.into())), + ("value", value.map_or(Value::Null, Value::Float64)), + ("c", Value::Float64(c)), + ] + }; + let rows = vec![ + row("a", None, 1.), + row("a", Some(5.), 3.), + row("b", None, 4.), + ]; + let want = vec![ + vec!["\"a\"".to_string(), "2".to_string()], + vec!["\"b\"".to_string(), "4".to_string()], + ]; + assert_eq!(execute(&exact, &rows), want); + assert_eq!(execute(&root, &rows), want); +} + +/// `count(*)` counts a non-null column, never a nullable one; executed, it +/// counts the rows whose nullable column is NULL too. +#[tokio::test] +async fn sql_count_star_counts_a_non_null_column() { + let catalog = table(&[("v", true), ("n", false)]); + let query = "SELECT k, count(*) FROM t GROUP BY k"; + for (candidate, composed) in every_candidate(&catalog, query).await { + let composed = composed.unwrap_or_else(|e| panic!("{candidate:?}: {e}")); + composed[0].1.validate_structure().unwrap(); + let updates = count_updates(&composed[0].1); + assert_eq!(updates.len(), 1, "{candidate:?}"); + assert!(reads_only(&updates[0], "n"), "{candidate:?}: {updates:?}"); + } + let (exact, root) = { + let exact = lower_sql(query, &catalog, accuracy()).await.unwrap(); + let inventory = + generate_per_computation_candidates(vec![(0, QueryRoot::Operator(exact.clone()))]) + .unwrap(); + let QueryRoot::Operator(root) = compose_logical_candidate(&inventory, &[0]) + .unwrap() + .remove(0) + .1 + else { + panic!("operator root expected") + }; + (exact, root) + }; + let row = |k: &str, v: Option| { + vec![ + ("k", Value::Utf8(k.into())), + ("v", v.map_or(Value::Null, Value::Float64)), + ("n", Value::Float64(0.)), + ] + }; + let rows = vec![row("a", None), row("a", Some(1.)), row("b", None)]; + assert_eq!(execute(&root, &rows), execute(&exact, &rows)); +} + +/// `count(*)` with no column known to be non-null (one nullable numeric +/// column, or several and none named `value`) is unsupported for every +/// candidate, not a count of a nullable column or an invalid DAG. +#[tokio::test] +async fn sql_count_star_without_a_non_null_column_is_unsupported() { + for columns in [&[("v", true)][..], &[("a", true), ("b", true)][..]] { + let catalog = table(columns); + for (candidate, composed) in + every_candidate(&catalog, "SELECT k, count(*) FROM t GROUP BY k").await + { + match composed { + Err(asap_aware_mapping::per_computation_candidates::PerComputationCandidateError::Unsupported(_)) => {} + other => panic!("{columns:?} {candidate:?}: {other:?}"), + } + } + } +} + +/// `count(v)` of a nullable `v` updates on `v` under its `v IS NOT NULL` +/// filter: it counts the non-null values. The runtime executes no filtered +/// state yet, so the shape is checked. +#[tokio::test] +async fn sql_count_of_a_nullable_column_counts_its_values() { + let catalog = table(&[("v", true), ("n", false)]); + for (candidate, composed) in + every_candidate(&catalog, "SELECT k, count(v) FROM t GROUP BY k").await + { + let composed = composed.unwrap_or_else(|e| panic!("{candidate:?}: {e}")); + composed[0].1.validate_structure().unwrap(); + let updates = count_updates(&composed[0].1); + assert_eq!(updates.len(), 1, "{candidate:?}"); + assert!(reads_only(&updates[0], "v"), "{candidate:?}: {updates:?}"); + } +} diff --git a/crates/planner/tests/summary_sharing.rs b/crates/planner/tests/summary_sharing.rs index ce9e0c9c..20b46c94 100644 --- a/crates/planner/tests/summary_sharing.rs +++ b/crates/planner/tests/summary_sharing.rs @@ -579,6 +579,52 @@ async fn combined_scalar_references_read_summaries() { } } +/// The vector form of a combination of approximate values keeps its summary +/// candidates (nothing prunes them), but its output has no composed guarantee +/// (#669) and the built-in models have no empirical output-level evidence +/// (plan selection, #576), so it is not selected automatically: the exact plan is. +#[tokio::test] +async fn combined_vector_quantiles_keep_unselected_summary_candidates() { + let sketches = |root: &Rc| { + OperatorNode::reachable(root).into_iter().any(|node| { + matches!( + node.asap(), + Some(ASAPOp::SummaryAgg { + family: FieldDataType::Sketch(..), + .. + }) + ) + }) + }; + for query in [ + "quantile(0.99, rate(x[5m])) - quantile(0.5, rate(x[5m]))", + "quantile(0.99, rate(x[5m])) > quantile(0.5, rate(x[5m]))", + ] { + let root = lower_promql_workload(&promql_workload(&[(query, 0.01)]), NOW_MS) + .unwrap() + .remove(0); + let space = search_workload_with_targets( + vec![("q", root, Some(AccuracyTarget::Epsilon(0.01)))], + &asap_aware_mapping::default_strategies(), + &DefaultAccuracyModel, + ) + .unwrap(); + let group = space.candidates_for_target(&space.roots[0].1).unwrap(); + assert!( + group.candidates.iter().any(|candidate| matches!( + &candidate.replacement, + Replacement::SubDAG(node) if sketches(node) + && node.guarantee.is_none() + && candidate.has_missing_accuracy_evidence() + )), + "{query}: summary candidate expected" + ); + + let output = plan_promql_with(&[(query, 0.01)], &PREFER_LARGE_KLL).await; + assert!(!sketches(&output.plans[0].root), "{query}"); + } +} + /// Front ends build structurally identical references as separate nodes and /// planning merges them: such a scalar plans each distinct reference once and /// satisfies the plan contract. diff --git a/crates/types/src/ir/asap.rs b/crates/types/src/ir/asap.rs index 16945d92..902616ea 100644 --- a/crates/types/src/ir/asap.rs +++ b/crates/types/src/ir/asap.rs @@ -270,10 +270,20 @@ impl ASAPOp { { dtype = DataType::Float64; } + // A filtered quantile is NULL for a group without a matching + // row, as the filtered aggregate it realizes is. + let nullable = matches!(query, SketchStatistic::Quantile { .. }) + && matches!( + summary_input.asap(), + Some(SummaryAgg { + filter: Some(_), + .. + }) + ); let mut schema = input.clone(); for field in &mut schema.fields { if !field.is_plain() { - *field = Field::plain(name, dtype.clone(), false); + *field = Field::plain(name, dtype.clone(), nullable); } } schema @@ -284,6 +294,7 @@ impl ASAPOp { family: FieldDataType::ExactAggregate(kind, _), input, reduction, + filter, .. }) = child.asap() { @@ -320,7 +331,9 @@ impl ASAPOp { reduction: reduction.clone(), measures: vec![measure], output_names: vec![], - filters: vec![], + // A filtered SUM/MIN/MAX is NULL for a + // group without a matching row. + filters: vec![filter.clone()], having: None, } .output_schema() @@ -340,6 +353,9 @@ impl ASAPOp { for f in &mut out.fields { if let FieldDataType::ExactAggregate(kind, _) = &f.dtype { if let Some(result) = &value_result { + // The finalized value replaces the aggregate it + // realizes, so it takes that aggregate's column. + f.name = result.name.clone(); f.dtype = result.dtype.clone(); f.nullable = result.nullable; } else { diff --git a/crates/types/src/ir/non_asap.rs b/crates/types/src/ir/non_asap.rs index dd0c2fd7..38b1606c 100644 --- a/crates/types/src/ir/non_asap.rs +++ b/crates/types/src/ir/non_asap.rs @@ -517,8 +517,19 @@ impl NonASAPOp { } => { let mut output = aggregate_output_schema(&child.schema, reduction, measures, output_names)?; + let offset = reduction.group_keys().map_or(0, |keys| keys.len()); + // A filtered quantile is NULL for a group without a matching + // row, as its summary estimate is. + for (index, measure) in measures.iter().enumerate() { + if matches!(measure, AggIntent::Quantile { .. }) + && filters.get(index).is_some_and(Option::is_some) + { + if let Some(field) = output.fields.get_mut(offset + index) { + field.nullable = true; + } + } + } if child.result_kind == OperatorResultKind::Relation { - let offset = reduction.group_keys().map_or(0, |keys| keys.len()); for (index, measure) in measures.iter().enumerate() { if matches!( measure, diff --git a/crates/types/src/ir/scalar.rs b/crates/types/src/ir/scalar.rs index 3d724b25..2315cde9 100644 --- a/crates/types/src/ir/scalar.rs +++ b/crates/types/src/ir/scalar.rs @@ -224,11 +224,32 @@ impl ScalarExpr { /// reads (recursively through scalar children). `f` may change the /// reference type, e.g. from `Rc` to a node id. pub fn map_operator_refs(&self, f: &mut impl FnMut(&C) -> D) -> ScalarExpr { - fn map_box(e: &ScalarExpr, f: &mut impl FnMut(&C) -> D) -> Box> { - Box::new(e.map_operator_refs(f)) + self.map_parts(&mut |column| column, f) + } + + /// This expression with every column reference renumbered by `column`; + /// operator nodes it reads keep their own column scope. + pub fn map_columns(&self, column: &mut impl FnMut(ColumnId) -> ColumnId) -> ScalarExpr + where + C: Clone, + { + self.map_parts(column, &mut C::clone) + } + + fn map_parts( + &self, + column: &mut impl FnMut(ColumnId) -> ColumnId, + f: &mut impl FnMut(&C) -> D, + ) -> ScalarExpr { + fn map_box( + e: &ScalarExpr, + column: &mut impl FnMut(ColumnId) -> ColumnId, + f: &mut impl FnMut(&C) -> D, + ) -> Box> { + Box::new(e.map_parts(column, f)) } match self { - ScalarExpr::Column(c) => ScalarExpr::Column(*c), + ScalarExpr::Column(c) => ScalarExpr::Column(column(*c)), ScalarExpr::Literal(v) => ScalarExpr::Literal(v.clone()), ScalarExpr::CurrentTimestamp => ScalarExpr::CurrentTimestamp, ScalarExpr::EvalTimestamp => ScalarExpr::EvalTimestamp, @@ -243,12 +264,12 @@ impl ScalarExpr { subquery, negated, } => ScalarExpr::InSubquery { - expr: map_box(expr, f), + expr: map_box(expr, column, f), subquery: f(subquery), negated: *negated, }, ScalarExpr::Negative { expr, semantics } => ScalarExpr::Negative { - expr: map_box(expr, f), + expr: map_box(expr, column, f), semantics: *semantics, }, ScalarExpr::Compare { @@ -257,22 +278,22 @@ impl ScalarExpr { right, semantics, } => ScalarExpr::Compare { - left: map_box(left, f), + left: map_box(left, column, f), op: op.clone(), - right: map_box(right, f), + right: map_box(right, column, f), semantics: *semantics, }, ScalarExpr::BoolAnd(parts) => { - ScalarExpr::BoolAnd(parts.iter().map(|p| p.map_operator_refs(f)).collect()) + ScalarExpr::BoolAnd(parts.iter().map(|p| p.map_parts(column, f)).collect()) } ScalarExpr::BoolOr(parts) => { - ScalarExpr::BoolOr(parts.iter().map(|p| p.map_operator_refs(f)).collect()) + ScalarExpr::BoolOr(parts.iter().map(|p| p.map_parts(column, f)).collect()) } - ScalarExpr::Not(e) => ScalarExpr::Not(map_box(e, f)), - ScalarExpr::IsNull(e) => ScalarExpr::IsNull(map_box(e, f)), - ScalarExpr::IsNotNull(e) => ScalarExpr::IsNotNull(map_box(e, f)), + ScalarExpr::Not(e) => ScalarExpr::Not(map_box(e, column, f)), + ScalarExpr::IsNull(e) => ScalarExpr::IsNull(map_box(e, column, f)), + ScalarExpr::IsNotNull(e) => ScalarExpr::IsNotNull(map_box(e, column, f)), ScalarExpr::Cast { expr, to, try_cast } => ScalarExpr::Cast { - expr: map_box(expr, f), + expr: map_box(expr, column, f), to: to.clone(), try_cast: *try_cast, }, @@ -281,13 +302,13 @@ impl ScalarExpr { list, negated, } => ScalarExpr::InList { - expr: map_box(expr, f), - list: list.iter().map(|p| p.map_operator_refs(f)).collect(), + expr: map_box(expr, column, f), + list: list.iter().map(|p| p.map_parts(column, f)).collect(), negated: *negated, }, ScalarExpr::FunctionCall { name, args } => ScalarExpr::FunctionCall { name: name.clone(), - args: args.iter().map(|p| p.map_operator_refs(f)).collect(), + args: args.iter().map(|p| p.map_parts(column, f)).collect(), }, ScalarExpr::Arithmetic { op, @@ -296,8 +317,8 @@ impl ScalarExpr { semantics, } => ScalarExpr::Arithmetic { op: op.clone(), - left: map_box(left, f), - right: map_box(right, f), + left: map_box(left, column, f), + right: map_box(right, column, f), semantics: *semantics, }, ScalarExpr::Case { @@ -305,12 +326,12 @@ impl ScalarExpr { branches, else_expr, } => ScalarExpr::Case { - operand: operand.as_ref().map(|e| map_box(e, f)), + operand: operand.as_ref().map(|e| map_box(e, column, f)), branches: branches .iter() - .map(|(w, t)| (w.map_operator_refs(f), t.map_operator_refs(f))) + .map(|(w, t)| (w.map_parts(column, f), t.map_parts(column, f))) .collect(), - else_expr: else_expr.as_ref().map(|e| map_box(e, f)), + else_expr: else_expr.as_ref().map(|e| map_box(e, column, f)), }, } } diff --git a/tools/dag-viewer/examples/planner-layering-example1.json b/tools/dag-viewer/examples/planner-layering-example1.json new file mode 100644 index 00000000..34a2b9f1 --- /dev/null +++ b/tools/dag-viewer/examples/planner-layering-example1.json @@ -0,0 +1,3083 @@ +{ + "format": "asap-stage-pipeline/v1", + "language_specific_frontends": { + "dag": { + "nodes": [ + { + "guarantee": null, + "materialization": "NotMaterialized", + "operator": { + "NonASAP": { + "Scan": { + "predicates": [], + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "source": { + "TimeSeries": { + "metric": "http_requests_total" + } + } + } + } + }, + "result_kind": "instant_vector", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": null, + "materialization": "NotMaterialized", + "operator": { + "NonASAP": { + "TimeRange": { + "child": 0, + "kind": "range", + "range": { + "nanos": 0, + "secs": 60 + } + } + } + }, + "result_kind": "range_vector", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": null, + "materialization": "NotMaterialized", + "operator": { + "NonASAP": { + "Aggregate": { + "child": 1, + "filters": [], + "having": null, + "measures": [ + { + "kind": "rate" + } + ], + "output_names": [ + "" + ], + "reduction": "PerEntity" + } + } + }, + "result_kind": "instant_vector", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": null, + "materialization": "NotMaterialized", + "operator": { + "NonASAP": { + "Aggregate": { + "child": 2, + "filters": [], + "having": null, + "measures": [ + { + "col": null, + "kind": "sum" + } + ], + "output_names": [ + "" + ], + "reduction": { + "Reduce": [ + 2 + ] + } + } + } + }, + "result_kind": "instant_vector", + "schema": { + "closed": true, + "fields": [ + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "sum", + "nullable": false, + "table": null + } + ], + "time_index": null, + "unique_keys": [ + [ + 0 + ] + ] + }, + "timing": null + }, + { + "guarantee": null, + "materialization": "NotMaterialized", + "operator": { + "NonASAP": { + "Scan": { + "predicates": [], + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "source": { + "TimeSeries": { + "metric": "http_requests_total" + } + } + } + } + }, + "result_kind": "instant_vector", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": null, + "materialization": "NotMaterialized", + "operator": { + "NonASAP": { + "TimeRange": { + "child": 4, + "kind": "range", + "range": { + "nanos": 0, + "secs": 60 + } + } + } + }, + "result_kind": "range_vector", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": null, + "materialization": "NotMaterialized", + "operator": { + "NonASAP": { + "Aggregate": { + "child": 5, + "filters": [], + "having": null, + "measures": [ + { + "col": null, + "kind": "sum" + } + ], + "output_names": [ + "" + ], + "reduction": "PerEntity" + } + } + }, + "result_kind": "instant_vector", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": null, + "materialization": "NotMaterialized", + "operator": { + "NonASAP": { + "Aggregate": { + "child": 6, + "filters": [], + "having": null, + "measures": [ + { + "accuracy": { + "EpsilonDelta": { + "delta": 0.001, + "epsilon": 0.01 + } + }, + "k": 10, + "kind": "top_k" + } + ], + "output_names": [], + "reduction": { + "Reduce": [ + 2 + ] + } + } + } + }, + "result_kind": "instant_vector", + "schema": { + "closed": true, + "fields": [ + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "topk_10", + "nullable": false, + "table": null + } + ], + "time_index": null, + "unique_keys": [ + [ + 0 + ] + ] + }, + "timing": null + } + ], + "roots": [ + { + "Operator": 3 + }, + { + "Operator": 7 + } + ] + } + }, + "logical_asap_optimization": { + "candidates": [ + { + "dag": { + "nodes": [ + { + "guarantee": { + "bound": { + "op": "zero" + }, + "failure_probability": { + "op": "zero" + }, + "metric": "absolute_value", + "provenance": [ + { + "kind": "exact", + "reason": "exact over exact inputs" + } + ] + }, + "materialization": "NotMaterialized", + "operator": { + "NonASAP": { + "Scan": { + "predicates": [], + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "source": { + "TimeSeries": { + "metric": "http_requests_total" + } + } + } + } + }, + "result_kind": "instant_vector", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": { + "bound": { + "op": "zero" + }, + "failure_probability": { + "op": "zero" + }, + "metric": "absolute_value", + "provenance": [ + { + "kind": "exact", + "reason": "exact over exact inputs" + } + ] + }, + "materialization": "NotMaterialized", + "operator": { + "NonASAP": { + "TimeRange": { + "child": 0, + "kind": "range", + "range": { + "nanos": 0, + "secs": 60 + } + } + } + }, + "result_kind": "range_vector", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": null, + "materialization": "NotMaterialized", + "operator": { + "ASAP": { + "SummaryAgg": { + "child": 1, + "family": { + "ExactAggregate": [ + "Rate", + "Rate" + ] + }, + "filter": null, + "grouping": "PerSubpopulationInstance", + "input": { + "item": null, + "weight": { + "Column": "SampleValue" + }, + "weight_domain": { + "kind": "unknown_or_signed" + } + }, + "reduction": "PerEntity" + } + } + }, + "result_kind": "state", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "ExactAggregate": [ + "Rate", + "Rate" + ] + }, + "name": "state", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": { + "bound": { + "op": "zero" + }, + "failure_probability": { + "op": "zero" + }, + "metric": "absolute_value", + "provenance": [ + { + "kind": "exact", + "reason": "exact accumulator over exact input" + } + ] + }, + "materialization": "NotMaterialized", + "operator": { + "ASAP": { + "FinalizeExactAccumulator": { + "child": 2 + } + } + }, + "result_kind": "instant_vector", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": null, + "materialization": "NotMaterialized", + "operator": { + "ASAP": { + "SummaryAgg": { + "child": 3, + "family": { + "ExactAggregate": [ + "Sum", + "Sum" + ] + }, + "filter": null, + "grouping": "PerSubpopulationInstance", + "input": { + "item": null, + "weight": { + "Column": "SampleValue" + }, + "weight_domain": { + "kind": "unknown_or_signed" + } + }, + "reduction": { + "Reduce": [ + 2 + ] + } + } + } + }, + "result_kind": "state", + "schema": { + "closed": true, + "fields": [ + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + }, + { + "dtype": { + "ExactAggregate": [ + "Sum", + "Sum" + ] + }, + "name": "state", + "nullable": false, + "table": null + } + ], + "time_index": null, + "unique_keys": [ + [ + 0 + ] + ] + }, + "timing": null + }, + { + "guarantee": { + "bound": { + "op": "zero" + }, + "failure_probability": { + "op": "zero" + }, + "metric": "absolute_value", + "provenance": [ + { + "kind": "exact", + "reason": "exact accumulator over exact input" + } + ] + }, + "materialization": "NotMaterialized", + "operator": { + "ASAP": { + "FinalizeExactAccumulator": { + "child": 4 + } + } + }, + "result_kind": "instant_vector", + "schema": { + "closed": true, + "fields": [ + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "sum", + "nullable": false, + "table": null + } + ], + "time_index": null, + "unique_keys": [ + [ + 0 + ] + ] + }, + "timing": null + }, + { + "guarantee": { + "bound": { + "op": "zero" + }, + "failure_probability": { + "op": "zero" + }, + "metric": "absolute_value", + "provenance": [ + { + "kind": "exact", + "reason": "exact over exact inputs" + } + ] + }, + "materialization": "NotMaterialized", + "operator": { + "NonASAP": { + "Scan": { + "predicates": [], + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "source": { + "TimeSeries": { + "metric": "http_requests_total" + } + } + } + } + }, + "result_kind": "instant_vector", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": { + "bound": { + "op": "zero" + }, + "failure_probability": { + "op": "zero" + }, + "metric": "absolute_value", + "provenance": [ + { + "kind": "exact", + "reason": "exact over exact inputs" + } + ] + }, + "materialization": "NotMaterialized", + "operator": { + "NonASAP": { + "TimeRange": { + "child": 6, + "kind": "range", + "range": { + "nanos": 0, + "secs": 60 + } + } + } + }, + "result_kind": "range_vector", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": null, + "materialization": "NotMaterialized", + "operator": { + "ASAP": { + "SummaryAgg": { + "child": 7, + "family": { + "ExactAggregate": [ + "Sum", + "Sum" + ] + }, + "filter": null, + "grouping": "PerSubpopulationInstance", + "input": { + "item": null, + "weight": { + "Column": "SampleValue" + }, + "weight_domain": { + "kind": "unknown_or_signed" + } + }, + "reduction": "PerEntity" + } + } + }, + "result_kind": "state", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "ExactAggregate": [ + "Sum", + "Sum" + ] + }, + "name": "state", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": { + "bound": { + "op": "zero" + }, + "failure_probability": { + "op": "zero" + }, + "metric": "absolute_value", + "provenance": [ + { + "kind": "exact", + "reason": "exact accumulator over exact input" + } + ] + }, + "materialization": "NotMaterialized", + "operator": { + "ASAP": { + "FinalizeExactAccumulator": { + "child": 8 + } + } + }, + "result_kind": "instant_vector", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": { + "bound": { + "op": "zero" + }, + "failure_probability": { + "op": "zero" + }, + "metric": "absolute_value", + "provenance": [ + { + "kind": "exact", + "reason": "exact over exact inputs" + } + ] + }, + "materialization": "NotMaterialized", + "operator": { + "NonASAP": { + "Aggregate": { + "child": 9, + "filters": [], + "having": null, + "measures": [ + { + "accuracy": { + "EpsilonDelta": { + "delta": 0.001, + "epsilon": 0.01 + } + }, + "k": 10, + "kind": "top_k" + } + ], + "output_names": [], + "reduction": { + "Reduce": [ + 2 + ] + } + } + } + }, + "result_kind": "instant_vector", + "schema": { + "closed": true, + "fields": [ + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "topk_10", + "nullable": false, + "table": null + } + ], + "time_index": null, + "unique_keys": [ + [ + 0 + ] + ] + }, + "timing": null + } + ], + "roots": [ + { + "Operator": 5 + }, + { + "Operator": 10 + } + ] + }, + "id": "L1", + "label": "Q1 exact (Sum acc, Rate acc) · Q2 exact (Sum acc)" + }, + { + "dag": { + "nodes": [ + { + "guarantee": { + "bound": { + "op": "zero" + }, + "failure_probability": { + "op": "zero" + }, + "metric": "absolute_value", + "provenance": [ + { + "kind": "exact", + "reason": "exact over exact inputs" + } + ] + }, + "materialization": "NotMaterialized", + "operator": { + "NonASAP": { + "Scan": { + "predicates": [], + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "source": { + "TimeSeries": { + "metric": "http_requests_total" + } + } + } + } + }, + "result_kind": "instant_vector", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": { + "bound": { + "op": "zero" + }, + "failure_probability": { + "op": "zero" + }, + "metric": "absolute_value", + "provenance": [ + { + "kind": "exact", + "reason": "exact over exact inputs" + } + ] + }, + "materialization": "NotMaterialized", + "operator": { + "NonASAP": { + "TimeRange": { + "child": 0, + "kind": "range", + "range": { + "nanos": 0, + "secs": 60 + } + } + } + }, + "result_kind": "range_vector", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": null, + "materialization": "NotMaterialized", + "operator": { + "ASAP": { + "SummaryAgg": { + "child": 1, + "family": { + "ExactAggregate": [ + "Rate", + "Rate" + ] + }, + "filter": null, + "grouping": "PerSubpopulationInstance", + "input": { + "item": null, + "weight": { + "Column": "SampleValue" + }, + "weight_domain": { + "kind": "unknown_or_signed" + } + }, + "reduction": "PerEntity" + } + } + }, + "result_kind": "state", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "ExactAggregate": [ + "Rate", + "Rate" + ] + }, + "name": "state", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": { + "bound": { + "op": "zero" + }, + "failure_probability": { + "op": "zero" + }, + "metric": "absolute_value", + "provenance": [ + { + "kind": "exact", + "reason": "exact accumulator over exact input" + } + ] + }, + "materialization": "NotMaterialized", + "operator": { + "ASAP": { + "FinalizeExactAccumulator": { + "child": 2 + } + } + }, + "result_kind": "instant_vector", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": null, + "materialization": "NotMaterialized", + "operator": { + "ASAP": { + "SummaryAgg": { + "child": 3, + "family": { + "ExactAggregate": [ + "Sum", + "Sum" + ] + }, + "filter": null, + "grouping": "PerSubpopulationInstance", + "input": { + "item": null, + "weight": { + "Column": "SampleValue" + }, + "weight_domain": { + "kind": "unknown_or_signed" + } + }, + "reduction": { + "Reduce": [ + 2 + ] + } + } + } + }, + "result_kind": "state", + "schema": { + "closed": true, + "fields": [ + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + }, + { + "dtype": { + "ExactAggregate": [ + "Sum", + "Sum" + ] + }, + "name": "state", + "nullable": false, + "table": null + } + ], + "time_index": null, + "unique_keys": [ + [ + 0 + ] + ] + }, + "timing": null + }, + { + "guarantee": { + "bound": { + "op": "zero" + }, + "failure_probability": { + "op": "zero" + }, + "metric": "absolute_value", + "provenance": [ + { + "kind": "exact", + "reason": "exact accumulator over exact input" + } + ] + }, + "materialization": "NotMaterialized", + "operator": { + "ASAP": { + "FinalizeExactAccumulator": { + "child": 4 + } + } + }, + "result_kind": "instant_vector", + "schema": { + "closed": true, + "fields": [ + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "sum", + "nullable": false, + "table": null + } + ], + "time_index": null, + "unique_keys": [ + [ + 0 + ] + ] + }, + "timing": null + }, + { + "guarantee": { + "bound": { + "op": "zero" + }, + "failure_probability": { + "op": "zero" + }, + "metric": "absolute_value", + "provenance": [ + { + "kind": "exact", + "reason": "exact over exact inputs" + } + ] + }, + "materialization": "NotMaterialized", + "operator": { + "NonASAP": { + "Scan": { + "predicates": [], + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "source": { + "TimeSeries": { + "metric": "http_requests_total" + } + } + } + } + }, + "result_kind": "instant_vector", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": { + "bound": { + "op": "zero" + }, + "failure_probability": { + "op": "zero" + }, + "metric": "absolute_value", + "provenance": [ + { + "kind": "exact", + "reason": "exact over exact inputs" + } + ] + }, + "materialization": "NotMaterialized", + "operator": { + "NonASAP": { + "TimeRange": { + "child": 6, + "kind": "range", + "range": { + "nanos": 0, + "secs": 60 + } + } + } + }, + "result_kind": "range_vector", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": null, + "materialization": "NotMaterialized", + "operator": { + "ASAP": { + "SummaryAgg": { + "child": 7, + "family": { + "ExactAggregate": [ + "Sum", + "Sum" + ] + }, + "filter": null, + "grouping": "PerSubpopulationInstance", + "input": { + "item": null, + "weight": { + "Column": "SampleValue" + }, + "weight_domain": { + "kind": "unknown_or_signed" + } + }, + "reduction": "PerEntity" + } + } + }, + "result_kind": "state", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "ExactAggregate": [ + "Sum", + "Sum" + ] + }, + "name": "state", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": { + "bound": { + "op": "zero" + }, + "failure_probability": { + "op": "zero" + }, + "metric": "absolute_value", + "provenance": [ + { + "kind": "exact", + "reason": "exact accumulator over exact input" + } + ] + }, + "materialization": "NotMaterialized", + "operator": { + "ASAP": { + "FinalizeExactAccumulator": { + "child": 8 + } + } + }, + "result_kind": "instant_vector", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": null, + "materialization": "NotMaterialized", + "operator": { + "ASAP": { + "SummaryAgg": { + "child": 9, + "family": { + "Sketch": [ + { + "algorithm": "CmsWithHeap", + "category": "TopK", + "params": { + "CmsWithHeap": { + "depth": 7, + "heap_size": 100, + "width": 272 + } + } + }, + "PerSubpopulationInstance" + ] + }, + "filter": null, + "grouping": "PerSubpopulationInstance", + "input": { + "item": { + "EntityIdentity": { + "PromqlLabelSet": { + "excluding": [ + { + "Named": "job" + } + ] + } + } + }, + "weight": { + "Column": "SampleValue" + }, + "weight_domain": { + "kind": "unknown_or_signed" + } + }, + "reduction": { + "Reduce": [ + 2 + ] + } + } + } + }, + "result_kind": "state", + "schema": { + "closed": true, + "fields": [ + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + }, + { + "dtype": { + "Sketch": [ + { + "algorithm": "CmsWithHeap", + "category": "TopK", + "params": { + "CmsWithHeap": { + "depth": 7, + "heap_size": 100, + "width": 272 + } + } + }, + "PerSubpopulationInstance" + ] + }, + "name": "state", + "nullable": false, + "table": null + } + ], + "time_index": null, + "unique_keys": [ + [ + 0 + ] + ] + }, + "timing": null + }, + { + "guarantee": null, + "materialization": "NotMaterialized", + "operator": { + "ASAP": { + "SummaryEstimate": { + "query": { + "TopK": { + "k": 10 + } + }, + "summary_input": 10 + } + } + }, + "result_kind": "instant_vector", + "schema": { + "closed": true, + "fields": [ + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "topk", + "nullable": false, + "table": null + } + ], + "time_index": null, + "unique_keys": [ + [ + 0 + ] + ] + }, + "timing": null + }, + { + "guarantee": null, + "materialization": "NotMaterialized", + "operator": { + "NonASAP": { + "Project": { + "child": 11, + "cols": [ + { + "alias": "job", + "expr": { + "Column": 0 + } + }, + { + "alias": "topk_10", + "expr": { + "Column": 1 + } + } + ], + "qualifier": null + } + } + }, + "result_kind": "instant_vector", + "schema": { + "closed": true, + "fields": [ + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "topk_10", + "nullable": false, + "table": null + } + ], + "time_index": null, + "unique_keys": [ + [ + 0 + ] + ] + }, + "timing": null + } + ], + "roots": [ + { + "Operator": 5 + }, + { + "Operator": 12 + } + ] + }, + "id": "L2", + "label": "Q1 exact (Sum acc, Rate acc) · Q2 CMS+heap (Sum acc)" + }, + { + "dag": { + "nodes": [ + { + "guarantee": { + "bound": { + "op": "zero" + }, + "failure_probability": { + "op": "zero" + }, + "metric": "absolute_value", + "provenance": [ + { + "kind": "exact", + "reason": "exact over exact inputs" + } + ] + }, + "materialization": "NotMaterialized", + "operator": { + "NonASAP": { + "Scan": { + "predicates": [], + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "source": { + "TimeSeries": { + "metric": "http_requests_total" + } + } + } + } + }, + "result_kind": "instant_vector", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": { + "bound": { + "op": "zero" + }, + "failure_probability": { + "op": "zero" + }, + "metric": "absolute_value", + "provenance": [ + { + "kind": "exact", + "reason": "exact over exact inputs" + } + ] + }, + "materialization": "NotMaterialized", + "operator": { + "NonASAP": { + "TimeRange": { + "child": 0, + "kind": "range", + "range": { + "nanos": 0, + "secs": 60 + } + } + } + }, + "result_kind": "range_vector", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": null, + "materialization": "NotMaterialized", + "operator": { + "ASAP": { + "SummaryAgg": { + "child": 1, + "family": { + "ExactAggregate": [ + "Rate", + "Rate" + ] + }, + "filter": null, + "grouping": "PerSubpopulationInstance", + "input": { + "item": null, + "weight": { + "Column": "SampleValue" + }, + "weight_domain": { + "kind": "unknown_or_signed" + } + }, + "reduction": "PerEntity" + } + } + }, + "result_kind": "state", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "ExactAggregate": [ + "Rate", + "Rate" + ] + }, + "name": "state", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": { + "bound": { + "op": "zero" + }, + "failure_probability": { + "op": "zero" + }, + "metric": "absolute_value", + "provenance": [ + { + "kind": "exact", + "reason": "exact accumulator over exact input" + } + ] + }, + "materialization": "NotMaterialized", + "operator": { + "ASAP": { + "FinalizeExactAccumulator": { + "child": 2 + } + } + }, + "result_kind": "instant_vector", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": null, + "materialization": "NotMaterialized", + "operator": { + "ASAP": { + "SummaryAgg": { + "child": 3, + "family": { + "ExactAggregate": [ + "Sum", + "Sum" + ] + }, + "filter": null, + "grouping": "PerSubpopulationInstance", + "input": { + "item": null, + "weight": { + "Column": "SampleValue" + }, + "weight_domain": { + "kind": "unknown_or_signed" + } + }, + "reduction": { + "Reduce": [ + 2 + ] + } + } + } + }, + "result_kind": "state", + "schema": { + "closed": true, + "fields": [ + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + }, + { + "dtype": { + "ExactAggregate": [ + "Sum", + "Sum" + ] + }, + "name": "state", + "nullable": false, + "table": null + } + ], + "time_index": null, + "unique_keys": [ + [ + 0 + ] + ] + }, + "timing": null + }, + { + "guarantee": { + "bound": { + "op": "zero" + }, + "failure_probability": { + "op": "zero" + }, + "metric": "absolute_value", + "provenance": [ + { + "kind": "exact", + "reason": "exact accumulator over exact input" + } + ] + }, + "materialization": "NotMaterialized", + "operator": { + "ASAP": { + "FinalizeExactAccumulator": { + "child": 4 + } + } + }, + "result_kind": "instant_vector", + "schema": { + "closed": true, + "fields": [ + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "sum", + "nullable": false, + "table": null + } + ], + "time_index": null, + "unique_keys": [ + [ + 0 + ] + ] + }, + "timing": null + }, + { + "guarantee": { + "bound": { + "op": "zero" + }, + "failure_probability": { + "op": "zero" + }, + "metric": "absolute_value", + "provenance": [ + { + "kind": "exact", + "reason": "exact over exact inputs" + } + ] + }, + "materialization": "NotMaterialized", + "operator": { + "NonASAP": { + "Scan": { + "predicates": [], + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "source": { + "TimeSeries": { + "metric": "http_requests_total" + } + } + } + } + }, + "result_kind": "instant_vector", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": { + "bound": { + "op": "zero" + }, + "failure_probability": { + "op": "zero" + }, + "metric": "absolute_value", + "provenance": [ + { + "kind": "exact", + "reason": "exact over exact inputs" + } + ] + }, + "materialization": "NotMaterialized", + "operator": { + "NonASAP": { + "TimeRange": { + "child": 6, + "kind": "range", + "range": { + "nanos": 0, + "secs": 60 + } + } + } + }, + "result_kind": "range_vector", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": null, + "materialization": "NotMaterialized", + "operator": { + "ASAP": { + "SummaryAgg": { + "child": 7, + "family": { + "ExactAggregate": [ + "Sum", + "Sum" + ] + }, + "filter": null, + "grouping": "PerSubpopulationInstance", + "input": { + "item": null, + "weight": { + "Column": "SampleValue" + }, + "weight_domain": { + "kind": "unknown_or_signed" + } + }, + "reduction": "PerEntity" + } + } + }, + "result_kind": "state", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "ExactAggregate": [ + "Sum", + "Sum" + ] + }, + "name": "state", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": { + "bound": { + "op": "zero" + }, + "failure_probability": { + "op": "zero" + }, + "metric": "absolute_value", + "provenance": [ + { + "kind": "exact", + "reason": "exact accumulator over exact input" + } + ] + }, + "materialization": "NotMaterialized", + "operator": { + "ASAP": { + "FinalizeExactAccumulator": { + "child": 8 + } + } + }, + "result_kind": "instant_vector", + "schema": { + "closed": false, + "fields": [ + { + "dtype": { + "Plain": "timestamp" + }, + "name": "ts", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "float64" + }, + "name": "value", + "nullable": false, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + } + ], + "time_index": 0, + "unique_keys": [] + }, + "timing": null + }, + { + "guarantee": null, + "materialization": "NotMaterialized", + "operator": { + "ASAP": { + "SummaryAgg": { + "child": 9, + "family": { + "Sketch": [ + { + "algorithm": "CountSketchWithHeap", + "category": "TopK", + "params": { + "CountSketchWithHeap": { + "depth": 125, + "heap_size": 100, + "width": 30000 + } + } + }, + "PerSubpopulationInstance" + ] + }, + "filter": null, + "grouping": "PerSubpopulationInstance", + "input": { + "item": { + "EntityIdentity": { + "PromqlLabelSet": { + "excluding": [ + { + "Named": "job" + } + ] + } + } + }, + "weight": { + "Column": "SampleValue" + }, + "weight_domain": { + "kind": "unknown_or_signed" + } + }, + "reduction": { + "Reduce": [ + 2 + ] + } + } + } + }, + "result_kind": "state", + "schema": { + "closed": true, + "fields": [ + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + }, + { + "dtype": { + "Sketch": [ + { + "algorithm": "CountSketchWithHeap", + "category": "TopK", + "params": { + "CountSketchWithHeap": { + "depth": 125, + "heap_size": 100, + "width": 30000 + } + } + }, + "PerSubpopulationInstance" + ] + }, + "name": "state", + "nullable": false, + "table": null + } + ], + "time_index": null, + "unique_keys": [ + [ + 0 + ] + ] + }, + "timing": null + }, + { + "guarantee": null, + "materialization": "NotMaterialized", + "operator": { + "ASAP": { + "SummaryEstimate": { + "query": { + "TopK": { + "k": 10 + } + }, + "summary_input": 10 + } + } + }, + "result_kind": "instant_vector", + "schema": { + "closed": true, + "fields": [ + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "topk", + "nullable": false, + "table": null + } + ], + "time_index": null, + "unique_keys": [ + [ + 0 + ] + ] + }, + "timing": null + }, + { + "guarantee": null, + "materialization": "NotMaterialized", + "operator": { + "NonASAP": { + "Project": { + "child": 11, + "cols": [ + { + "alias": "job", + "expr": { + "Column": 0 + } + }, + { + "alias": "topk_10", + "expr": { + "Column": 1 + } + } + ], + "qualifier": null + } + } + }, + "result_kind": "instant_vector", + "schema": { + "closed": true, + "fields": [ + { + "dtype": { + "Plain": "utf8" + }, + "name": "job", + "nullable": true, + "table": null + }, + { + "dtype": { + "Plain": "utf8" + }, + "name": "topk_10", + "nullable": false, + "table": null + } + ], + "time_index": null, + "unique_keys": [ + [ + 0 + ] + ] + }, + "timing": null + } + ], + "roots": [ + { + "Operator": 5 + }, + { + "Operator": 12 + } + ] + }, + "id": "L3", + "label": "Q1 exact (Sum acc, Rate acc) · Q2 CountSketch+heap (Sum acc)" + } + ], + "capped": false, + "combinations": 3 + }, + "workload": { + "queries": [ + { + "id": "q1", + "language": "promql", + "requirements": { + "accuracy": "exact", + "repeat_interval_ms": 10000 + }, + "text": "sum by (job) (rate(http_requests_total[1m]))" + }, + { + "id": "q2", + "language": "promql", + "requirements": { + "accuracy": { + "delta": 0.001, + "epsilon": 0.01 + }, + "latency_ms": 100.0, + "repeat_interval_ms": 10000 + }, + "text": "topk by (job) (10, sum_over_time(http_requests_total[1m]))" + } + ] + } +}