diff --git a/crates/asap-aware-mapping/src/lib.rs b/crates/asap-aware-mapping/src/lib.rs index b2a1df3da..41b585513 100644 --- a/crates/asap-aware-mapping/src/lib.rs +++ b/crates/asap-aware-mapping/src/lib.rs @@ -211,3 +211,7 @@ pub use rewrite::{AvgToSumOverCountStrategy, SemanticEquivalentRewriteStrategy}; pub use topk_reuse::TopKLimitReuseStrategy; pub mod maintained_population; + +/// Stage 1, Pass 1: per-computation candidate generation. No execution timing is +/// assigned: that is a Stage 2 materialization decision. +pub mod per_computation_candidates; diff --git a/crates/asap-aware-mapping/src/per_computation_candidates.rs b/crates/asap-aware-mapping/src/per_computation_candidates.rs new file mode 100644 index 000000000..2afe46036 --- /dev/null +++ b/crates/asap-aware-mapping/src/per_computation_candidates.rs @@ -0,0 +1,414 @@ +//! Stage 1 (logical ASAP-aware optimization), Pass 1: per-computation +//! candidate generation, as defined in +//! `docs/design_docs/proposals/planner-layering.md`. +//! +//! [`generate_per_computation_candidates`] takes the frontends' logical DAGs +//! (`QueryRoot`s) and returns, for every eligible computation, its unranked +//! candidates. A computation ([`ComputationTarget`]) is one measure of an +//! `Aggregate` whose intent has a realization besides plain exact execution; +//! nested aggregates and operators read by scalar roots are found too. +//! Pass 1 does not consider sharing across computations; that is Pass 2. +//! +//! - Each target has exactly one exact option: the exact accumulator +//! (`ExactAccumulator`, or `AvgAsSumCount` for `avg`) when it can be built, +//! otherwise plain `Exact`. Whether it is materialized is decided in Stage 2. +//! - Summary candidates are unsized; they are sized after Pass 2 sharing. +//! - Candidates are pruned only when provably unable to meet the accuracy +//! requirement, and each pruned candidate keeps its [`PruneReason`]. +//! - A measure filter or `HAVING` prunes nothing. The target records the +//! measure filter (applied as `SummaryAgg.filter`) and one [`HavingEntry`] +//! per consumer `Filter` directly above the aggregate: a key-only predicate +//! applied by every consumer may go below the aggregate, anything else stays +//! above the output (approximate outputs then get guarantee `None`). +//! +//! 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. +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::types::AccuracyTarget; +use thiserror::Error; + +use crate::replacement::{accuracy_budget, accuracy_target, summary_candidates}; + +/// One Pass 1 candidate for a computation (one measure of an aggregate). +/// +/// Each target has exactly one exact option: `ExactAccumulator` or +/// `AvgAsSumCount` when the intent has one and it can be built, otherwise +/// `Exact`. The plain form would duplicate "accumulator, not materialized"; +/// Stage 2/3 decide materialization and sharing. +#[derive(Debug, Clone, PartialEq)] +pub enum ComputationCandidate { + /// Plain exact execution of the original sub-DAG; the exact option only + /// when the intent has no exact accumulator form or that form is pruned. + Exact, + /// An exact mergeable accumulator state. + ExactAccumulator { + kind: ExactKind, + params: ExactParams, + }, + /// An unsized summary family, sized later for the measure's accuracy. + Summary(SketchAlgorithm), + /// `avg` rewritten as exact `sum` and `count` accumulator states. + AvgAsSumCount, +} + +/// Why Pass 1 proved a candidate unable to realize its target. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum PruneReason { + /// The measure requires an exact answer and the summary is approximate. + RequiresExact, + /// `count` counts rows, so a possibly-null `avg` input changes the result. + NullableAvgInput, +} + +/// Unranked candidates of one measure and the candidates pruned for it. +#[derive(Debug, Clone, PartialEq)] +pub struct ComputationChoices { + pub candidates: Vec, + pub pruned: Vec<(ComputationCandidate, PruneReason)>, +} + +impl ComputationChoices { + /// Prune matching candidates; a pruned exact accumulator form is replaced by + /// plain `Exact`, so the target keeps one exact option. + fn prune(&mut self, reason: PruneReason, applies: impl Fn(&ComputationCandidate) -> bool) { + let (pruned, kept): (Vec<_>, _) = std::mem::take(&mut self.candidates) + .into_iter() + .partition(|choice| *choice != ComputationCandidate::Exact && applies(choice)); + self.candidates = kept; + if pruned.iter().any(|choice| { + matches!( + choice, + ComputationCandidate::ExactAccumulator { .. } | ComputationCandidate::AvgAsSumCount + ) + }) { + self.candidates.insert(0, ComputationCandidate::Exact); + } + self.pruned.extend( + pruned + .into_iter() + .map(|choice: ComputationCandidate| (choice, reason)), + ); + } +} + +/// Where a target's `HAVING` is applied. Composition (#575) applies it. +/// +/// Only the adjacent form is recorded for push-down: the aggregate's own +/// `having`, or a `Filter` directly above the aggregate (the form the SQL +/// frontend lowers `HAVING` to). A value predicate anywhere above an +/// approximate output gets `guarantee = None` through guarantee propagation +/// (#575); multi-level push-down is tracked in #664. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +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. + BelowAggregate, + /// The predicate reads aggregate values, so it stays a `Filter` above the + /// finalize/estimate output; approximate candidates of this target carry + /// `guarantee = None`, since membership near the threshold may change. + AboveOutput, +} + +/// One `HAVING` over a target's aggregate output. +#[derive(Debug, Clone, PartialEq)] +pub struct HavingEntry { + /// The consumer `Filter` that applies it; `None` for the aggregate's own + /// `having`, which every consumer reads. + pub filter: Option>, + pub predicate: Predicate, + pub placement: HavingPlacement, +} + +/// One measure of an aggregate with at least one realization besides exact +/// execution. The target retains source, grouping, filters, input expressions +/// and evaluation context. +#[derive(Debug, Clone)] +pub struct ComputationTarget { + pub target: Rc, + /// Index into the aggregate's measures. + pub measure: usize, + pub choices: ComputationChoices, + /// The measure's own filter; only rows that pass it update the state + /// (`SummaryAgg.filter`). + pub filter: Option, + /// The aggregate's own `having`, then one entry per consumer `Filter` + /// directly above the aggregate. + pub having: Vec, +} + +/// Compact Pass 1 inventory; roots and nested producer dependencies are retained. +#[derive(Debug, Clone)] +pub struct PerComputationCandidates { + pub roots: Vec<(Id, QueryRoot)>, + pub targets: Vec, +} + +#[derive(Debug, Error)] +pub enum PerComputationCandidateError { + #[error(transparent)] + Structure(#[from] SchemaDerivationError), + #[error("logical candidate input already has assigned execution timing or materialization")] + AssignedPhysicalAnnotation, + #[error("logical candidate input already contains ASAP operators")] + ContainsAsap, + /// `root` (the root id's `Debug` form) and `measure` locate the target; + /// [`computation_candidates_for_intent`] alone leaves them `None`. + #[error( + "approximate accuracy of root {}, measure {} requires epsilon and delta in (0, 1)", + root.as_deref().unwrap_or("?"), + measure.map_or_else(|| "?".to_string(), |measure| measure.to_string()) + )] + InvalidAccuracy { + root: Option, + measure: Option, + }, +} + +/// Enumerate exact, rewrite and summary choices of one measure in stable catalog +/// order, without ranking or empirical assessment. A summary is not a claim that +/// a deployment meets the measure's accuracy requirement. The one exact option +/// comes first: the exact accumulator form if the intent has one, else `Exact`. +pub fn computation_candidates_for_intent( + intent: &AggIntent, +) -> Result { + let exact = match intent { + AggIntent::Count { .. } => Some((ExactKind::Count, ExactParams::Count)), + AggIntent::Sum { .. } => Some((ExactKind::Sum, ExactParams::Sum)), + AggIntent::Min { .. } => Some((ExactKind::Min, ExactParams::Min)), + AggIntent::Max { .. } => Some((ExactKind::Max, ExactParams::Max)), + AggIntent::Rate => Some((ExactKind::Rate, ExactParams::Rate)), + AggIntent::IRate => Some((ExactKind::IRate, ExactParams::IRate)), + AggIntent::Increase => Some((ExactKind::Increase, ExactParams::Increase)), + _ => None, + }; + let exact = match exact { + Some((kind, params)) => ComputationCandidate::ExactAccumulator { kind, params }, + None if matches!(intent, AggIntent::Avg { .. }) => ComputationCandidate::AvgAsSumCount, + None => ComputationCandidate::Exact, + }; + let mut choices = ComputationChoices { + candidates: vec![exact], + pruned: vec![], + }; + if let Some(target) = accuracy_target(intent) { + if *target != AccuracyTarget::Exact { + let (epsilon, delta) = accuracy_budget(target); + let unit = |value: f64| value > 0.0 && value < 1.0; + if !unit(epsilon) || !unit(delta) { + return Err(PerComputationCandidateError::InvalidAccuracy { + root: None, + measure: None, + }); + } + } + let summaries = summary_candidates(intent).iter().cloned(); + if *target == AccuracyTarget::Exact { + choices.pruned.extend(summaries.map(|algorithm| { + ( + ComputationCandidate::Summary(algorithm), + PruneReason::RequiresExact, + ) + })); + } else { + choices + .candidates + .extend(summaries.map(ComputationCandidate::Summary)); + } + } + Ok(choices) +} + +/// Whether an `avg` measure's input may be null, so `count` rows differ from +/// counted values. An unresolvable input is treated as possibly null. +fn nullable_avg_input(node: &OperatorNode, intent: &AggIntent) -> bool { + let Some(NonASAPOp::Aggregate { child, .. }) = node.non_asap() else { + return true; + }; + let column = match intent.input_cols().first() { + Some(column) => Some(*column), + None => resolve_column_ref(&ColumnRef::SampleValue, &child.schema).ok(), + }; + column + .and_then(|column| child.schema.fields.get(column)) + .is_none_or(|field| field.nullable) +} + +/// Whether `having` over `aggregate`'s output reads only `by(...)` group keys. +/// A global aggregate (no group keys) never qualifies: its one output row +/// exists even when no input row does, so no row predicate can stand in. +fn key_only_having(aggregate: &OperatorNode, having: &Predicate) -> bool { + let keys = match aggregate.non_asap() { + Some(NonASAPOp::Aggregate { + reduction: Reduction::Reduce(by), + .. + }) if !by.is_without() => by.len(), + _ => 0, + }; + keys > 0 + && having.0.operator_refs().is_empty() + && having.0.columns_referenced().iter().all(|id| *id < keys) +} + +fn placement(below: bool) -> HavingPlacement { + if below { + HavingPlacement::BelowAggregate + } else { + HavingPlacement::AboveOutput + } +} + +/// Discover one target per measure of each eligible aggregate, including +/// operator plans read by scalar roots. An aggregate measure is eligible when it +/// has a realization besides exact execution, even if every such realization +/// is pruned; the prune reasons are reported on the target. Measure filters +/// and `HAVING` keep every candidate; the target records how they are applied. +/// A consumer `Filter`'s key-only `HAVING` is pushed below only when every +/// consumer of the aggregate applies the same predicate. +pub fn generate_per_computation_candidates( + roots: Vec<(Id, QueryRoot)>, +) -> Result, PerComputationCandidateError> { + let mut seen = HashSet::new(); + // Each aggregate with the first root that reaches it. + let mut aggregates = Vec::new(); + // Every consumer of each node: `Some(parent)`, or `None` for a root or + // scalar reference. + let mut consumers: HashMap<*const OperatorNode, Vec>>> = HashMap::new(); + for (id, root) in &roots { + root.validate_structure()?; + let operators = match root { + QueryRoot::Operator(node) => vec![node], + QueryRoot::Scalar(expr) => expr.operator_refs(), + }; + for root in operators { + if root.contains_asap() { + return Err(PerComputationCandidateError::ContainsAsap); + } + consumers.entry(Rc::as_ptr(root)).or_default().push(None); + for node in OperatorNode::reachable(root) { + if !seen.insert(Rc::as_ptr(&node)) { + continue; + } + if node.timing.is_some() || node.materialization != Materialization::NotMaterialized + { + return Err(PerComputationCandidateError::AssignedPhysicalAnnotation); + } + for child in node.children() { + consumers + .entry(Rc::as_ptr(child)) + .or_default() + .push(Some(Rc::clone(&node))); + } + if matches!(node.non_asap(), Some(NonASAPOp::Aggregate { .. })) { + aggregates.push((format!("{id:?}"), node)); + } + } + } + } + let mut targets = Vec::new(); + for (root, node) in aggregates { + let Some(NonASAPOp::Aggregate { + measures, + filters, + having, + .. + }) = node.non_asap() + else { + continue; + }; + let mut entries: Vec = having + .iter() + .map(|having| HavingEntry { + filter: None, + predicate: having.clone(), + placement: placement(key_only_having(&node, having)), + }) + .collect(); + let parents = &consumers[&Rc::as_ptr(&node)]; + // Only a `Filter` whose input is this aggregate is its `HAVING`; one + // that reads it from a scalar expression stays a plain consumer. + let mut filters_above: Vec<(&Rc, &Predicate)> = Vec::new(); + for parent in parents.iter().flatten() { + if let Some(NonASAPOp::Filter { child, pred }) = parent.non_asap() { + if Rc::ptr_eq(child, &node) + && !filters_above + .iter() + .any(|(filter, _)| Rc::ptr_eq(filter, parent)) + { + filters_above.push((parent, pred)); + } + } + } + for (filter, predicate) in &filters_above { + let shared_by_all = filters_above.len() == parents.len() + && filters_above.iter().all(|(_, other)| other == predicate); + entries.push(HavingEntry { + filter: Some(Rc::clone(filter)), + predicate: (*predicate).clone(), + placement: placement(shared_by_all && key_only_having(&node, predicate)), + }); + } + for (measure, intent) in measures.iter().enumerate() { + let mut choices = + computation_candidates_for_intent(intent).map_err(|error| match error { + PerComputationCandidateError::InvalidAccuracy { .. } => { + PerComputationCandidateError::InvalidAccuracy { + root: Some(root.clone()), + measure: Some(measure), + } + } + error => error, + })?; + if choices.candidates == [ComputationCandidate::Exact] && choices.pruned.is_empty() { + continue; + } + if matches!(intent, AggIntent::Avg { .. }) && nullable_avg_input(&node, intent) { + choices.prune(PruneReason::NullableAvgInput, |choice| { + *choice == ComputationCandidate::AvgAsSumCount + }); + } + targets.push(ComputationTarget { + target: node.clone(), + measure, + choices, + filter: filters.get(measure).cloned().flatten(), + having: entries.clone(), + }); + } + } + Ok(PerComputationCandidates { roots, targets }) +} + +#[cfg(test)] +mod tests { + use super::*; + /// Approximate count keeps one exact option, its accumulator, beside the + /// universal summary; plain exact execution is not offered. + #[test] + fn approximate_count_keeps_one_exact_accumulator_and_universal_choices() { + let choices = computation_candidates_for_intent(&AggIntent::Count { + accuracy: AccuracyTarget::EpsilonDelta { + epsilon: 0.05, + delta: 0.01, + }, + }) + .unwrap() + .candidates; + assert!(!choices.contains(&ComputationCandidate::Exact)); + assert_eq!( + choices[0], + ComputationCandidate::ExactAccumulator { + kind: ExactKind::Count, + params: ExactParams::Count + } + ); + assert!(choices.contains(&ComputationCandidate::Summary(SketchAlgorithm::UnivMon))); + } +} diff --git a/crates/asap-aware-mapping/tests/per_computation_candidates.rs b/crates/asap-aware-mapping/tests/per_computation_candidates.rs new file mode 100644 index 000000000..fe3bde190 --- /dev/null +++ b/crates/asap-aware-mapping/tests/per_computation_candidates.rs @@ -0,0 +1,852 @@ +//! Frontend-to-Pass-1 acceptance: candidate discovery precedes empirical selection. +use asap_aware_mapping::per_computation_candidates::{ + computation_candidates_for_intent, generate_per_computation_candidates, ComputationCandidate, + HavingEntry, HavingPlacement, PerComputationCandidateError, PruneReason, +}; +use asap_types::{ + ir::operator_properties::{Reduction, Source}, + ir::{ + ASAPOp, ExprSemantics, NonASAPOp, Operator, OperatorNode, Predicate, QueryRoot, ScalarExpr, + }, + post_asap::{ + ExactKind, ExactParams, GroupingStrategy, SketchAlgorithm, SketchKind, SketchParams, + SummaryUpdate, + }, + pre_asap::{ + AggIntent, ColumnRef, CompareOpKind, DataType, Field, FieldDataType, ScalarValue, Schema, + }, + types::AccuracyTarget, +}; +use std::rc::Rc; + +fn approximate() -> AccuracyTarget { + AccuracyTarget::EpsilonDelta { + epsilon: 0.05, + delta: 0.01, + } +} +fn candidates_for(intent: AggIntent) -> Vec { + computation_candidates_for_intent(&intent) + .unwrap() + .candidates +} +fn algorithms(choices: &[ComputationCandidate]) -> Vec { + choices + .iter() + .filter_map(|choice| match choice { + ComputationCandidate::Summary(algorithm) => Some(algorithm.clone()), + _ => None, + }) + .collect() +} +fn scan(nullable_value: bool) -> Rc { + OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Scan { + source: Source::Table { + table_ref: "flows".into(), + }, + predicates: vec![], + schema: Schema::lifted( + vec![ + Field::plain("src_ip", DataType::Utf8, false), + Field::plain("value", DataType::Float64, nullable_value), + ], + None, + ), + })) + .unwrap() +} +fn aggregate_over( + child: Rc, + measures: Vec, + filters: Vec>, + having: Option, +) -> Rc { + OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Aggregate { + child, + reduction: Reduction::by(vec![]), + measures, + output_names: vec![], + filters, + having, + })) + .unwrap() +} +fn aggregate(intent: AggIntent) -> Rc { + aggregate_over(scan(false), vec![intent], vec![], None) +} +fn always() -> Predicate { + Predicate(ScalarExpr::Literal(ScalarValue::Boolean(true))) +} + +/// Example 2 preserves specialized distinct summaries and the universal alternative. +#[test] +fn cardinality_keeps_exact_specialized_and_universal_candidates() { + let choices = candidates_for(AggIntent::Cardinality { + cols: vec![0], + accuracy: approximate(), + }); + assert_eq!(choices[0], ComputationCandidate::Exact); + assert_eq!( + algorithms(&choices), + vec![ + SketchAlgorithm::Hll, + SketchAlgorithm::Theta, + SketchAlgorithm::Kmv, + SketchAlgorithm::UnivMon + ] + ); + let tuple = candidates_for(AggIntent::Cardinality { + cols: vec![0, 1], + accuracy: approximate(), + }); + assert!(!algorithms(&tuple).contains(&SketchAlgorithm::UnivMon)); +} + +/// Frequency moments retain exact execution and a universal sketch without certification. +#[test] +fn frequency_statistics_keep_universal_choices() { + for intent in [ + AggIntent::FrequencyL2 { + col: Some(0), + accuracy: approximate(), + }, + AggIntent::FrequencyEntropy { + col: Some(0), + accuracy: approximate(), + }, + ] { + let choices = candidates_for(intent); + assert_eq!(choices[0], ComputationCandidate::Exact); + assert_eq!(algorithms(&choices), vec![SketchAlgorithm::UnivMon]); + } +} + +/// An exact request cannot acquire an approximate sketch merely because one is +/// available; each pruned family records why. +#[test] +fn exact_quantile_stays_exact_and_approximate_keeps_both_families() { + let choices = candidates_for(AggIntent::Quantile { + col: Some(0), + q: 0.99, + accuracy: approximate(), + }); + assert_eq!( + algorithms(&choices), + vec![SketchAlgorithm::Kll, SketchAlgorithm::DDSketch] + ); + let exact = computation_candidates_for_intent(&AggIntent::Quantile { + col: Some(0), + q: 0.99, + accuracy: AccuracyTarget::Exact, + }) + .unwrap(); + assert_eq!(exact.candidates, vec![ComputationCandidate::Exact]); + assert_eq!( + exact.pruned, + vec![ + ( + ComputationCandidate::Summary(SketchAlgorithm::Kll), + PruneReason::RequiresExact + ), + ( + ComputationCandidate::Summary(SketchAlgorithm::DDSketch), + PruneReason::RequiresExact + ), + ] + ); +} + +/// Scalar roots expose their producer targets; repeated references retain one target identity. +#[test] +fn scalar_root_producers_are_discovered_once() { + let producer = aggregate(AggIntent::Cardinality { + cols: vec![0], + accuracy: approximate(), + }); + let roots = vec![ + ( + "scalar", + QueryRoot::Scalar(ScalarExpr::ScalarSubquery(producer.clone())), + ), + ("relation", QueryRoot::Operator(producer.clone())), + ]; + let candidates = generate_per_computation_candidates(roots).unwrap(); + assert_eq!(candidates.roots.len(), 2); + for (_, root) in &candidates.roots { + root.validate_structure().unwrap(); + } + assert_eq!(candidates.targets.len(), 1); + assert!(Rc::ptr_eq(&candidates.targets[0].target, &producer)); + assert!(producer.timing.is_none()); + assert!(producer.guarantee.is_none()); + producer.validate_structure().unwrap(); +} + +/// #509 Example 1 Q1 is exact: each target's one exact option is its exact +/// accumulator state (grouped sum, rate); no plain exact form and no summary. +/// One ImplicitExact PromQL query lowered through the workload frontend. +fn implicit_exact_promql(query: &str) -> Vec { + use asap_types::workload::{ + AccuracyRequirement, BatchEntry, PlanningWorkload, Query, QueryLanguage, QueryRequirements, + QueryWorkload, + }; + let workload = PlanningWorkload { + query_workload: QueryWorkload { + language: QueryLanguage::PromQL, + query_batch: Some(vec![BatchEntry { + query: Query(query.into()), + requirements: QueryRequirements { + accuracy: AccuracyRequirement::ImplicitExact, + ..Default::default() + }, + predictability: Default::default(), + invocations: 1, + execute_at: None, + time_selection: Default::default(), + }]), + repeating_queries: None, + }, + data_workload: Some(asap_types::workload::DataWorkload { + data_ingestion_interval: asap_types::workload::Evidence { + value: Some(asap_types::workload::DurationMs(1000)), + ..Default::default() + }, + ..Default::default() + }), + }; + asap_frontend_promql::lower_promql_query_workload(&workload, 0).unwrap() +} + +#[test] +fn example_1_exact_q1_keeps_exact_accumulators_only() { + let roots = implicit_exact_promql("sum by (job) (rate(http_requests_total[1m]))"); + let candidates = + generate_per_computation_candidates(roots.into_iter().enumerate().collect()).unwrap(); + let found: Vec<_> = candidates + .targets + .iter() + .map(|target| target.choices.candidates.clone()) + .collect(); + assert_eq!( + found, + vec![ + vec![ComputationCandidate::ExactAccumulator { + kind: ExactKind::Sum, + params: ExactParams::Sum + }], + vec![ComputationCandidate::ExactAccumulator { + kind: ExactKind::Rate, + params: ExactParams::Rate + }], + ] + ); + assert!(candidates + .targets + .iter() + .all(|target| target.target.timing.is_none())); +} + +/// Physical annotations, ASAP operators and invalid probability requirements +/// fail at the stage boundary. +#[test] +fn assigned_physical_annotation_asap_input_and_invalid_accuracy_are_rejected() { + let mut producer = (*aggregate(AggIntent::Count { + accuracy: approximate(), + })) + .clone(); + producer.timing = Some(asap_types::post_asap::ExecutionTiming::QueryTime); + assert!(matches!( + generate_per_computation_candidates(vec![( + 0, + QueryRoot::Operator(Rc::new(producer.clone())) + )]), + Err(PerComputationCandidateError::AssignedPhysicalAnnotation) + )); + let mut stored = (*aggregate(AggIntent::Count { + accuracy: approximate(), + })) + .clone(); + stored.materialization = + asap_types::post_asap::Materialization::MatAtQuery(asap_types::post_asap::Medium::Memory); + assert!(matches!( + generate_per_computation_candidates(vec![( + 0, + QueryRoot::Operator(Rc::new(stored.clone())) + )]), + Err(PerComputationCandidateError::AssignedPhysicalAnnotation) + )); + // An annotated node below an unannotated root is rejected too. + for annotated in [producer, stored] { + let root = having_filter(Rc::new(annotated), always()); + assert!(matches!( + generate_per_computation_candidates(vec![(0, QueryRoot::Operator(root))]), + Err(PerComputationCandidateError::AssignedPhysicalAnnotation) + )); + } + let summary = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryAgg { + child: scan(false), + family: FieldDataType::Sketch( + SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 200 }), + Default::default(), + ), + input: SummaryUpdate::column(ColumnRef::SampleValue), + reduction: Reduction::by(vec![]), + grouping: GroupingStrategy::default(), + filter: None, + })) + .unwrap(); + assert!(matches!( + generate_per_computation_candidates(vec![(0, QueryRoot::Operator(summary))]), + Err(PerComputationCandidateError::ContainsAsap) + )); + let epsilon_delta = |epsilon, delta| AccuracyTarget::EpsilonDelta { epsilon, delta }; + for target in [ + AccuracyTarget::Epsilon(f64::NAN), + epsilon_delta(0.0, 0.01), + epsilon_delta(-0.1, 0.01), + epsilon_delta(1.0, 0.01), + epsilon_delta(f64::INFINITY, 0.01), + epsilon_delta(0.1, 0.0), + epsilon_delta(0.1, 1.0), + ] { + assert!( + matches!( + computation_candidates_for_intent(&AggIntent::Count { + accuracy: target.clone() + }), + Err(PerComputationCandidateError::InvalidAccuracy { + root: None, + measure: None + }) + ), + "{target:?}" + ); + } + let invalid = aggregate_over( + scan(false), + vec![ + AggIntent::Count { + accuracy: approximate(), + }, + AggIntent::Count { + accuracy: epsilon_delta(1.5, 0.01), + }, + ], + vec![], + None, + ); + let error = + generate_per_computation_candidates(vec![("q", QueryRoot::Operator(invalid))]).unwrap_err(); + assert!( + matches!( + &error, + PerComputationCandidateError::InvalidAccuracy { + root: Some(root), + measure: Some(1) + } if root == "\"q\"" + ), + "{error:?}" + ); + assert!( + error.to_string().contains("root \"q\", measure 1 "), + "{error}" + ); +} + +/// TopK keeps both declared heap substrates without choosing an implementation. +#[test] +fn topk_keeps_both_specialized_heap_choices() { + let choices = candidates_for(AggIntent::TopK { + k: 10, + accuracy: approximate(), + }); + assert_eq!(choices[0], ComputationCandidate::Exact); + assert_eq!( + algorithms(&choices), + vec![ + SketchAlgorithm::CmsWithHeap, + SketchAlgorithm::CountSketchWithHeap + ] + ); +} + +/// Summary candidates are unsized, so they do not depend on the requested error. +#[test] +fn summary_candidates_are_unsized() { + let loose = candidates_for(AggIntent::Count { + accuracy: approximate(), + }); + let tight = candidates_for(AggIntent::Count { + accuracy: AccuracyTarget::EpsilonDelta { + epsilon: 0.0001, + delta: 0.0001, + }, + }); + assert_eq!(loose, tight); +} + +/// Each measure of a multi-measure aggregate is its own target; a measure whose +/// only realization is its exact accumulator is still a target. +#[test] +fn multi_measure_aggregate_has_one_target_per_measure() { + let node = aggregate_over( + scan(false), + vec![ + AggIntent::Sum { col: Some(1) }, + AggIntent::Count { + accuracy: approximate(), + }, + ], + vec![], + None, + ); + let candidates = generate_per_computation_candidates(vec![(0, QueryRoot::Operator(node))]) + .unwrap() + .targets; + assert_eq!( + candidates + .iter() + .map(|target| target.measure) + .collect::>(), + vec![0, 1] + ); + assert_eq!( + candidates[0].choices.candidates, + vec![ComputationCandidate::ExactAccumulator { + kind: ExactKind::Sum, + params: ExactParams::Sum + }] + ); + assert_eq!( + algorithms(&candidates[1].choices.candidates), + vec![ + SketchAlgorithm::Cms, + SketchAlgorithm::CountSketch, + SketchAlgorithm::UnivMon + ] + ); +} + +/// An aggregate with no realization besides exact execution is not a target. +#[test] +fn aggregate_without_realization_is_not_a_target() { + let node = aggregate(AggIntent::StdDev { + col: Some(1), + population: true, + }); + assert!( + generate_per_computation_candidates(vec![(0, QueryRoot::Operator(node))]) + .unwrap() + .targets + .is_empty() + ); +} + +/// A filtered measure keeps its accumulator and summaries and records its +/// filter, which composition applies to the state. +#[test] +fn filtered_measure_keeps_candidates_and_records_its_filter() { + let count = || AggIntent::Count { + accuracy: approximate(), + }; + let filtered = aggregate_over( + scan(false), + vec![count(), count()], + vec![Some(always()), None], + None, + ); + let targets = generate_per_computation_candidates(vec![(0, QueryRoot::Operator(filtered))]) + .unwrap() + .targets; + assert_eq!(targets.len(), 2); + for target in &targets { + assert_eq!( + target.choices.candidates, + candidates_for(count()), + "measure {}", + target.measure + ); + assert!(target.choices.pruned.is_empty()); + assert!(target.having.is_empty()); + } + assert_eq!(targets[0].filter, Some(always())); + assert_eq!(targets[1].filter, None); +} + +/// `count(*) by (src_ip)`, the form the SQL frontend gives `GROUP BY src_ip`. +fn grouped_count(having: Option) -> Rc { + 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![], + having, + })) + .unwrap() +} +/// `HAVING` as the SQL frontend lowers it: a `Filter` above the aggregate. +fn having_filter(child: Rc, pred: Predicate) -> Rc { + OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Filter { pred, child })).unwrap() +} +/// `column = 'a'` over the aggregate output (0: `src_ip`, 1: the count). +fn equals(column: usize, value: ScalarValue) -> Predicate { + Predicate(ScalarExpr::Compare { + left: Box::new(ScalarExpr::Column(column)), + op: CompareOpKind::Eq, + right: Box::new(ScalarExpr::Literal(value)), + semantics: ExprSemantics::Sql, + }) +} + +/// A `HAVING` over group keys only is pushed below the aggregate and keeps +/// every candidate. +#[test] +fn key_only_having_is_pushed_below_and_keeps_candidates() { + let node = grouped_count(None); + let root = having_filter(node.clone(), equals(0, ScalarValue::Utf8("a".into()))); + let targets = generate_per_computation_candidates(vec![(0, QueryRoot::Operator(root.clone()))]) + .unwrap() + .targets; + assert_eq!(targets.len(), 1); + assert!(Rc::ptr_eq(&targets[0].target, &node)); + assert_eq!( + targets[0].having, + vec![HavingEntry { + filter: Some(root.clone()), + predicate: equals(0, ScalarValue::Utf8("a".into())), + placement: HavingPlacement::BelowAggregate, + }] + ); + assert_eq!( + targets[0].choices.candidates, + candidates_for(AggIntent::Count { + accuracy: approximate() + }) + ); + assert!(targets[0].choices.pruned.is_empty()); +} + +/// A `HAVING` over an aggregate value, as a `Filter` above or as the +/// aggregate's own `having`, stays above the output and keeps every candidate. +#[test] +fn value_having_stays_above_output_and_keeps_candidates() { + let value = || equals(1, ScalarValue::Int64(10)); + let targets = generate_per_computation_candidates(vec![ + ( + 0, + QueryRoot::Operator(having_filter(grouped_count(None), value())), + ), + (1, QueryRoot::Operator(grouped_count(Some(value())))), + ]) + .unwrap() + .targets; + assert_eq!(targets.len(), 2); + for target in &targets { + let placements: Vec<_> = target.having.iter().map(|entry| entry.placement).collect(); + assert_eq!(placements, vec![HavingPlacement::AboveOutput]); + assert_eq!( + target.choices.candidates, + candidates_for(AggIntent::Count { + accuracy: approximate() + }) + ); + assert!(target.choices.pruned.is_empty()); + } +} + +/// `avg`'s one exact option is sum and count states when its input is never +/// null, and plain `Exact` when that rewrite is pruned. +#[test] +fn avg_rewrites_to_sum_and_count_states() { + let avg = || vec![AggIntent::Avg { col: None }]; + let targets = generate_per_computation_candidates(vec![ + ( + 0, + QueryRoot::Operator(aggregate_over(scan(false), avg(), vec![], None)), + ), + ( + 1, + QueryRoot::Operator(aggregate_over(scan(true), avg(), vec![], None)), + ), + ]) + .unwrap() + .targets; + assert_eq!( + targets[0].choices.candidates, + vec![ComputationCandidate::AvgAsSumCount] + ); + assert_eq!( + targets[1].choices.candidates, + vec![ComputationCandidate::Exact] + ); + assert_eq!( + targets[1].choices.pruned, + vec![( + ComputationCandidate::AvgAsSumCount, + PruneReason::NullableAvgInput + )] + ); +} + +/// A key-only `HAVING` above a shared aggregate stays above the output when +/// another consumer reads the aggregate without it. +#[test] +fn having_on_one_consumer_of_a_shared_aggregate_stays_above() { + let node = grouped_count(None); + let key = equals(0, ScalarValue::Utf8("a".into())); + let filtered = having_filter(node.clone(), key.clone()); + let targets = generate_per_computation_candidates(vec![ + (0, QueryRoot::Operator(filtered.clone())), + (1, QueryRoot::Operator(node.clone())), + ]) + .unwrap() + .targets; + assert_eq!(targets.len(), 1); + assert_eq!( + targets[0].having, + vec![HavingEntry { + filter: Some(filtered), + predicate: key.clone(), + placement: HavingPlacement::AboveOutput, + }] + ); + + // Two consumers applying the same key-only predicate push it below. + let first = having_filter(node.clone(), key.clone()); + let second = having_filter(node, key); + let targets = generate_per_computation_candidates(vec![ + (0, QueryRoot::Operator(first)), + (1, QueryRoot::Operator(second)), + ]) + .unwrap() + .targets; + assert!(targets[0] + .having + .iter() + .all(|entry| entry.placement == HavingPlacement::BelowAggregate)); + assert_eq!(targets[0].having.len(), 2); +} + +/// A global aggregate's `HAVING` always stays above the output, even a +/// constant one: the one output row exists with no input rows. +#[test] +fn global_aggregate_having_stays_above() { + let count = || { + vec![AggIntent::Count { + accuracy: approximate(), + }] + }; + let never = Predicate(ScalarExpr::Literal(ScalarValue::Boolean(false))); + let targets = generate_per_computation_candidates(vec![ + ( + 0, + QueryRoot::Operator(having_filter( + aggregate_over(scan(false), count(), vec![], None), + never.clone(), + )), + ), + ( + 1, + QueryRoot::Operator(aggregate_over(scan(false), count(), vec![], Some(never))), + ), + ]) + .unwrap() + .targets; + assert_eq!(targets.len(), 2); + for target in &targets { + let placements: Vec<_> = target.having.iter().map(|entry| entry.placement).collect(); + assert_eq!(placements, vec![HavingPlacement::AboveOutput]); + } +} + +/// #509 Example 1 Q2 under ImplicitExact: the top-k's one option is plain +/// `Exact`, its heap summaries pruned `RequiresExact`; `sum_over_time` gets +/// its exact Sum accumulator. +#[test] +fn example_1_exact_q2_prunes_topk_heaps() { + let roots = implicit_exact_promql("topk by (job) (10, sum_over_time(http_requests_total[1m]))"); + let candidates = + generate_per_computation_candidates(roots.into_iter().enumerate().collect()).unwrap(); + let found: Vec<_> = candidates + .targets + .iter() + .map(|target| target.choices.clone()) + .collect(); + let topk = found + .iter() + .find(|choices| choices.candidates == [ComputationCandidate::Exact]) + .expect("top-k target"); + assert_eq!( + topk.pruned, + vec![ + ( + ComputationCandidate::Summary(SketchAlgorithm::CmsWithHeap), + PruneReason::RequiresExact + ), + ( + ComputationCandidate::Summary(SketchAlgorithm::CountSketchWithHeap), + PruneReason::RequiresExact + ), + ] + ); + assert!(found.iter().any(|choices| choices.candidates + == [ComputationCandidate::ExactAccumulator { + kind: ExactKind::Sum, + params: ExactParams::Sum + }] + && choices.pruned.is_empty())); + assert_eq!(found.len(), 2, "{found:#?}"); +} + +/// The `HAVING` entries of the target whose measure satisfies `is`. +fn having_of( + targets: &[asap_aware_mapping::per_computation_candidates::ComputationTarget], + is: impl Fn(&AggIntent) -> bool, +) -> Vec { + let target = targets + .iter() + .find(|target| match target.target.non_asap() { + Some(NonASAPOp::Aggregate { measures, .. }) => is(&measures[target.measure]), + _ => false, + }) + .expect("target"); + target.having.clone() +} + +/// `Column(column) > ScalarSubquery(subquery)`. +fn above_scalar(column: usize, subquery: Rc) -> Predicate { + Predicate(ScalarExpr::Compare { + left: Box::new(ScalarExpr::Column(column)), + op: CompareOpKind::Gt, + right: Box::new(ScalarExpr::ScalarSubquery(subquery)), + semantics: ExprSemantics::Sql, + }) +} + +/// A `Filter` that reads an aggregate only from its predicate is not that +/// aggregate's `HAVING`; one that also filters the aggregate gives one entry. +#[test] +fn scalar_reference_filter_is_not_having() { + let count = || { + vec![AggIntent::Count { + accuracy: approximate(), + }] + }; + let agg = aggregate_over(scan(false), count(), vec![], None); + let root = having_filter(scan(false), above_scalar(1, agg.clone())); + let targets = generate_per_computation_candidates(vec![(0, QueryRoot::Operator(root))]) + .unwrap() + .targets; + assert_eq!(targets.len(), 1); + assert!(Rc::ptr_eq(&targets[0].target, &agg)); + assert!(targets[0].having.is_empty(), "{:?}", targets[0].having); + + let agg = aggregate_over(scan(false), count(), vec![], None); + let root = having_filter(agg.clone(), above_scalar(0, agg.clone())); + let targets = generate_per_computation_candidates(vec![(0, QueryRoot::Operator(root.clone()))]) + .unwrap() + .targets; + assert_eq!(targets.len(), 1); + assert_eq!( + targets[0].having, + vec![HavingEntry { + filter: Some(root), + predicate: above_scalar(0, agg), + placement: HavingPlacement::AboveOutput, + }] + ); +} + +/// PromQL `scalar(...)` operands are scalar references, not `HAVING`. +#[test] +fn promql_scalar_operand_is_not_having() { + let roots = implicit_exact_promql("http_requests_total > scalar(sum(http_requests_total))"); + let targets = generate_per_computation_candidates(roots.into_iter().enumerate().collect()) + .unwrap() + .targets; + assert!(!targets.is_empty()); + for target in &targets { + assert!(target.having.is_empty(), "{:?}", target.having); + } + + let roots = implicit_exact_promql("sum by (job) (x) > scalar(count(x))"); + let targets = generate_per_computation_candidates(roots.into_iter().enumerate().collect()) + .unwrap() + .targets; + assert!(having_of(&targets, |intent| matches!(intent, AggIntent::Count { .. })).is_empty()); + let sum = having_of(&targets, |intent| matches!(intent, AggIntent::Sum { .. })); + assert_eq!(sum.len(), 1, "{sum:?}"); + assert_eq!(sum[0].placement, HavingPlacement::AboveOutput); +} + +/// A non-root, non-`Filter` consumer keeps a key-only `HAVING` above. +#[test] +fn having_shared_with_a_project_consumer_stays_above() { + let node = grouped_count(None); + let key = equals(0, ScalarValue::Utf8("a".into())); + let filtered = having_filter(node.clone(), key); + let project = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Project { + cols: vec![asap_types::ir::ProjectItem { + alias: None, + expr: ScalarExpr::Column(0), + }], + qualifier: None, + child: node, + })) + .unwrap(); + let targets = generate_per_computation_candidates(vec![ + (0, QueryRoot::Operator(filtered)), + (1, QueryRoot::Operator(project)), + ]) + .unwrap() + .targets; + assert_eq!(targets.len(), 1); + let placements: Vec<_> = targets[0] + .having + .iter() + .map(|entry| entry.placement) + .collect(); + assert_eq!(placements, vec![HavingPlacement::AboveOutput]); +} + +/// `without(...)` and per-entity aggregates have no fixed key prefix, so a +/// `HAVING` over column 0 stays above. +#[test] +fn without_and_per_entity_having_stays_above() { + for reduction in [ + Reduction::Reduce(asap_types::ir::operator_properties::GroupKeys::without( + vec![1], + )), + Reduction::PerEntity, + ] { + let node = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Aggregate { + child: scan(false), + reduction: reduction.clone(), + measures: vec![AggIntent::Count { + accuracy: approximate(), + }], + output_names: vec![], + filters: vec![], + having: None, + })) + .unwrap(); + let root = having_filter(node, equals(0, ScalarValue::Utf8("a".into()))); + let targets = generate_per_computation_candidates(vec![(0, QueryRoot::Operator(root))]) + .unwrap() + .targets; + assert_eq!(targets.len(), 1, "{reduction:?}"); + let placements: Vec<_> = targets[0] + .having + .iter() + .map(|entry| entry.placement) + .collect(); + assert_eq!( + placements, + vec![HavingPlacement::AboveOutput], + "{reduction:?}" + ); + } +} diff --git a/crates/integration-tests/tests/per_computation_candidates_sql.rs b/crates/integration-tests/tests/per_computation_candidates_sql.rs new file mode 100644 index 000000000..f0d78f526 --- /dev/null +++ b/crates/integration-tests/tests/per_computation_candidates_sql.rs @@ -0,0 +1,63 @@ +//! SQL `HAVING` → Pass 1 per-computation candidates: where each `HAVING` is applied. + +use asap_aware_mapping::per_computation_candidates::{ + generate_per_computation_candidates, ComputationTarget, HavingPlacement, +}; +use asap_frontend_sql::{lower_sql, SqlCatalog}; +use asap_types::ir::QueryRoot; +use asap_types::pre_asap::schema::{DataType, Field, Schema}; +use asap_types::types::AccuracyTarget; + +async fn targets(sql: &str) -> Vec { + let catalog = SqlCatalog::new().with_table( + "flows", + Schema::new(vec![ + Field::plain("src", DataType::Utf8, false), + Field::plain("bytes", DataType::Float64, false), + ]), + ); + let accuracy = AccuracyTarget::EpsilonDelta { + epsilon: 0.01, + delta: 0.01, + }; + let root = lower_sql(sql, &catalog, accuracy) + .await + .unwrap_or_else(|e| panic!("lower failed for {sql:?}: {e}")); + generate_per_computation_candidates(vec![(0, QueryRoot::Operator(root))]) + .unwrap() + .targets +} + +fn placements(targets: &[ComputationTarget]) -> Vec { + targets + .iter() + .flat_map(|target| target.having.iter().map(|entry| entry.placement)) + .collect() +} + +#[tokio::test] +async fn key_only_having_is_pushed_below() { + let targets = targets("SELECT src, count(*) FROM flows GROUP BY src HAVING src = 'a'").await; + assert_eq!(placements(&targets), vec![HavingPlacement::BelowAggregate]); +} + +#[tokio::test] +async fn value_having_stays_above() { + let targets = + targets("SELECT src, count(*) FROM flows GROUP BY src HAVING count(*) > 10").await; + assert_eq!(placements(&targets), vec![HavingPlacement::AboveOutput]); +} + +#[tokio::test] +async fn global_constant_having_stays_above() { + let targets = targets("SELECT count(*) FROM flows HAVING false").await; + assert_eq!(placements(&targets), vec![HavingPlacement::AboveOutput]); +} + +#[tokio::test] +async fn mixed_key_and_value_having_stays_above() { + let targets = + targets("SELECT src, count(*) FROM flows GROUP BY src HAVING src = 'a' AND count(*) > 10") + .await; + assert_eq!(placements(&targets), vec![HavingPlacement::AboveOutput]); +} diff --git a/docs/design_docs/proposals/planner-layering.md b/docs/design_docs/proposals/planner-layering.md index 93f353f56..dbfc7be12 100644 --- a/docs/design_docs/proposals/planner-layering.md +++ b/docs/design_docs/proposals/planner-layering.md @@ -319,9 +319,14 @@ Each target gets one exact option. If its intent has an exact mergeable accumulator (Count, Sum, Min, Max, Rate, IRate, Increase; avg via exact sum and count accumulator states) and the accumulator can be built, the exact option is **exact accumulator state**: a `SummaryAgg` with an exact family → -`FinalizeExactAccumulator`. Otherwise, for example when a filter or HAVING -prevents the accumulator or the intent has none, it is the plain exact -computation. Whether the accumulator is materialized is decided in Stage 2. +`FinalizeExactAccumulator`. Otherwise, when the intent has no exact +accumulator or it cannot be built (for example avg over a nullable input), it +is the plain exact computation. A filter or HAVING does not prevent the +accumulator or summary candidates: a per-measure filter goes into +`SummaryAgg.filter`, a HAVING that reads only group keys is pushed below the +aggregate as a row predicate, and a HAVING that reads aggregate values is a +`Filter` above the output, with `guarantee = None` for approximate +candidates. Whether the accumulator is materialized is decided in Stage 2. Each candidate records its input expression, filter, grouping, window, supported estimates and accuracy requirement. Pass 2 uses these to decide @@ -400,7 +405,7 @@ flowchart LR subgraph R2["Replacements(s₂)"] direction TB d3["Exact Sum accumulator
per job"]:::exact - a2["summary candidates
✗ pruned: cannot be exact"]:::pruned + a2["no summary candidates:
Sum has no summary family"]:::exact end subgraph R3["Replacements(s₃)"] direction TB