From 406e3f70a3f31017ed7b639817ae75e7b18cfd0c Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Sat, 3 Oct 2026 14:50:55 +0000 Subject: [PATCH 1/9] feat(planner): enumerate phase-free local logical alternatives --- crates/asap-aware-mapping/src/lib.rs | 3 + .../src/logical_candidates.rs | 147 +++++++++++ .../tests/logical_candidates.rs | 243 ++++++++++++++++++ docs/develop_docs/local-logical-candidates.md | 39 +++ 4 files changed, 432 insertions(+) create mode 100644 crates/asap-aware-mapping/src/logical_candidates.rs create mode 100644 crates/asap-aware-mapping/tests/logical_candidates.rs create mode 100644 docs/develop_docs/local-logical-candidates.md diff --git a/crates/asap-aware-mapping/src/lib.rs b/crates/asap-aware-mapping/src/lib.rs index b2a1df3da..c9becbc25 100644 --- a/crates/asap-aware-mapping/src/lib.rs +++ b/crates/asap-aware-mapping/src/lib.rs @@ -211,3 +211,6 @@ pub use rewrite::{AvgToSumOverCountStrategy, SemanticEquivalentRewriteStrategy}; pub use topk_reuse::TopKLimitReuseStrategy; pub mod maintained_population; + +/// Phase-free local candidate generation over the unified IR. +pub mod logical_candidates; diff --git a/crates/asap-aware-mapping/src/logical_candidates.rs b/crates/asap-aware-mapping/src/logical_candidates.rs new file mode 100644 index 000000000..8c8a28913 --- /dev/null +++ b/crates/asap-aware-mapping/src/logical_candidates.rs @@ -0,0 +1,147 @@ +//! Pass 1 local alternatives over the unified logical IR. +//! +//! Alternatives are nominal realization descriptors attached to their original +//! target, not ranked plans or accuracy certificates. Workload composition and +//! physical planning consume this inventory later; empirical models belong to +//! selection. The legacy search API remains until planner cutover. +use std::collections::HashSet; +use std::rc::Rc; + +use asap_types::ir::{NonASAPOp, OperatorNode, QueryRoot, SchemaDerivationError}; +use asap_types::post_asap::{ExactKind, ExactParams, SketchKind}; +use asap_types::pre_asap::AggIntent; +use asap_types::types::AccuracyTarget; +use thiserror::Error; + +use crate::replacement::{ + accuracy_budget, accuracy_target, default_size_params, summary_candidates, Realization, +}; + +/// All local realizations of one single-measure aggregate. The target retains +/// source, grouping, filters, input expressions and evaluation context. +#[derive(Debug, Clone)] +pub struct LocalLogicalTarget { + pub target: Rc, + pub alternatives: Vec, +} + +/// Compact Pass 1 inventory; roots and nested producer dependencies are retained. +#[derive(Debug, Clone)] +pub struct LocalLogicalCandidates { + pub roots: Vec<(Id, QueryRoot)>, + pub targets: Vec, +} + +#[derive(Debug, Error)] +pub enum LogicalCandidateError { + #[error(transparent)] + Structure(#[from] SchemaDerivationError), + #[error("logical candidate input already has assigned execution timing")] + AssignedTiming, + #[error("approximate accuracy requires finite positive epsilon and delta in (0, 1)")] + InvalidAccuracy, +} + +/// Enumerate exact and summary choices in stable catalog order, without ranking +/// or empirical assessment. Parameters are candidate dimensions, not a claim +/// that a deployment meets the request's accuracy requirement. +pub fn local_realizations_for_intent( + intent: &AggIntent, +) -> Result, LogicalCandidateError> { + let mut choices = vec![Realization::PassThrough]; + 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, + }; + if let Some((kind, params)) = exact { + choices.push(Realization::ExactAggregate { kind, params }); + } + if let Some(target) = accuracy_target(intent) { + if *target != AccuracyTarget::Exact { + let (epsilon, delta) = accuracy_budget(target); + if !epsilon.is_finite() + || epsilon <= 0.0 + || !delta.is_finite() + || !(0.0..1.0).contains(&delta) + || delta == 0.0 + { + return Err(LogicalCandidateError::InvalidAccuracy); + } + for algorithm in summary_candidates(intent) { + choices.push(Realization::Sketch(SketchKind::new( + algorithm.clone(), + default_size_params(algorithm.clone(), intent, epsilon, delta), + ))); + } + } + } + Ok(choices) +} + +/// Discover single-measure targets, including operator plans read by scalar roots. +/// Multi-measure aggregates remain intact pending a semantics-preserving split. +pub fn enumerate_local_logical_candidates( + roots: Vec<(Id, QueryRoot)>, +) -> Result, LogicalCandidateError> { + let mut seen = HashSet::new(); + let mut targets = Vec::new(); + for (_, 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 { + for node in OperatorNode::reachable(root) { + if !seen.insert(Rc::as_ptr(&node)) { + continue; + } + if node.timing.is_some() { + return Err(LogicalCandidateError::AssignedTiming); + } + if let Some(NonASAPOp::Aggregate { measures, .. }) = node.non_asap() { + if let [intent] = measures.as_slice() { + targets.push(LocalLogicalTarget { + alternatives: local_realizations_for_intent(intent)?, + target: node, + }); + } + } + } + } + } + Ok(LocalLogicalCandidates { roots, targets }) +} + +#[cfg(test)] +mod tests { + use super::*; + /// Approximate requests must retain the exact execution alternative too. + #[test] + fn approximate_count_keeps_exact_and_universal_choices() { + let choices = local_realizations_for_intent(&AggIntent::Count { + accuracy: AccuracyTarget::EpsilonDelta { + epsilon: 0.05, + delta: 0.01, + }, + }) + .unwrap(); + assert!(choices + .iter() + .any(|choice| matches!(choice, Realization::PassThrough))); + assert!(choices.iter().any(|choice| matches!( + choice, + Realization::ExactAggregate { + kind: ExactKind::Count, + .. + } + ))); + assert!(choices.iter().any(|choice| matches!(choice, Realization::Sketch(kind) if *kind.algorithm() == asap_types::post_asap::SketchAlgorithm::UnivMon))); + } +} diff --git a/crates/asap-aware-mapping/tests/logical_candidates.rs b/crates/asap-aware-mapping/tests/logical_candidates.rs new file mode 100644 index 000000000..a08c0739d --- /dev/null +++ b/crates/asap-aware-mapping/tests/logical_candidates.rs @@ -0,0 +1,243 @@ +//! Frontend-to-Pass-1 acceptance: candidate discovery precedes empirical selection. +use asap_aware_mapping::{ + logical_candidates::{ + enumerate_local_logical_candidates, local_realizations_for_intent, LogicalCandidateError, + }, + Realization, +}; +use asap_types::{ + ir::operator_properties::{Reduction, Source}, + ir::{NonASAPOp, Operator, OperatorNode, QueryRoot, ScalarExpr}, + post_asap::{ExactKind, SketchAlgorithm}, + pre_asap::{AggIntent, DataType, Field, Schema}, + types::AccuracyTarget, +}; +use std::rc::Rc; + +fn approximate() -> AccuracyTarget { + AccuracyTarget::EpsilonDelta { + epsilon: 0.05, + delta: 0.01, + } +} +fn algorithms(choices: &[Realization]) -> Vec { + choices + .iter() + .filter_map(|choice| match choice { + Realization::Sketch(kind) => Some(kind.algorithm().clone()), + _ => None, + }) + .collect() +} +fn aggregate(intent: AggIntent) -> Rc { + let child = 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)], None), + })) + .unwrap(); + OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Aggregate { + child, + reduction: Reduction::by(vec![]), + measures: vec![intent], + output_names: vec![], + filters: vec![], + having: None, + })) + .unwrap() +} + +/// Example 2 preserves specialized distinct summaries and the universal alternative. +#[test] +fn cardinality_keeps_exact_specialized_and_universal_alternatives() { + let choices = local_realizations_for_intent(&AggIntent::Cardinality { + cols: vec![0], + accuracy: approximate(), + }) + .unwrap(); + assert!(matches!(choices[0], Realization::PassThrough)); + assert_eq!( + algorithms(&choices), + vec![ + SketchAlgorithm::Hll, + SketchAlgorithm::Theta, + SketchAlgorithm::Kmv, + SketchAlgorithm::UnivMon + ] + ); + let tuple = local_realizations_for_intent(&AggIntent::Cardinality { + cols: vec![0, 1], + accuracy: approximate(), + }) + .unwrap(); + 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 = local_realizations_for_intent(&intent).unwrap(); + assert!(matches!(choices[0], Realization::PassThrough)); + assert_eq!(algorithms(&choices), vec![SketchAlgorithm::UnivMon]); + } +} + +/// An exact request cannot acquire an approximate sketch merely because one is available. +#[test] +fn exact_quantile_stays_exact_and_approximate_keeps_both_families() { + let choices = local_realizations_for_intent(&AggIntent::Quantile { + col: Some(0), + q: 0.99, + accuracy: approximate(), + }) + .unwrap(); + assert_eq!( + algorithms(&choices), + vec![SketchAlgorithm::Kll, SketchAlgorithm::DDSketch] + ); + let exact = local_realizations_for_intent(&AggIntent::Quantile { + col: Some(0), + q: 0.99, + accuracy: AccuracyTarget::Exact, + }) + .unwrap(); + assert_eq!(exact, vec![Realization::PassThrough]); +} + +/// 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 = enumerate_local_logical_candidates(roots).unwrap(); + assert_eq!(candidates.roots.len(), 2); + for (_, root) in &candidates.roots { + asap_types::ir::export::compile_logical_asap_query(root) + .unwrap() + .validate() + .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()); + asap_types::ir::export::compile_logical_asap_dag(&producer) + .unwrap() + .validate() + .unwrap(); +} + +/// Example 1 rate lowering reaches the exact accumulator choice without a cost model. +#[test] +fn promql_lowering_reaches_phase_free_local_candidates() { + 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("sum by (job) (rate(http_requests_total[1m]))".into()), + requirements: QueryRequirements { + accuracy: AccuracyRequirement::Explicit(approximate()), + ..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() + }), + }; + let roots = asap_frontend_promql::unified::lower_promql_query_workload(&workload, 0).unwrap(); + let candidates = + enumerate_local_logical_candidates(roots.into_iter().enumerate().collect()).unwrap(); + assert!(candidates + .targets + .iter() + .any(|target| target.alternatives.iter().any(|choice| matches!( + choice, + Realization::ExactAggregate { + kind: ExactKind::Rate, + .. + } + )))); + assert!(candidates + .targets + .iter() + .all(|target| target.target.timing.is_none())); +} + +/// Physical annotations and invalid probability requirements fail at the stage boundary. +#[test] +fn assigned_timing_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!( + enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(Rc::new(producer)))]), + Err(LogicalCandidateError::AssignedTiming) + )); + for target in [ + AccuracyTarget::Epsilon(f64::NAN), + AccuracyTarget::EpsilonDelta { + epsilon: 0.1, + delta: 0.0, + }, + ] { + assert!(matches!( + local_realizations_for_intent(&AggIntent::Count { accuracy: target }), + Err(LogicalCandidateError::InvalidAccuracy) + )); + } +} + +/// Local TopK keeps both declared heap substrates without choosing an implementation. +#[test] +fn topk_keeps_both_specialized_heap_choices() { + let choices = local_realizations_for_intent(&AggIntent::TopK { + k: 10, + accuracy: approximate(), + }) + .unwrap(); + assert!(matches!(choices[0], Realization::PassThrough)); + assert_eq!( + algorithms(&choices), + vec![ + SketchAlgorithm::CmsWithHeap, + SketchAlgorithm::CountSketchWithHeap + ] + ); +} diff --git a/docs/develop_docs/local-logical-candidates.md b/docs/develop_docs/local-logical-candidates.md new file mode 100644 index 000000000..b27e9b2d2 --- /dev/null +++ b/docs/develop_docs/local-logical-candidates.md @@ -0,0 +1,39 @@ +# Local logical alternatives (Pass 1) + +`asap_aware_mapping::logical_candidates` enumerates local realization choices over +unified `OperatorNode` and `QueryRoot` inputs. It is the first part of logical +ASAP optimization in [planner layering](../design_docs/proposals/planner-layering.md). + +`enumerate_local_logical_candidates(roots)` returns a `LocalLogicalCandidates` +inventory containing the original named roots and one `LocalLogicalTarget` per +reachable single-measure aggregate. Discovery includes operator producers read by +scalar roots and expressions. Pointer identity prevents repeated discovery of one +shared producer. Inputs with assigned execution timing are rejected. + +Each target retains its original operator, including grouping, filter and input +context, and has an unranked list of existing `Realization` descriptors: + +- Exact execution of the original sub-DAG is always retained as `PassThrough`. +- Mergeable exact intents also offer their exact accumulator kind and parameters. +- Approximate-capable intents offer all declared specialized/universal sketch + algorithms with nominal dimensions from the built-in sizing contracts. +- Exact accuracy requests do not acquire approximate alternatives. Distinct-tuple + counts do not acquire single-value UnivMon alternatives. + +For example, an approximate single-column distinct count offers exact execution, +HLL, Theta, KMV and UnivMon. These are candidate choices, not assessed accuracy +certificates. Catalog order is stable and has no cost/preference meaning. + +The API accepts no empirical cost or accuracy model, runtime capabilities, storage +policy or materialization assignment. It does not rank, select, construct runtime +state or claim physical feasibility. Pass 2 must compose and structurally validate +replacement sub-DAGs and retain independent/shared alternatives before physical +planning and complete workload selection. The descriptors are not executable +plans, and callers must not execute the first choice as a selection policy. + +Multi-measure aggregates remain intact in the roots until an explicit semantic +split is supported. Opaque deployment extensions retain exact execution here; +additional local alternatives require an explicit logical rule rather than a cost +model making a generation decision. The legacy ranked search remains available +for the existing pipeline until its later cutover; this module supplies the new +logical-only entry point without changing production selection prematurely. From 17e6ca75ca3d251e060ed36f5780f1c99cd29201 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Sat, 3 Oct 2026 19:47:09 +0000 Subject: [PATCH 2/9] docs(planner): say local candidates have no execution timing assigned Co-Authored-By: Claude Opus 5.5 --- crates/asap-aware-mapping/src/lib.rs | 3 ++- crates/asap-aware-mapping/tests/logical_candidates.rs | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/crates/asap-aware-mapping/src/lib.rs b/crates/asap-aware-mapping/src/lib.rs index c9becbc25..d81d538ec 100644 --- a/crates/asap-aware-mapping/src/lib.rs +++ b/crates/asap-aware-mapping/src/lib.rs @@ -212,5 +212,6 @@ pub use topk_reuse::TopKLimitReuseStrategy; pub mod maintained_population; -/// Phase-free local candidate generation over the unified IR. +/// Local candidate generation over the unified IR. No execution timing is +/// assigned: that is a Stage 2 materialization decision. pub mod logical_candidates; diff --git a/crates/asap-aware-mapping/tests/logical_candidates.rs b/crates/asap-aware-mapping/tests/logical_candidates.rs index a08c0739d..d25a4cfdb 100644 --- a/crates/asap-aware-mapping/tests/logical_candidates.rs +++ b/crates/asap-aware-mapping/tests/logical_candidates.rs @@ -150,7 +150,7 @@ fn scalar_root_producers_are_discovered_once() { /// Example 1 rate lowering reaches the exact accumulator choice without a cost model. #[test] -fn promql_lowering_reaches_phase_free_local_candidates() { +fn promql_lowering_reaches_local_candidates_without_execution_timing() { use asap_types::workload::{ AccuracyRequirement, BatchEntry, PlanningWorkload, Query, QueryLanguage, QueryRequirements, QueryWorkload, From 19905ebad7d4e10ad8f042e1baac9f70bcc064dd Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Sat, 3 Oct 2026 21:40:48 +0000 Subject: [PATCH 3/9] test(planner): call the promoted PromQL workload lowering The PromQL frontend's unified module was promoted to the crate root later in the stack; asap_frontend_promql::unified no longer exists. Co-Authored-By: Claude Opus 5.5 --- crates/asap-aware-mapping/tests/logical_candidates.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/asap-aware-mapping/tests/logical_candidates.rs b/crates/asap-aware-mapping/tests/logical_candidates.rs index d25a4cfdb..20211ce69 100644 --- a/crates/asap-aware-mapping/tests/logical_candidates.rs +++ b/crates/asap-aware-mapping/tests/logical_candidates.rs @@ -179,7 +179,7 @@ fn promql_lowering_reaches_local_candidates_without_execution_timing() { ..Default::default() }), }; - let roots = asap_frontend_promql::unified::lower_promql_query_workload(&workload, 0).unwrap(); + let roots = asap_frontend_promql::lower_promql_query_workload(&workload, 0).unwrap(); let candidates = enumerate_local_logical_candidates(roots.into_iter().enumerate().collect()).unwrap(); assert!(candidates From 1d1f9f05ba6acf7beeaabf18dbee13bd997a121c Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Thu, 8 Oct 2026 13:47:03 +0000 Subject: [PATCH 4/9] test(planner): validate local candidates with validate_structure after the export removal Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01W7qG9aFyPij5uWsyAJCxDW --- crates/asap-aware-mapping/tests/logical_candidates.rs | 10 ++-------- 1 file changed, 2 insertions(+), 8 deletions(-) diff --git a/crates/asap-aware-mapping/tests/logical_candidates.rs b/crates/asap-aware-mapping/tests/logical_candidates.rs index 20211ce69..27e3cc659 100644 --- a/crates/asap-aware-mapping/tests/logical_candidates.rs +++ b/crates/asap-aware-mapping/tests/logical_candidates.rs @@ -133,19 +133,13 @@ fn scalar_root_producers_are_discovered_once() { let candidates = enumerate_local_logical_candidates(roots).unwrap(); assert_eq!(candidates.roots.len(), 2); for (_, root) in &candidates.roots { - asap_types::ir::export::compile_logical_asap_query(root) - .unwrap() - .validate() - .unwrap(); + 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()); - asap_types::ir::export::compile_logical_asap_dag(&producer) - .unwrap() - .validate() - .unwrap(); + producer.validate_structure().unwrap(); } /// Example 1 rate lowering reaches the exact accumulator choice without a cost model. From 230dd87b4ce9131527e4b0c6e8aece306aafdc5b Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Mon, 12 Oct 2026 20:06:48 +0000 Subject: [PATCH 5/9] fix(planner): Pass 1 local candidates per #509 Summary candidates are unsized (LocalCandidate::Summary carries only the algorithm); sizing happens after Pass 2 sharing. Every pruned candidate is recorded on its target with a PruneReason (RequiresExact, NullableAvgInput). Each measure of an aggregate with a realization besides exact execution is its own target; input containing ASAP operators is rejected. avg is rewritten as exact sum/count accumulator states. The Example 1 test runs Q1 as exact, and the dev doc places composition in Pass 1. - One exact option per target: the exact accumulator (ExactAccumulator, or AvgAsSumCount for avg) when it can be built, otherwise plain Exact (NullableAvgInput prune, or no accumulator form). The plain form would duplicate "accumulator, not materialized"; Stage 2/3 decide materialization and sharing. - Per-measure filters and HAVING keep every candidate. The target records the measure's filter (applied to the state as SummaryAgg.filter) and a HavingPlacement for the aggregate's having or a Filter directly above it (the SQL frontend's HAVING form): BelowAggregate when it reads only group keys (pushed below as a row predicate), else AboveOutput (a Filter above the finalize/estimate output; approximate candidates carry guarantee = None). Composition (#575) applies them; PruneReason::FilteredAggregate is removed. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01W7qG9aFyPij5uWsyAJCxDW --- .../src/logical_candidates.rs | 281 ++++++++++--- .../tests/logical_candidates.rs | 368 +++++++++++++++--- docs/develop_docs/local-logical-candidates.md | 98 ++++- 3 files changed, 627 insertions(+), 120 deletions(-) diff --git a/crates/asap-aware-mapping/src/logical_candidates.rs b/crates/asap-aware-mapping/src/logical_candidates.rs index 8c8a28913..c2aa1ab8b 100644 --- a/crates/asap-aware-mapping/src/logical_candidates.rs +++ b/crates/asap-aware-mapping/src/logical_candidates.rs @@ -1,28 +1,112 @@ //! Pass 1 local alternatives over the unified logical IR. //! //! Alternatives are nominal realization descriptors attached to their original -//! target, not ranked plans or accuracy certificates. Workload composition and -//! physical planning consume this inventory later; empirical models belong to -//! selection. The legacy search API remains until planner cutover. -use std::collections::HashSet; +//! target, not ranked plans or accuracy certificates. Summary alternatives are +//! unsized: Pass 2 may share a summary and tighten it to the strictest consumer, +//! so sizing happens later. Workload composition and physical planning consume +//! this inventory later; empirical models belong to selection. The legacy search +//! API remains until planner cutover. +use std::collections::{HashMap, HashSet}; use std::rc::Rc; -use asap_types::ir::{NonASAPOp, OperatorNode, QueryRoot, SchemaDerivationError}; -use asap_types::post_asap::{ExactKind, ExactParams, SketchKind}; -use asap_types::pre_asap::AggIntent; +use asap_types::ir::operator_properties::Reduction; +use asap_types::ir::{NonASAPOp, OperatorNode, Predicate, QueryRoot, SchemaDerivationError}; +use asap_types::post_asap::{ExactKind, ExactParams, 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, default_size_params, summary_candidates, Realization, -}; +use crate::replacement::{accuracy_budget, accuracy_target, summary_candidates}; -/// All local realizations of one single-measure aggregate. The target retains -/// source, grouping, filters, input expressions and evaluation context. +/// One local realization of a target measure. +/// +/// 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 LocalCandidate { + /// 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 LocalChoices { + pub alternatives: Vec, + pub pruned: Vec<(LocalCandidate, PruneReason)>, +} + +impl LocalChoices { + /// Prune matching candidates; a pruned exact accumulator form falls back to + /// plain `Exact`, so the target keeps one exact option. + fn prune(&mut self, reason: PruneReason, applies: impl Fn(&LocalCandidate) -> bool) { + let (pruned, kept): (Vec<_>, _) = std::mem::take(&mut self.alternatives) + .into_iter() + .partition(|choice| *choice != LocalCandidate::Exact && applies(choice)); + self.alternatives = kept; + if pruned.iter().any(|choice| { + matches!( + choice, + LocalCandidate::ExactAccumulator { .. } | LocalCandidate::AvgAsSumCount + ) + }) { + self.alternatives.insert(0, LocalCandidate::Exact); + } + self.pruned.extend( + pruned + .into_iter() + .map(|choice: LocalCandidate| (choice, reason)), + ); + } +} + +/// Where a target's `HAVING` is applied. It covers the aggregate's own +/// `having` and a `Filter` directly above the aggregate (the form the SQL +/// frontend lowers `HAVING` to). Composition (#575) applies it. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +pub enum HavingPlacement { + /// The predicate reads only group keys, 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 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 LocalLogicalTarget { pub target: Rc, - pub alternatives: Vec, + /// Index into the aggregate's measures. + pub measure: usize, + pub choices: LocalChoices, + /// The measure's own filter; only rows that pass it update the state + /// (`SummaryAgg.filter`). + pub filter: Option, + pub having: Option, } /// Compact Pass 1 inventory; roots and nested producer dependencies are retained. @@ -38,17 +122,19 @@ pub enum LogicalCandidateError { Structure(#[from] SchemaDerivationError), #[error("logical candidate input already has assigned execution timing")] AssignedTiming, + #[error("logical candidate input already contains ASAP operators")] + ContainsAsap, #[error("approximate accuracy requires finite positive epsilon and delta in (0, 1)")] InvalidAccuracy, } -/// Enumerate exact and summary choices in stable catalog order, without ranking -/// or empirical assessment. Parameters are candidate dimensions, not a claim -/// that a deployment meets the request's accuracy requirement. -pub fn local_realizations_for_intent( +/// 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 local_candidates_for_intent( intent: &AggIntent, -) -> Result, LogicalCandidateError> { - let mut choices = vec![Realization::PassThrough]; +) -> Result { let exact = match intent { AggIntent::Count { .. } => Some((ExactKind::Count, ExactParams::Count)), AggIntent::Sum { .. } => Some((ExactKind::Sum, ExactParams::Sum)), @@ -59,9 +145,15 @@ pub fn local_realizations_for_intent( AggIntent::Increase => Some((ExactKind::Increase, ExactParams::Increase)), _ => None, }; - if let Some((kind, params)) = exact { - choices.push(Realization::ExactAggregate { kind, params }); - } + let exact = match exact { + Some((kind, params)) => LocalCandidate::ExactAccumulator { kind, params }, + None if matches!(intent, AggIntent::Avg { .. }) => LocalCandidate::AvgAsSumCount, + None => LocalCandidate::Exact, + }; + let mut choices = LocalChoices { + alternatives: vec![exact], + pruned: vec![], + }; if let Some(target) = accuracy_target(intent) { if *target != AccuracyTarget::Exact { let (epsilon, delta) = accuracy_budget(target); @@ -73,24 +165,71 @@ pub fn local_realizations_for_intent( { return Err(LogicalCandidateError::InvalidAccuracy); } - for algorithm in summary_candidates(intent) { - choices.push(Realization::Sketch(SketchKind::new( - algorithm.clone(), - default_size_params(algorithm.clone(), intent, epsilon, delta), - ))); - } + } + let summaries = summary_candidates(intent).iter().cloned(); + if *target == AccuracyTarget::Exact { + choices.pruned.extend(summaries.map(|algorithm| { + ( + LocalCandidate::Summary(algorithm), + PruneReason::RequiresExact, + ) + })); + } else { + choices + .alternatives + .extend(summaries.map(LocalCandidate::Summary)); } } Ok(choices) } -/// Discover single-measure targets, including operator plans read by scalar roots. -/// Multi-measure aggregates remain intact pending a semantics-preserving split. +/// 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) +} + +/// Where `having` over `aggregate`'s output is applied: below the aggregate +/// when it reads only `by(...)` group keys, otherwise above the output. +fn having_placement(aggregate: &OperatorNode, having: &Predicate) -> HavingPlacement { + let keys = match aggregate.non_asap() { + Some(NonASAPOp::Aggregate { + reduction: Reduction::Reduce(by), + .. + }) if !by.is_without() => by.len(), + _ => 0, + }; + if having.0.operator_refs().is_empty() + && having.0.columns_referenced().iter().all(|id| *id < keys) + { + 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. pub fn enumerate_local_logical_candidates( roots: Vec<(Id, QueryRoot)>, ) -> Result, LogicalCandidateError> { let mut seen = HashSet::new(); - let mut targets = Vec::new(); + let mut aggregates = Vec::new(); + // `HAVING` lowered as a `Filter` directly above an aggregate; a shared + // aggregate keeps the placement that is safe for every such filter. + let mut havings: HashMap<*const OperatorNode, HavingPlacement> = HashMap::new(); for (_, root) in &roots { root.validate_structure()?; let operators = match root { @@ -98,6 +237,9 @@ pub fn enumerate_local_logical_candidates( QueryRoot::Scalar(expr) => expr.operator_refs(), }; for root in operators { + if root.contains_asap() { + return Err(LogicalCandidateError::ContainsAsap); + } for node in OperatorNode::reachable(root) { if !seen.insert(Rc::as_ptr(&node)) { continue; @@ -105,43 +247,82 @@ pub fn enumerate_local_logical_candidates( if node.timing.is_some() { return Err(LogicalCandidateError::AssignedTiming); } - if let Some(NonASAPOp::Aggregate { measures, .. }) = node.non_asap() { - if let [intent] = measures.as_slice() { - targets.push(LocalLogicalTarget { - alternatives: local_realizations_for_intent(intent)?, - target: node, - }); + match node.non_asap() { + Some(NonASAPOp::Aggregate { .. }) => aggregates.push(node), + Some(NonASAPOp::Filter { pred, child }) + if matches!(child.non_asap(), Some(NonASAPOp::Aggregate { .. })) => + { + let placement = having_placement(child, pred); + havings + .entry(Rc::as_ptr(child)) + .and_modify(|having| *having = (*having).max(placement)) + .or_insert(placement); } + _ => {} } } } } + let mut targets = Vec::new(); + for node in aggregates { + let Some(NonASAPOp::Aggregate { + measures, + filters, + having, + .. + }) = node.non_asap() + else { + continue; + }; + let having = having + .as_ref() + .map(|having| having_placement(&node, having)) + .max(havings.get(&Rc::as_ptr(&node)).copied()); + for (measure, intent) in measures.iter().enumerate() { + let mut choices = local_candidates_for_intent(intent)?; + if choices.alternatives == [LocalCandidate::Exact] && choices.pruned.is_empty() { + continue; + } + if nullable_avg_input(&node, intent) { + choices.prune(PruneReason::NullableAvgInput, |choice| { + *choice == LocalCandidate::AvgAsSumCount + }); + } + targets.push(LocalLogicalTarget { + target: node.clone(), + measure, + choices, + filter: filters.get(measure).cloned().flatten(), + having, + }); + } + } Ok(LocalLogicalCandidates { roots, targets }) } #[cfg(test)] mod tests { use super::*; - /// Approximate requests must retain the exact execution alternative too. + /// Approximate count keeps one exact option, its accumulator, beside the + /// universal summary; plain exact execution is not offered. #[test] - fn approximate_count_keeps_exact_and_universal_choices() { - let choices = local_realizations_for_intent(&AggIntent::Count { + fn approximate_count_keeps_one_exact_accumulator_and_universal_choices() { + let choices = local_candidates_for_intent(&AggIntent::Count { accuracy: AccuracyTarget::EpsilonDelta { epsilon: 0.05, delta: 0.01, }, }) - .unwrap(); - assert!(choices - .iter() - .any(|choice| matches!(choice, Realization::PassThrough))); - assert!(choices.iter().any(|choice| matches!( - choice, - Realization::ExactAggregate { + .unwrap() + .alternatives; + assert!(!choices.contains(&LocalCandidate::Exact)); + assert_eq!( + choices[0], + LocalCandidate::ExactAccumulator { kind: ExactKind::Count, - .. + params: ExactParams::Count } - ))); - assert!(choices.iter().any(|choice| matches!(choice, Realization::Sketch(kind) if *kind.algorithm() == asap_types::post_asap::SketchAlgorithm::UnivMon))); + ); + assert!(choices.contains(&LocalCandidate::Summary(SketchAlgorithm::UnivMon))); } } diff --git a/crates/asap-aware-mapping/tests/logical_candidates.rs b/crates/asap-aware-mapping/tests/logical_candidates.rs index 27e3cc659..f150036ea 100644 --- a/crates/asap-aware-mapping/tests/logical_candidates.rs +++ b/crates/asap-aware-mapping/tests/logical_candidates.rs @@ -1,15 +1,20 @@ //! Frontend-to-Pass-1 acceptance: candidate discovery precedes empirical selection. -use asap_aware_mapping::{ - logical_candidates::{ - enumerate_local_logical_candidates, local_realizations_for_intent, LogicalCandidateError, - }, - Realization, +use asap_aware_mapping::logical_candidates::{ + enumerate_local_logical_candidates, local_candidates_for_intent, HavingPlacement, + LocalCandidate, LogicalCandidateError, PruneReason, }; use asap_types::{ ir::operator_properties::{Reduction, Source}, - ir::{NonASAPOp, Operator, OperatorNode, QueryRoot, ScalarExpr}, - post_asap::{ExactKind, SketchAlgorithm}, - pre_asap::{AggIntent, DataType, Field, Schema}, + 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; @@ -20,44 +25,65 @@ fn approximate() -> AccuracyTarget { delta: 0.01, } } -fn algorithms(choices: &[Realization]) -> Vec { +fn alternatives(intent: AggIntent) -> Vec { + local_candidates_for_intent(&intent).unwrap().alternatives +} +fn algorithms(choices: &[LocalCandidate]) -> Vec { choices .iter() .filter_map(|choice| match choice { - Realization::Sketch(kind) => Some(kind.algorithm().clone()), + LocalCandidate::Summary(algorithm) => Some(algorithm.clone()), _ => None, }) .collect() } -fn aggregate(intent: AggIntent) -> Rc { - let child = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Scan { +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)], None), + schema: Schema::lifted( + vec![ + Field::plain("src_ip", DataType::Utf8, false), + Field::plain("value", DataType::Float64, nullable_value), + ], + None, + ), })) - .unwrap(); + .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: vec![intent], + measures, output_names: vec![], - filters: vec![], - having: None, + 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_alternatives() { - let choices = local_realizations_for_intent(&AggIntent::Cardinality { + let choices = alternatives(AggIntent::Cardinality { cols: vec![0], accuracy: approximate(), - }) - .unwrap(); - assert!(matches!(choices[0], Realization::PassThrough)); + }); + assert_eq!(choices[0], LocalCandidate::Exact); assert_eq!( algorithms(&choices), vec![ @@ -67,11 +93,10 @@ fn cardinality_keeps_exact_specialized_and_universal_alternatives() { SketchAlgorithm::UnivMon ] ); - let tuple = local_realizations_for_intent(&AggIntent::Cardinality { + let tuple = alternatives(AggIntent::Cardinality { cols: vec![0, 1], accuracy: approximate(), - }) - .unwrap(); + }); assert!(!algorithms(&tuple).contains(&SketchAlgorithm::UnivMon)); } @@ -88,32 +113,45 @@ fn frequency_statistics_keep_universal_choices() { accuracy: approximate(), }, ] { - let choices = local_realizations_for_intent(&intent).unwrap(); - assert!(matches!(choices[0], Realization::PassThrough)); + let choices = alternatives(intent); + assert_eq!(choices[0], LocalCandidate::Exact); assert_eq!(algorithms(&choices), vec![SketchAlgorithm::UnivMon]); } } -/// An exact request cannot acquire an approximate sketch merely because one is available. +/// 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 = local_realizations_for_intent(&AggIntent::Quantile { + let choices = alternatives(AggIntent::Quantile { col: Some(0), q: 0.99, accuracy: approximate(), - }) - .unwrap(); + }); assert_eq!( algorithms(&choices), vec![SketchAlgorithm::Kll, SketchAlgorithm::DDSketch] ); - let exact = local_realizations_for_intent(&AggIntent::Quantile { + let exact = local_candidates_for_intent(&AggIntent::Quantile { col: Some(0), q: 0.99, accuracy: AccuracyTarget::Exact, }) .unwrap(); - assert_eq!(exact, vec![Realization::PassThrough]); + assert_eq!(exact.alternatives, vec![LocalCandidate::Exact]); + assert_eq!( + exact.pruned, + vec![ + ( + LocalCandidate::Summary(SketchAlgorithm::Kll), + PruneReason::RequiresExact + ), + ( + LocalCandidate::Summary(SketchAlgorithm::DDSketch), + PruneReason::RequiresExact + ), + ] + ); } /// Scalar roots expose their producer targets; repeated references retain one target identity. @@ -142,9 +180,10 @@ fn scalar_root_producers_are_discovered_once() { producer.validate_structure().unwrap(); } -/// Example 1 rate lowering reaches the exact accumulator choice without a cost model. +/// #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. #[test] -fn promql_lowering_reaches_local_candidates_without_execution_timing() { +fn example_1_exact_q1_keeps_exact_accumulators_only() { use asap_types::workload::{ AccuracyRequirement, BatchEntry, PlanningWorkload, Query, QueryLanguage, QueryRequirements, QueryWorkload, @@ -155,7 +194,7 @@ fn promql_lowering_reaches_local_candidates_without_execution_timing() { query_batch: Some(vec![BatchEntry { query: Query("sum by (job) (rate(http_requests_total[1m]))".into()), requirements: QueryRequirements { - accuracy: AccuracyRequirement::Explicit(approximate()), + accuracy: AccuracyRequirement::ImplicitExact, ..Default::default() }, predictability: Default::default(), @@ -176,25 +215,34 @@ fn promql_lowering_reaches_local_candidates_without_execution_timing() { let roots = asap_frontend_promql::lower_promql_query_workload(&workload, 0).unwrap(); let candidates = enumerate_local_logical_candidates(roots.into_iter().enumerate().collect()).unwrap(); - assert!(candidates + let found: Vec<_> = candidates .targets .iter() - .any(|target| target.alternatives.iter().any(|choice| matches!( - choice, - Realization::ExactAggregate { + .map(|target| target.choices.alternatives.clone()) + .collect(); + assert_eq!( + found, + vec![ + vec![LocalCandidate::ExactAccumulator { + kind: ExactKind::Sum, + params: ExactParams::Sum + }], + vec![LocalCandidate::ExactAccumulator { kind: ExactKind::Rate, - .. - } - )))); + params: ExactParams::Rate + }], + ] + ); assert!(candidates .targets .iter() .all(|target| target.target.timing.is_none())); } -/// Physical annotations and invalid probability requirements fail at the stage boundary. +/// Physical annotations, ASAP operators and invalid probability requirements +/// fail at the stage boundary. #[test] -fn assigned_timing_and_invalid_accuracy_are_rejected() { +fn assigned_timing_asap_input_and_invalid_accuracy_are_rejected() { let mut producer = (*aggregate(AggIntent::Count { accuracy: approximate(), })) @@ -204,6 +252,22 @@ fn assigned_timing_and_invalid_accuracy_are_rejected() { enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(Rc::new(producer)))]), Err(LogicalCandidateError::AssignedTiming) )); + 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!( + enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(summary))]), + Err(LogicalCandidateError::ContainsAsap) + )); for target in [ AccuracyTarget::Epsilon(f64::NAN), AccuracyTarget::EpsilonDelta { @@ -212,7 +276,7 @@ fn assigned_timing_and_invalid_accuracy_are_rejected() { }, ] { assert!(matches!( - local_realizations_for_intent(&AggIntent::Count { accuracy: target }), + local_candidates_for_intent(&AggIntent::Count { accuracy: target }), Err(LogicalCandidateError::InvalidAccuracy) )); } @@ -221,12 +285,11 @@ fn assigned_timing_and_invalid_accuracy_are_rejected() { /// Local TopK keeps both declared heap substrates without choosing an implementation. #[test] fn topk_keeps_both_specialized_heap_choices() { - let choices = local_realizations_for_intent(&AggIntent::TopK { + let choices = alternatives(AggIntent::TopK { k: 10, accuracy: approximate(), - }) - .unwrap(); - assert!(matches!(choices[0], Realization::PassThrough)); + }); + assert_eq!(choices[0], LocalCandidate::Exact); assert_eq!( algorithms(&choices), vec![ @@ -235,3 +298,210 @@ fn topk_keeps_both_specialized_heap_choices() { ] ); } + +/// Summary candidates are unsized, so they do not depend on the requested error. +#[test] +fn summary_candidates_are_unsized() { + let loose = alternatives(AggIntent::Count { + accuracy: approximate(), + }); + let tight = alternatives(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 = enumerate_local_logical_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.alternatives, + vec![LocalCandidate::ExactAccumulator { + kind: ExactKind::Sum, + params: ExactParams::Sum + }] + ); + assert_eq!( + algorithms(&candidates[1].choices.alternatives), + 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!( + enumerate_local_logical_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 = enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(filtered))]) + .unwrap() + .targets; + assert_eq!(targets.len(), 2); + for target in &targets { + assert_eq!( + target.choices.alternatives, + alternatives(count()), + "measure {}", + target.measure + ); + assert!(target.choices.pruned.is_empty()); + assert_eq!(target.having, None); + } + 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 = enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(root))]) + .unwrap() + .targets; + assert_eq!(targets.len(), 1); + assert!(Rc::ptr_eq(&targets[0].target, &node)); + assert_eq!(targets[0].having, Some(HavingPlacement::BelowAggregate)); + assert_eq!( + targets[0].choices.alternatives, + alternatives(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 = enumerate_local_logical_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 { + assert_eq!(target.having, Some(HavingPlacement::AboveOutput)); + assert_eq!( + target.choices.alternatives, + alternatives(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 = enumerate_local_logical_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.alternatives, + vec![LocalCandidate::AvgAsSumCount] + ); + assert_eq!(targets[1].choices.alternatives, vec![LocalCandidate::Exact]); + assert_eq!( + targets[1].choices.pruned, + vec![(LocalCandidate::AvgAsSumCount, PruneReason::NullableAvgInput)] + ); +} diff --git a/docs/develop_docs/local-logical-candidates.md b/docs/develop_docs/local-logical-candidates.md index b27e9b2d2..3d6aa9b32 100644 --- a/docs/develop_docs/local-logical-candidates.md +++ b/docs/develop_docs/local-logical-candidates.md @@ -6,34 +6,90 @@ ASAP optimization in [planner layering](../design_docs/proposals/planner-layerin `enumerate_local_logical_candidates(roots)` returns a `LocalLogicalCandidates` inventory containing the original named roots and one `LocalLogicalTarget` per -reachable single-measure aggregate. Discovery includes operator producers read by +eligible aggregate measure. Discovery includes operator producers read by scalar roots and expressions. Pointer identity prevents repeated discovery of one -shared producer. Inputs with assigned execution timing are rejected. +shared producer. Inputs with assigned execution timing or ASAP operators are +rejected: the input `LogicalDAG` contains no ASAP summary choices. + +## Eligibility + +A target is one measure of an `Aggregate` whose intent has a realization besides +exact execution: an exact accumulator state, a summary family or a rewrite rule. +Each measure of a multi-measure aggregate is its own target, identified by the +aggregate node and the measure index. A measure with no such realization (for +example `stddev`) is not a target and keeps exact execution. Each target retains its original operator, including grouping, filter and input -context, and has an unranked list of existing `Realization` descriptors: +context, and has an unranked list of `LocalCandidate` descriptors: + +- Each target has exactly one exact option, listed first. A mergeable exact + intent (`count`, `sum`, `min`, `max`, `rate`, `irate`, `increase`) offers + its exact accumulator kind and parameters (`ExactAccumulator`); `avg` offers + `AvgAsSumCount` (below). Plain exact execution of the original sub-DAG + (`Exact`) is offered only when the intent has no exact accumulator form or + that form is pruned. The plain form would duplicate "accumulator, not + materialized", which is the raw plan for such targets; Stage 2/3 decide + materialization and sharing. +- Approximate-capable intents offer all declared specialized/universal summary + algorithms. Summary candidates are unsized: Pass 2 may share a summary and + tighten it to the strictest consumer, so it is sized for the accuracy + requirement later. +- `avg` is rewritten as exact `sum` and `count` accumulator states + (`AvgAsSumCount`). `avg` carries no approximate accuracy requirement, so no + summary family applies to its components. -- Exact execution of the original sub-DAG is always retained as `PassThrough`. -- Mergeable exact intents also offer their exact accumulator kind and parameters. -- Approximate-capable intents offer all declared specialized/universal sketch - algorithms with nominal dimensions from the built-in sizing contracts. -- Exact accuracy requests do not acquire approximate alternatives. Distinct-tuple - counts do not acquire single-value UnivMon alternatives. +A measure filter or `HAVING` does not limit a target to its exact option; +the target keeps every candidate and records how the predicate is applied: + +- A measure's own filter is recorded as `LocalLogicalTarget.filter`; only + rows that pass it update the state (`SummaryAgg.filter`). +- `HAVING` is recorded as `LocalLogicalTarget.having`. The SQL frontend lowers + `HAVING` to a `Filter` directly above the aggregate, so that `Filter` counts, + as does the aggregate's own `having`. A predicate that reads only `by(...)` + group keys is `HavingPlacement::BelowAggregate`: it is pushed below the + aggregate as a row predicate, like a measure filter. A predicate that reads + aggregate values is `HavingPlacement::AboveOutput`: it stays a `Filter` + above the finalize/estimate output, and approximate candidates of the + target carry `guarantee = None`, since membership near the threshold may + change. When several such filters read one shared aggregate, + `AboveOutput` wins. For example, an approximate single-column distinct count offers exact execution, -HLL, Theta, KMV and UnivMon. These are candidate choices, not assessed accuracy +HLL, Theta, KMV and UnivMon; an approximate `count` offers its exact `Count` +accumulator, CMS, Count Sketch and UnivMon, but not plain `Exact`. These are candidate choices, not assessed accuracy certificates. Catalog order is stable and has no cost/preference meaning. +## Pruning + +Every pruned candidate is recorded on its target with a `PruneReason`: + +- `RequiresExact`: an exact accuracy request does not acquire approximate + summaries. +- `NullableAvgInput`: `count` counts rows, so `avg` over a possibly-null input + is not rewritten. + +When the exact accumulator form (`ExactAccumulator` or `AvgAsSumCount`) is +pruned, it stays in the pruned list with its reason and plain `Exact` becomes +the target's one exact option. + +Distinct-tuple counts do not acquire single-value UnivMon alternatives; that +family is not declared for them, so it is not a pruned candidate. + +## Scope + The API accepts no empirical cost or accuracy model, runtime capabilities, storage policy or materialization assignment. It does not rank, select, construct runtime -state or claim physical feasibility. Pass 2 must compose and structurally validate -replacement sub-DAGs and retain independent/shared alternatives before physical -planning and complete workload selection. The descriptors are not executable -plans, and callers must not execute the first choice as a selection policy. - -Multi-measure aggregates remain intact in the roots until an explicit semantic -split is supported. Opaque deployment extensions retain exact execution here; -additional local alternatives require an explicit logical rule rather than a cost -model making a generation decision. The legacy ranked search remains available -for the existing pipeline until its later cutover; this module supplies the new -logical-only entry point without changing production selection prematurely. +state or claim physical feasibility. Composing each candidate into a replacement +sub-DAG, including applying measure filters and `HAVING` as recorded on the +target, and validating its structure is still Pass 1 work, done later when +candidates are realized (#575); Pass 2 then retains independent and shared +alternatives before physical planning and complete workload selection. The +descriptors are not +executable plans, and callers must not execute the first choice as a selection +policy. + +Opaque deployment extensions retain exact execution here; additional local +alternatives require an explicit logical rule rather than a cost model making a +generation decision. The legacy ranked search remains available for the existing +pipeline until its later cutover; this module supplies the new logical-only entry +point without changing production selection prematurely. From 69faebdfac1cd375a3bcded1cbf2f8c174f04653 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Mon, 12 Oct 2026 21:10:23 +0000 Subject: [PATCH 6/9] fix(planner): per-consumer HAVING entries, global HAVING above, stricter Pass 1 input checks LocalLogicalTarget.having is now one HavingEntry (consumer Filter, predicate, placement) per HAVING. A consumer Filter's key-only predicate is pushed below only when every consumer of the aggregate applies it, and a global aggregate's HAVING always stays above the output. Only the adjacent form is recorded; value predicates elsewhere get guarantee None via #575, multi-level push-down is #664. Inputs with assigned materialization are rejected too (AssignedPhysicalAnnotation). Approximate accuracy needs epsilon and delta in (0, 1), and InvalidAccuracy names the root and measure. The avg nullability check runs only for avg. Tests cover SQL HAVING placement, a shared aggregate, a global constant HAVING and #509 Example 1 Q2. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01W7qG9aFyPij5uWsyAJCxDW --- .../src/logical_candidates.rs | 161 +++++++++----- .../tests/logical_candidates.rs | 207 ++++++++++++++++-- .../tests/logical_candidates_sql.rs | 55 +++++ 3 files changed, 353 insertions(+), 70 deletions(-) create mode 100644 crates/integration-tests/tests/logical_candidates_sql.rs diff --git a/crates/asap-aware-mapping/src/logical_candidates.rs b/crates/asap-aware-mapping/src/logical_candidates.rs index c2aa1ab8b..3d2dc26ce 100644 --- a/crates/asap-aware-mapping/src/logical_candidates.rs +++ b/crates/asap-aware-mapping/src/logical_candidates.rs @@ -11,7 +11,7 @@ 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, SketchAlgorithm}; +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; @@ -80,13 +80,18 @@ impl LocalChoices { } } -/// Where a target's `HAVING` is applied. It covers the aggregate's own -/// `having` and a `Filter` directly above the aggregate (the form the SQL -/// frontend lowers `HAVING` to). Composition (#575) applies it. -#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +/// 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, so it is pushed below the - /// aggregate as a row predicate, like a measure filter. + /// 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 @@ -94,6 +99,16 @@ pub enum HavingPlacement { 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. @@ -106,7 +121,9 @@ pub struct LocalLogicalTarget { /// The measure's own filter; only rows that pass it update the state /// (`SummaryAgg.filter`). pub filter: Option, - pub having: 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. @@ -120,12 +137,19 @@ pub struct LocalLogicalCandidates { pub enum LogicalCandidateError { #[error(transparent)] Structure(#[from] SchemaDerivationError), - #[error("logical candidate input already has assigned execution timing")] - AssignedTiming, + #[error("logical candidate input already has assigned execution timing or materialization")] + AssignedPhysicalAnnotation, #[error("logical candidate input already contains ASAP operators")] ContainsAsap, - #[error("approximate accuracy requires finite positive epsilon and delta in (0, 1)")] - InvalidAccuracy, + /// `root` and `measure` locate the target; [`local_candidates_for_intent`] + /// alone leaves them `None`. + #[error( + "approximate accuracy of root {root:?}, measure {measure:?} requires epsilon and delta in (0, 1)" + )] + InvalidAccuracy { + root: Option, + measure: Option, + }, } /// Enumerate exact, rewrite and summary choices of one measure in stable catalog @@ -157,13 +181,12 @@ pub fn local_candidates_for_intent( if let Some(target) = accuracy_target(intent) { if *target != AccuracyTarget::Exact { let (epsilon, delta) = accuracy_budget(target); - if !epsilon.is_finite() - || epsilon <= 0.0 - || !delta.is_finite() - || !(0.0..1.0).contains(&delta) - || delta == 0.0 - { - return Err(LogicalCandidateError::InvalidAccuracy); + let unit = |value: f64| value > 0.0 && value < 1.0; + if !unit(epsilon) || !unit(delta) { + return Err(LogicalCandidateError::InvalidAccuracy { + root: None, + measure: None, + }); } } let summaries = summary_candidates(intent).iter().cloned(); @@ -198,9 +221,10 @@ fn nullable_avg_input(node: &OperatorNode, intent: &AggIntent) -> bool { .is_none_or(|field| field.nullable) } -/// Where `having` over `aggregate`'s output is applied: below the aggregate -/// when it reads only `by(...)` group keys, otherwise above the output. -fn having_placement(aggregate: &OperatorNode, having: &Predicate) -> HavingPlacement { +/// 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), @@ -208,9 +232,13 @@ fn having_placement(aggregate: &OperatorNode, having: &Predicate) -> HavingPlace }) if !by.is_without() => by.len(), _ => 0, }; - if having.0.operator_refs().is_empty() + 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 @@ -222,15 +250,18 @@ fn having_placement(aggregate: &OperatorNode, having: &Predicate) -> HavingPlace /// 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. -pub fn enumerate_local_logical_candidates( +/// A consumer `Filter`'s key-only `HAVING` is pushed below only when every +/// consumer of the aggregate applies the same predicate. +pub fn enumerate_local_logical_candidates( roots: Vec<(Id, QueryRoot)>, ) -> Result, LogicalCandidateError> { let mut seen = HashSet::new(); + // Each aggregate with the first root that reaches it. let mut aggregates = Vec::new(); - // `HAVING` lowered as a `Filter` directly above an aggregate; a shared - // aggregate keeps the placement that is safe for every such filter. - let mut havings: HashMap<*const OperatorNode, HavingPlacement> = HashMap::new(); - for (_, root) in &roots { + // 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], @@ -240,31 +271,29 @@ pub fn enumerate_local_logical_candidates( if root.contains_asap() { return Err(LogicalCandidateError::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() { - return Err(LogicalCandidateError::AssignedTiming); + if node.timing.is_some() || node.materialization != Materialization::NotMaterialized + { + return Err(LogicalCandidateError::AssignedPhysicalAnnotation); } - match node.non_asap() { - Some(NonASAPOp::Aggregate { .. }) => aggregates.push(node), - Some(NonASAPOp::Filter { pred, child }) - if matches!(child.non_asap(), Some(NonASAPOp::Aggregate { .. })) => - { - let placement = having_placement(child, pred); - havings - .entry(Rc::as_ptr(child)) - .and_modify(|having| *having = (*having).max(placement)) - .or_insert(placement); - } - _ => {} + 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 node in aggregates { + for (root, node) in aggregates { let Some(NonASAPOp::Aggregate { measures, filters, @@ -274,16 +303,46 @@ pub fn enumerate_local_logical_candidates( else { continue; }; - let having = having - .as_ref() - .map(|having| having_placement(&node, having)) - .max(havings.get(&Rc::as_ptr(&node)).copied()); + 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)]; + let filters_above: Vec<(&Rc, &Predicate)> = parents + .iter() + .flatten() + .filter_map(|parent| match parent.non_asap() { + Some(NonASAPOp::Filter { pred, .. }) => Some((parent, pred)), + _ => None, + }) + .collect(); + 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 = local_candidates_for_intent(intent)?; + let mut choices = local_candidates_for_intent(intent).map_err(|error| match error { + LogicalCandidateError::InvalidAccuracy { .. } => { + LogicalCandidateError::InvalidAccuracy { + root: Some(root.clone()), + measure: Some(measure), + } + } + error => error, + })?; if choices.alternatives == [LocalCandidate::Exact] && choices.pruned.is_empty() { continue; } - if nullable_avg_input(&node, intent) { + if matches!(intent, AggIntent::Avg { .. }) && nullable_avg_input(&node, intent) { choices.prune(PruneReason::NullableAvgInput, |choice| { *choice == LocalCandidate::AvgAsSumCount }); @@ -293,7 +352,7 @@ pub fn enumerate_local_logical_candidates( measure, choices, filter: filters.get(measure).cloned().flatten(), - having, + having: entries.clone(), }); } } diff --git a/crates/asap-aware-mapping/tests/logical_candidates.rs b/crates/asap-aware-mapping/tests/logical_candidates.rs index f150036ea..15ae1bf3f 100644 --- a/crates/asap-aware-mapping/tests/logical_candidates.rs +++ b/crates/asap-aware-mapping/tests/logical_candidates.rs @@ -1,6 +1,6 @@ //! Frontend-to-Pass-1 acceptance: candidate discovery precedes empirical selection. use asap_aware_mapping::logical_candidates::{ - enumerate_local_logical_candidates, local_candidates_for_intent, HavingPlacement, + enumerate_local_logical_candidates, local_candidates_for_intent, HavingEntry, HavingPlacement, LocalCandidate, LogicalCandidateError, PruneReason, }; use asap_types::{ @@ -182,8 +182,8 @@ fn scalar_root_producers_are_discovered_once() { /// #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. -#[test] -fn example_1_exact_q1_keeps_exact_accumulators_only() { +/// 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, @@ -192,7 +192,7 @@ fn example_1_exact_q1_keeps_exact_accumulators_only() { query_workload: QueryWorkload { language: QueryLanguage::PromQL, query_batch: Some(vec![BatchEntry { - query: Query("sum by (job) (rate(http_requests_total[1m]))".into()), + query: Query(query.into()), requirements: QueryRequirements { accuracy: AccuracyRequirement::ImplicitExact, ..Default::default() @@ -212,7 +212,12 @@ fn example_1_exact_q1_keeps_exact_accumulators_only() { ..Default::default() }), }; - let roots = asap_frontend_promql::lower_promql_query_workload(&workload, 0).unwrap(); + 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 = enumerate_local_logical_candidates(roots.into_iter().enumerate().collect()).unwrap(); let found: Vec<_> = candidates @@ -242,7 +247,7 @@ fn example_1_exact_q1_keeps_exact_accumulators_only() { /// Physical annotations, ASAP operators and invalid probability requirements /// fail at the stage boundary. #[test] -fn assigned_timing_asap_input_and_invalid_accuracy_are_rejected() { +fn assigned_physical_annotation_asap_input_and_invalid_accuracy_are_rejected() { let mut producer = (*aggregate(AggIntent::Count { accuracy: approximate(), })) @@ -250,7 +255,17 @@ fn assigned_timing_asap_input_and_invalid_accuracy_are_rejected() { producer.timing = Some(asap_types::post_asap::ExecutionTiming::QueryTime); assert!(matches!( enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(Rc::new(producer)))]), - Err(LogicalCandidateError::AssignedTiming) + Err(LogicalCandidateError::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!( + enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(Rc::new(stored)))]), + Err(LogicalCandidateError::AssignedPhysicalAnnotation) )); let summary = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryAgg { child: scan(false), @@ -268,18 +283,54 @@ fn assigned_timing_asap_input_and_invalid_accuracy_are_rejected() { enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(summary))]), Err(LogicalCandidateError::ContainsAsap) )); + let epsilon_delta = |epsilon, delta| AccuracyTarget::EpsilonDelta { epsilon, delta }; for target in [ AccuracyTarget::Epsilon(f64::NAN), - AccuracyTarget::EpsilonDelta { - epsilon: 0.1, - delta: 0.0, - }, + 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!( - local_candidates_for_intent(&AggIntent::Count { accuracy: target }), - Err(LogicalCandidateError::InvalidAccuracy) - )); + assert!( + matches!( + local_candidates_for_intent(&AggIntent::Count { + accuracy: target.clone() + }), + Err(LogicalCandidateError::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 = + enumerate_local_logical_candidates(vec![("q", QueryRoot::Operator(invalid))]).unwrap_err(); + assert!( + matches!( + &error, + LogicalCandidateError::InvalidAccuracy { + root: Some(root), + measure: Some(1) + } if root == "\"q\"" + ), + "{error:?}" + ); } /// Local TopK keeps both declared heap substrates without choosing an implementation. @@ -396,7 +447,7 @@ fn filtered_measure_keeps_candidates_and_records_its_filter() { target.measure ); assert!(target.choices.pruned.is_empty()); - assert_eq!(target.having, None); + assert!(target.having.is_empty()); } assert_eq!(targets[0].filter, Some(always())); assert_eq!(targets[1].filter, None); @@ -436,12 +487,19 @@ fn equals(column: usize, value: ScalarValue) -> Predicate { 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 = enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(root))]) + let targets = enumerate_local_logical_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, Some(HavingPlacement::BelowAggregate)); + 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.alternatives, alternatives(AggIntent::Count { @@ -467,7 +525,8 @@ fn value_having_stays_above_output_and_keeps_candidates() { .targets; assert_eq!(targets.len(), 2); for target in &targets { - assert_eq!(target.having, Some(HavingPlacement::AboveOutput)); + let placements: Vec<_> = target.having.iter().map(|entry| entry.placement).collect(); + assert_eq!(placements, vec![HavingPlacement::AboveOutput]); assert_eq!( target.choices.alternatives, alternatives(AggIntent::Count { @@ -505,3 +564,113 @@ fn avg_rewrites_to_sum_and_count_states() { vec![(LocalCandidate::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 = enumerate_local_logical_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 = enumerate_local_logical_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 = enumerate_local_logical_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 = + enumerate_local_logical_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.alternatives == [LocalCandidate::Exact]) + .expect("top-k target"); + assert_eq!( + topk.pruned, + vec![ + ( + LocalCandidate::Summary(SketchAlgorithm::CmsWithHeap), + PruneReason::RequiresExact + ), + ( + LocalCandidate::Summary(SketchAlgorithm::CountSketchWithHeap), + PruneReason::RequiresExact + ), + ] + ); + assert!(found.iter().any(|choices| choices.alternatives + == [LocalCandidate::ExactAccumulator { + kind: ExactKind::Sum, + params: ExactParams::Sum + }] + && choices.pruned.is_empty())); + assert_eq!(found.len(), 2, "{found:#?}"); +} diff --git a/crates/integration-tests/tests/logical_candidates_sql.rs b/crates/integration-tests/tests/logical_candidates_sql.rs new file mode 100644 index 000000000..0cec77237 --- /dev/null +++ b/crates/integration-tests/tests/logical_candidates_sql.rs @@ -0,0 +1,55 @@ +//! SQL `HAVING` → Pass 1 local candidates: where each `HAVING` is applied. + +use asap_aware_mapping::logical_candidates::{ + enumerate_local_logical_candidates, HavingPlacement, LocalLogicalTarget, +}; +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}")); + enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(root))]) + .unwrap() + .targets +} + +fn placements(targets: &[LocalLogicalTarget]) -> 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]); +} From b8e490a8548a1b1f0079a18265d9678f8ec9802e Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Mon, 12 Oct 2026 21:10:23 +0000 Subject: [PATCH 7/9] docs: Pass 1 HAVING placement and filters keep candidates The dev doc describes per-consumer HAVING entries, the adjacent-form scope and that TopK Hydra is added in #605. planner-layering says a filter or HAVING does not prevent accumulator or summary candidates, and Example 1 says Sum has no summary family rather than pruned ones. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01W7qG9aFyPij5uWsyAJCxDW --- .../design_docs/proposals/planner-layering.md | 13 +++++-- docs/develop_docs/local-logical-candidates.md | 37 +++++++++++++------ 2 files changed, 34 insertions(+), 16 deletions(-) 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 diff --git a/docs/develop_docs/local-logical-candidates.md b/docs/develop_docs/local-logical-candidates.md index 3d6aa9b32..e00078b13 100644 --- a/docs/develop_docs/local-logical-candidates.md +++ b/docs/develop_docs/local-logical-candidates.md @@ -8,8 +8,11 @@ ASAP optimization in [planner layering](../design_docs/proposals/planner-layerin inventory containing the original named roots and one `LocalLogicalTarget` per eligible aggregate measure. Discovery includes operator producers read by scalar roots and expressions. Pointer identity prevents repeated discovery of one -shared producer. Inputs with assigned execution timing or ASAP operators are -rejected: the input `LogicalDAG` contains no ASAP summary choices. +shared producer. Inputs with assigned execution timing or materialization, or +with ASAP operators, are rejected: the input `LogicalDAG` contains no ASAP +summary choices or physical annotations. An approximate accuracy requirement +needs epsilon and delta in (0, 1); otherwise enumeration fails with +`InvalidAccuracy`, naming the root and measure. ## Eligibility @@ -43,16 +46,22 @@ the target keeps every candidate and records how the predicate is applied: - A measure's own filter is recorded as `LocalLogicalTarget.filter`; only rows that pass it update the state (`SummaryAgg.filter`). -- `HAVING` is recorded as `LocalLogicalTarget.having`. The SQL frontend lowers - `HAVING` to a `Filter` directly above the aggregate, so that `Filter` counts, - as does the aggregate's own `having`. A predicate that reads only `by(...)` - group keys is `HavingPlacement::BelowAggregate`: it is pushed below the - aggregate as a row predicate, like a measure filter. A predicate that reads - aggregate values is `HavingPlacement::AboveOutput`: it stays a `Filter` - above the finalize/estimate output, and approximate candidates of the - target carry `guarantee = None`, since membership near the threshold may - change. When several such filters read one shared aggregate, - `AboveOutput` wins. +- `HAVING` is recorded as `LocalLogicalTarget.having`, one `HavingEntry` + (consumer `Filter`, predicate, placement) each. Only the adjacent form is + recorded for push-down: the aggregate's own `having` (an entry with no + `Filter`) and a `Filter` directly above the aggregate, which is how the SQL + frontend lowers `HAVING`. A predicate that reads only `by(...)` group keys + is `HavingPlacement::BelowAggregate`: it is pushed below the aggregate as a + row predicate, like a measure filter. A consumer `Filter`'s predicate is + pushed below only when every consumer of the aggregate applies the same + predicate; otherwise another consumer would lose rows. Any other `HAVING`, + and every `HAVING` of a global aggregate (no group keys, so its one output + row exists even with no input rows), is `HavingPlacement::AboveOutput`: it + stays a `Filter` above the finalize/estimate output, and approximate + candidates of the target carry `guarantee = None`, since membership near + the threshold may change. A value predicate anywhere above an approximate + output, adjacent or not, gets `guarantee = None` through guarantee + propagation (#575); multi-level push-down is tracked in #664. For example, an approximate single-column distinct count offers exact execution, HLL, Theta, KMV and UnivMon; an approximate `count` offers its exact `Count` @@ -75,6 +84,10 @@ the target's one exact option. Distinct-tuple counts do not acquire single-value UnivMon alternatives; that family is not declared for them, so it is not a pruned candidate. +TopK offers its per-group heap summaries (`CmsWithHeap`, +`CountSketchWithHeap`). The grouping-level Hydra candidate ("over all groups") +is not generated in Pass 1 here; it is added in #605. + ## Scope The API accepts no empirical cost or accuracy model, runtime capabilities, storage From d3e39fe48e6dfb03250c2324a61b87844456ce0d Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Tue, 13 Oct 2026 01:55:44 +0000 Subject: [PATCH 8/9] fix(planner): only a Filter over the aggregate is its HAVING A Filter that reads an aggregate only from a scalar expression (PromQL scalar(), subqueries) stays a plain consumer instead of a HavingEntry; filters above are deduplicated by pointer. Print the InvalidAccuracy root once, add tests for Project/scalar consumers, mixed SQL HAVING, without/PerEntity and annotated non-root nodes, and rewrap the doc. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01W7qG9aFyPij5uWsyAJCxDW --- .../src/logical_candidates.rs | 30 ++-- .../tests/logical_candidates.rs | 169 +++++++++++++++++- .../tests/logical_candidates_sql.rs | 8 + docs/develop_docs/local-logical-candidates.md | 39 ++-- 4 files changed, 215 insertions(+), 31 deletions(-) diff --git a/crates/asap-aware-mapping/src/logical_candidates.rs b/crates/asap-aware-mapping/src/logical_candidates.rs index 3d2dc26ce..ca0938a4d 100644 --- a/crates/asap-aware-mapping/src/logical_candidates.rs +++ b/crates/asap-aware-mapping/src/logical_candidates.rs @@ -141,10 +141,12 @@ pub enum LogicalCandidateError { AssignedPhysicalAnnotation, #[error("logical candidate input already contains ASAP operators")] ContainsAsap, - /// `root` and `measure` locate the target; [`local_candidates_for_intent`] - /// alone leaves them `None`. + /// `root` (the root id's `Debug` form) and `measure` locate the target; + /// [`local_candidates_for_intent`] alone leaves them `None`. #[error( - "approximate accuracy of root {root:?}, measure {measure:?} requires epsilon and delta in (0, 1)" + "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, @@ -312,14 +314,20 @@ pub fn enumerate_local_logical_candidates( }) .collect(); let parents = &consumers[&Rc::as_ptr(&node)]; - let filters_above: Vec<(&Rc, &Predicate)> = parents - .iter() - .flatten() - .filter_map(|parent| match parent.non_asap() { - Some(NonASAPOp::Filter { pred, .. }) => Some((parent, pred)), - _ => None, - }) - .collect(); + // 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); diff --git a/crates/asap-aware-mapping/tests/logical_candidates.rs b/crates/asap-aware-mapping/tests/logical_candidates.rs index 15ae1bf3f..bbc9fb0f9 100644 --- a/crates/asap-aware-mapping/tests/logical_candidates.rs +++ b/crates/asap-aware-mapping/tests/logical_candidates.rs @@ -254,7 +254,10 @@ fn assigned_physical_annotation_asap_input_and_invalid_accuracy_are_rejected() { .clone(); producer.timing = Some(asap_types::post_asap::ExecutionTiming::QueryTime); assert!(matches!( - enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(Rc::new(producer)))]), + enumerate_local_logical_candidates(vec![( + 0, + QueryRoot::Operator(Rc::new(producer.clone())) + )]), Err(LogicalCandidateError::AssignedPhysicalAnnotation) )); let mut stored = (*aggregate(AggIntent::Count { @@ -264,9 +267,17 @@ fn assigned_physical_annotation_asap_input_and_invalid_accuracy_are_rejected() { stored.materialization = asap_types::post_asap::Materialization::MatAtQuery(asap_types::post_asap::Medium::Memory); assert!(matches!( - enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(Rc::new(stored)))]), + enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(Rc::new(stored.clone())))]), Err(LogicalCandidateError::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!( + enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(root))]), + Err(LogicalCandidateError::AssignedPhysicalAnnotation) + )); + } let summary = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryAgg { child: scan(false), family: FieldDataType::Sketch( @@ -331,6 +342,10 @@ fn assigned_physical_annotation_asap_input_and_invalid_accuracy_are_rejected() { ), "{error:?}" ); + assert!( + error.to_string().contains("root \"q\", measure 1 "), + "{error}" + ); } /// Local TopK keeps both declared heap substrates without choosing an implementation. @@ -674,3 +689,153 @@ fn example_1_exact_q2_prunes_topk_heaps() { && 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::logical_candidates::LocalLogicalTarget], + 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 = enumerate_local_logical_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 = enumerate_local_logical_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 = enumerate_local_logical_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 = enumerate_local_logical_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 = enumerate_local_logical_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 = enumerate_local_logical_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/logical_candidates_sql.rs b/crates/integration-tests/tests/logical_candidates_sql.rs index 0cec77237..3128d9c27 100644 --- a/crates/integration-tests/tests/logical_candidates_sql.rs +++ b/crates/integration-tests/tests/logical_candidates_sql.rs @@ -53,3 +53,11 @@ 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/develop_docs/local-logical-candidates.md b/docs/develop_docs/local-logical-candidates.md index e00078b13..11517a048 100644 --- a/docs/develop_docs/local-logical-candidates.md +++ b/docs/develop_docs/local-logical-candidates.md @@ -49,24 +49,28 @@ the target keeps every candidate and records how the predicate is applied: - `HAVING` is recorded as `LocalLogicalTarget.having`, one `HavingEntry` (consumer `Filter`, predicate, placement) each. Only the adjacent form is recorded for push-down: the aggregate's own `having` (an entry with no - `Filter`) and a `Filter` directly above the aggregate, which is how the SQL - frontend lowers `HAVING`. A predicate that reads only `by(...)` group keys - is `HavingPlacement::BelowAggregate`: it is pushed below the aggregate as a - row predicate, like a measure filter. A consumer `Filter`'s predicate is - pushed below only when every consumer of the aggregate applies the same - predicate; otherwise another consumer would lose rows. Any other `HAVING`, - and every `HAVING` of a global aggregate (no group keys, so its one output - row exists even with no input rows), is `HavingPlacement::AboveOutput`: it - stays a `Filter` above the finalize/estimate output, and approximate - candidates of the target carry `guarantee = None`, since membership near - the threshold may change. A value predicate anywhere above an approximate - output, adjacent or not, gets `guarantee = None` through guarantee - propagation (#575); multi-level push-down is tracked in #664. + `Filter`) and a `Filter` whose input is the aggregate, which is how the SQL + frontend lowers `HAVING`. A `Filter` that reads the aggregate only from a + scalar expression (for example PromQL `scalar(...)`) is not its `HAVING`, + though it still counts as a consumer. A predicate that reads only `by(...)` + group keys is `HavingPlacement::BelowAggregate`: it is pushed below the + aggregate as a row predicate, like a measure filter. A consumer `Filter`'s + predicate is pushed below only when every consumer of the aggregate applies + the same predicate; otherwise another consumer would lose rows. Any other + `HAVING`, and every `HAVING` of a global aggregate (no group keys, so its + one output row exists even with no input rows), is + `HavingPlacement::AboveOutput`: it stays a `Filter` above the + finalize/estimate output, and approximate candidates of the target carry + `guarantee = None`, since membership near the threshold may change. A value + predicate anywhere above an approximate output, adjacent or not, gets + `guarantee = None` through guarantee propagation (#575); multi-level + push-down is tracked in #664. For example, an approximate single-column distinct count offers exact execution, HLL, Theta, KMV and UnivMon; an approximate `count` offers its exact `Count` -accumulator, CMS, Count Sketch and UnivMon, but not plain `Exact`. These are candidate choices, not assessed accuracy -certificates. Catalog order is stable and has no cost/preference meaning. +accumulator, CMS, Count Sketch and UnivMon, but not plain `Exact`. These are +candidate choices, not assessed accuracy certificates. Catalog order is stable +and has no cost/preference meaning. ## Pruning @@ -97,9 +101,8 @@ sub-DAG, including applying measure filters and `HAVING` as recorded on the target, and validating its structure is still Pass 1 work, done later when candidates are realized (#575); Pass 2 then retains independent and shared alternatives before physical planning and complete workload selection. The -descriptors are not -executable plans, and callers must not execute the first choice as a selection -policy. +descriptors are not executable plans, and callers must not execute the first +choice as a selection policy. Opaque deployment extensions retain exact execution here; additional local alternatives require an explicit logical rule rather than a cost model making a From e683b110be610de992a87aa2b894b32cf3e5399a Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Tue, 13 Oct 2026 03:03:24 +0000 Subject: [PATCH 9/9] refactor(planner): name Pass 1 per-computation candidates as in the planning-stages doc Follow the review on #561: use the terms of planner-layering.md (Stage 1, Pass 1: per-computation candidate generation) instead of "local". - Module logical_candidates -> per_computation_candidates; LocalCandidate -> ComputationCandidate, LocalLogicalTarget -> ComputationTarget, LocalChoices -> ComputationChoices (field alternatives -> candidates), LocalLogicalCandidates -> PerComputationCandidates, enumerate_local_logical_candidates -> generate_per_computation_candidates, local_candidates_for_intent -> computation_candidates_for_intent. - The module doc states the stage and pass and carries the rules that were in docs/develop_docs/local-logical-candidates.md, which is removed. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01W7qG9aFyPij5uWsyAJCxDW --- crates/asap-aware-mapping/src/lib.rs | 4 +- ...dates.rs => per_computation_candidates.rs} | 137 ++++++++------- ...dates.rs => per_computation_candidates.rs} | 161 ++++++++++-------- ...l.rs => per_computation_candidates_sql.rs} | 12 +- docs/develop_docs/local-logical-candidates.md | 111 ------------ 5 files changed, 172 insertions(+), 253 deletions(-) rename crates/asap-aware-mapping/src/{logical_candidates.rs => per_computation_candidates.rs} (73%) rename crates/asap-aware-mapping/tests/{logical_candidates.rs => per_computation_candidates.rs} (81%) rename crates/integration-tests/tests/{logical_candidates_sql.rs => per_computation_candidates_sql.rs} (80%) delete mode 100644 docs/develop_docs/local-logical-candidates.md diff --git a/crates/asap-aware-mapping/src/lib.rs b/crates/asap-aware-mapping/src/lib.rs index d81d538ec..41b585513 100644 --- a/crates/asap-aware-mapping/src/lib.rs +++ b/crates/asap-aware-mapping/src/lib.rs @@ -212,6 +212,6 @@ pub use topk_reuse::TopKLimitReuseStrategy; pub mod maintained_population; -/// Local candidate generation over the unified IR. No execution timing is +/// Stage 1, Pass 1: per-computation candidate generation. No execution timing is /// assigned: that is a Stage 2 materialization decision. -pub mod logical_candidates; +pub mod per_computation_candidates; diff --git a/crates/asap-aware-mapping/src/logical_candidates.rs b/crates/asap-aware-mapping/src/per_computation_candidates.rs similarity index 73% rename from crates/asap-aware-mapping/src/logical_candidates.rs rename to crates/asap-aware-mapping/src/per_computation_candidates.rs index ca0938a4d..2afe46036 100644 --- a/crates/asap-aware-mapping/src/logical_candidates.rs +++ b/crates/asap-aware-mapping/src/per_computation_candidates.rs @@ -1,11 +1,29 @@ -//! Pass 1 local alternatives over the unified logical IR. +//! Stage 1 (logical ASAP-aware optimization), Pass 1: per-computation +//! candidate generation, as defined in +//! `docs/design_docs/proposals/planner-layering.md`. //! -//! Alternatives are nominal realization descriptors attached to their original -//! target, not ranked plans or accuracy certificates. Summary alternatives are -//! unsized: Pass 2 may share a summary and tighten it to the strictest consumer, -//! so sizing happens later. Workload composition and physical planning consume -//! this inventory later; empirical models belong to selection. The legacy search -//! API remains until planner cutover. +//! [`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; @@ -18,14 +36,14 @@ use thiserror::Error; use crate::replacement::{accuracy_budget, accuracy_target, summary_candidates}; -/// One local realization of a target measure. +/// 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 LocalCandidate { +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, @@ -51,31 +69,31 @@ pub enum PruneReason { /// Unranked candidates of one measure and the candidates pruned for it. #[derive(Debug, Clone, PartialEq)] -pub struct LocalChoices { - pub alternatives: Vec, - pub pruned: Vec<(LocalCandidate, PruneReason)>, +pub struct ComputationChoices { + pub candidates: Vec, + pub pruned: Vec<(ComputationCandidate, PruneReason)>, } -impl LocalChoices { - /// Prune matching candidates; a pruned exact accumulator form falls back to +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(&LocalCandidate) -> bool) { - let (pruned, kept): (Vec<_>, _) = std::mem::take(&mut self.alternatives) + 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 != LocalCandidate::Exact && applies(choice)); - self.alternatives = kept; + .partition(|choice| *choice != ComputationCandidate::Exact && applies(choice)); + self.candidates = kept; if pruned.iter().any(|choice| { matches!( choice, - LocalCandidate::ExactAccumulator { .. } | LocalCandidate::AvgAsSumCount + ComputationCandidate::ExactAccumulator { .. } | ComputationCandidate::AvgAsSumCount ) }) { - self.alternatives.insert(0, LocalCandidate::Exact); + self.candidates.insert(0, ComputationCandidate::Exact); } self.pruned.extend( pruned .into_iter() - .map(|choice: LocalCandidate| (choice, reason)), + .map(|choice: ComputationCandidate| (choice, reason)), ); } } @@ -113,11 +131,11 @@ pub struct HavingEntry { /// execution. The target retains source, grouping, filters, input expressions /// and evaluation context. #[derive(Debug, Clone)] -pub struct LocalLogicalTarget { +pub struct ComputationTarget { pub target: Rc, /// Index into the aggregate's measures. pub measure: usize, - pub choices: LocalChoices, + pub choices: ComputationChoices, /// The measure's own filter; only rows that pass it update the state /// (`SummaryAgg.filter`). pub filter: Option, @@ -128,13 +146,13 @@ pub struct LocalLogicalTarget { /// Compact Pass 1 inventory; roots and nested producer dependencies are retained. #[derive(Debug, Clone)] -pub struct LocalLogicalCandidates { +pub struct PerComputationCandidates { pub roots: Vec<(Id, QueryRoot)>, - pub targets: Vec, + pub targets: Vec, } #[derive(Debug, Error)] -pub enum LogicalCandidateError { +pub enum PerComputationCandidateError { #[error(transparent)] Structure(#[from] SchemaDerivationError), #[error("logical candidate input already has assigned execution timing or materialization")] @@ -142,7 +160,7 @@ pub enum LogicalCandidateError { #[error("logical candidate input already contains ASAP operators")] ContainsAsap, /// `root` (the root id's `Debug` form) and `measure` locate the target; - /// [`local_candidates_for_intent`] alone leaves them `None`. + /// [`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("?"), @@ -158,9 +176,9 @@ pub enum LogicalCandidateError { /// 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 local_candidates_for_intent( +pub fn computation_candidates_for_intent( intent: &AggIntent, -) -> Result { +) -> Result { let exact = match intent { AggIntent::Count { .. } => Some((ExactKind::Count, ExactParams::Count)), AggIntent::Sum { .. } => Some((ExactKind::Sum, ExactParams::Sum)), @@ -172,12 +190,12 @@ pub fn local_candidates_for_intent( _ => None, }; let exact = match exact { - Some((kind, params)) => LocalCandidate::ExactAccumulator { kind, params }, - None if matches!(intent, AggIntent::Avg { .. }) => LocalCandidate::AvgAsSumCount, - None => LocalCandidate::Exact, + Some((kind, params)) => ComputationCandidate::ExactAccumulator { kind, params }, + None if matches!(intent, AggIntent::Avg { .. }) => ComputationCandidate::AvgAsSumCount, + None => ComputationCandidate::Exact, }; - let mut choices = LocalChoices { - alternatives: vec![exact], + let mut choices = ComputationChoices { + candidates: vec![exact], pruned: vec![], }; if let Some(target) = accuracy_target(intent) { @@ -185,7 +203,7 @@ pub fn local_candidates_for_intent( let (epsilon, delta) = accuracy_budget(target); let unit = |value: f64| value > 0.0 && value < 1.0; if !unit(epsilon) || !unit(delta) { - return Err(LogicalCandidateError::InvalidAccuracy { + return Err(PerComputationCandidateError::InvalidAccuracy { root: None, measure: None, }); @@ -195,14 +213,14 @@ pub fn local_candidates_for_intent( if *target == AccuracyTarget::Exact { choices.pruned.extend(summaries.map(|algorithm| { ( - LocalCandidate::Summary(algorithm), + ComputationCandidate::Summary(algorithm), PruneReason::RequiresExact, ) })); } else { choices - .alternatives - .extend(summaries.map(LocalCandidate::Summary)); + .candidates + .extend(summaries.map(ComputationCandidate::Summary)); } } Ok(choices) @@ -254,9 +272,9 @@ fn placement(below: bool) -> HavingPlacement { /// 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 enumerate_local_logical_candidates( +pub fn generate_per_computation_candidates( roots: Vec<(Id, QueryRoot)>, -) -> Result, LogicalCandidateError> { +) -> Result, PerComputationCandidateError> { let mut seen = HashSet::new(); // Each aggregate with the first root that reaches it. let mut aggregates = Vec::new(); @@ -271,7 +289,7 @@ pub fn enumerate_local_logical_candidates( }; for root in operators { if root.contains_asap() { - return Err(LogicalCandidateError::ContainsAsap); + return Err(PerComputationCandidateError::ContainsAsap); } consumers.entry(Rc::as_ptr(root)).or_default().push(None); for node in OperatorNode::reachable(root) { @@ -280,7 +298,7 @@ pub fn enumerate_local_logical_candidates( } if node.timing.is_some() || node.materialization != Materialization::NotMaterialized { - return Err(LogicalCandidateError::AssignedPhysicalAnnotation); + return Err(PerComputationCandidateError::AssignedPhysicalAnnotation); } for child in node.children() { consumers @@ -338,24 +356,25 @@ pub fn enumerate_local_logical_candidates( }); } for (measure, intent) in measures.iter().enumerate() { - let mut choices = local_candidates_for_intent(intent).map_err(|error| match error { - LogicalCandidateError::InvalidAccuracy { .. } => { - LogicalCandidateError::InvalidAccuracy { - root: Some(root.clone()), - measure: Some(measure), + 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.alternatives == [LocalCandidate::Exact] && choices.pruned.is_empty() { + 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 == LocalCandidate::AvgAsSumCount + *choice == ComputationCandidate::AvgAsSumCount }); } - targets.push(LocalLogicalTarget { + targets.push(ComputationTarget { target: node.clone(), measure, choices, @@ -364,7 +383,7 @@ pub fn enumerate_local_logical_candidates( }); } } - Ok(LocalLogicalCandidates { roots, targets }) + Ok(PerComputationCandidates { roots, targets }) } #[cfg(test)] @@ -374,22 +393,22 @@ mod tests { /// universal summary; plain exact execution is not offered. #[test] fn approximate_count_keeps_one_exact_accumulator_and_universal_choices() { - let choices = local_candidates_for_intent(&AggIntent::Count { + let choices = computation_candidates_for_intent(&AggIntent::Count { accuracy: AccuracyTarget::EpsilonDelta { epsilon: 0.05, delta: 0.01, }, }) .unwrap() - .alternatives; - assert!(!choices.contains(&LocalCandidate::Exact)); + .candidates; + assert!(!choices.contains(&ComputationCandidate::Exact)); assert_eq!( choices[0], - LocalCandidate::ExactAccumulator { + ComputationCandidate::ExactAccumulator { kind: ExactKind::Count, params: ExactParams::Count } ); - assert!(choices.contains(&LocalCandidate::Summary(SketchAlgorithm::UnivMon))); + assert!(choices.contains(&ComputationCandidate::Summary(SketchAlgorithm::UnivMon))); } } diff --git a/crates/asap-aware-mapping/tests/logical_candidates.rs b/crates/asap-aware-mapping/tests/per_computation_candidates.rs similarity index 81% rename from crates/asap-aware-mapping/tests/logical_candidates.rs rename to crates/asap-aware-mapping/tests/per_computation_candidates.rs index bbc9fb0f9..fe3bde190 100644 --- a/crates/asap-aware-mapping/tests/logical_candidates.rs +++ b/crates/asap-aware-mapping/tests/per_computation_candidates.rs @@ -1,7 +1,7 @@ //! Frontend-to-Pass-1 acceptance: candidate discovery precedes empirical selection. -use asap_aware_mapping::logical_candidates::{ - enumerate_local_logical_candidates, local_candidates_for_intent, HavingEntry, HavingPlacement, - LocalCandidate, LogicalCandidateError, PruneReason, +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}, @@ -25,14 +25,16 @@ fn approximate() -> AccuracyTarget { delta: 0.01, } } -fn alternatives(intent: AggIntent) -> Vec { - local_candidates_for_intent(&intent).unwrap().alternatives +fn candidates_for(intent: AggIntent) -> Vec { + computation_candidates_for_intent(&intent) + .unwrap() + .candidates } -fn algorithms(choices: &[LocalCandidate]) -> Vec { +fn algorithms(choices: &[ComputationCandidate]) -> Vec { choices .iter() .filter_map(|choice| match choice { - LocalCandidate::Summary(algorithm) => Some(algorithm.clone()), + ComputationCandidate::Summary(algorithm) => Some(algorithm.clone()), _ => None, }) .collect() @@ -78,12 +80,12 @@ fn always() -> Predicate { /// Example 2 preserves specialized distinct summaries and the universal alternative. #[test] -fn cardinality_keeps_exact_specialized_and_universal_alternatives() { - let choices = alternatives(AggIntent::Cardinality { +fn cardinality_keeps_exact_specialized_and_universal_candidates() { + let choices = candidates_for(AggIntent::Cardinality { cols: vec![0], accuracy: approximate(), }); - assert_eq!(choices[0], LocalCandidate::Exact); + assert_eq!(choices[0], ComputationCandidate::Exact); assert_eq!( algorithms(&choices), vec![ @@ -93,7 +95,7 @@ fn cardinality_keeps_exact_specialized_and_universal_alternatives() { SketchAlgorithm::UnivMon ] ); - let tuple = alternatives(AggIntent::Cardinality { + let tuple = candidates_for(AggIntent::Cardinality { cols: vec![0, 1], accuracy: approximate(), }); @@ -113,8 +115,8 @@ fn frequency_statistics_keep_universal_choices() { accuracy: approximate(), }, ] { - let choices = alternatives(intent); - assert_eq!(choices[0], LocalCandidate::Exact); + let choices = candidates_for(intent); + assert_eq!(choices[0], ComputationCandidate::Exact); assert_eq!(algorithms(&choices), vec![SketchAlgorithm::UnivMon]); } } @@ -123,7 +125,7 @@ fn frequency_statistics_keep_universal_choices() { /// available; each pruned family records why. #[test] fn exact_quantile_stays_exact_and_approximate_keeps_both_families() { - let choices = alternatives(AggIntent::Quantile { + let choices = candidates_for(AggIntent::Quantile { col: Some(0), q: 0.99, accuracy: approximate(), @@ -132,22 +134,22 @@ fn exact_quantile_stays_exact_and_approximate_keeps_both_families() { algorithms(&choices), vec![SketchAlgorithm::Kll, SketchAlgorithm::DDSketch] ); - let exact = local_candidates_for_intent(&AggIntent::Quantile { + let exact = computation_candidates_for_intent(&AggIntent::Quantile { col: Some(0), q: 0.99, accuracy: AccuracyTarget::Exact, }) .unwrap(); - assert_eq!(exact.alternatives, vec![LocalCandidate::Exact]); + assert_eq!(exact.candidates, vec![ComputationCandidate::Exact]); assert_eq!( exact.pruned, vec![ ( - LocalCandidate::Summary(SketchAlgorithm::Kll), + ComputationCandidate::Summary(SketchAlgorithm::Kll), PruneReason::RequiresExact ), ( - LocalCandidate::Summary(SketchAlgorithm::DDSketch), + ComputationCandidate::Summary(SketchAlgorithm::DDSketch), PruneReason::RequiresExact ), ] @@ -168,7 +170,7 @@ fn scalar_root_producers_are_discovered_once() { ), ("relation", QueryRoot::Operator(producer.clone())), ]; - let candidates = enumerate_local_logical_candidates(roots).unwrap(); + let candidates = generate_per_computation_candidates(roots).unwrap(); assert_eq!(candidates.roots.len(), 2); for (_, root) in &candidates.roots { root.validate_structure().unwrap(); @@ -219,20 +221,20 @@ fn implicit_exact_promql(query: &str) -> Vec { fn example_1_exact_q1_keeps_exact_accumulators_only() { let roots = implicit_exact_promql("sum by (job) (rate(http_requests_total[1m]))"); let candidates = - enumerate_local_logical_candidates(roots.into_iter().enumerate().collect()).unwrap(); + generate_per_computation_candidates(roots.into_iter().enumerate().collect()).unwrap(); let found: Vec<_> = candidates .targets .iter() - .map(|target| target.choices.alternatives.clone()) + .map(|target| target.choices.candidates.clone()) .collect(); assert_eq!( found, vec![ - vec![LocalCandidate::ExactAccumulator { + vec![ComputationCandidate::ExactAccumulator { kind: ExactKind::Sum, params: ExactParams::Sum }], - vec![LocalCandidate::ExactAccumulator { + vec![ComputationCandidate::ExactAccumulator { kind: ExactKind::Rate, params: ExactParams::Rate }], @@ -254,11 +256,11 @@ fn assigned_physical_annotation_asap_input_and_invalid_accuracy_are_rejected() { .clone(); producer.timing = Some(asap_types::post_asap::ExecutionTiming::QueryTime); assert!(matches!( - enumerate_local_logical_candidates(vec![( + generate_per_computation_candidates(vec![( 0, QueryRoot::Operator(Rc::new(producer.clone())) )]), - Err(LogicalCandidateError::AssignedPhysicalAnnotation) + Err(PerComputationCandidateError::AssignedPhysicalAnnotation) )); let mut stored = (*aggregate(AggIntent::Count { accuracy: approximate(), @@ -267,15 +269,18 @@ fn assigned_physical_annotation_asap_input_and_invalid_accuracy_are_rejected() { stored.materialization = asap_types::post_asap::Materialization::MatAtQuery(asap_types::post_asap::Medium::Memory); assert!(matches!( - enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(Rc::new(stored.clone())))]), - Err(LogicalCandidateError::AssignedPhysicalAnnotation) + 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!( - enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(root))]), - Err(LogicalCandidateError::AssignedPhysicalAnnotation) + generate_per_computation_candidates(vec![(0, QueryRoot::Operator(root))]), + Err(PerComputationCandidateError::AssignedPhysicalAnnotation) )); } let summary = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryAgg { @@ -291,8 +296,8 @@ fn assigned_physical_annotation_asap_input_and_invalid_accuracy_are_rejected() { })) .unwrap(); assert!(matches!( - enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(summary))]), - Err(LogicalCandidateError::ContainsAsap) + generate_per_computation_candidates(vec![(0, QueryRoot::Operator(summary))]), + Err(PerComputationCandidateError::ContainsAsap) )); let epsilon_delta = |epsilon, delta| AccuracyTarget::EpsilonDelta { epsilon, delta }; for target in [ @@ -306,10 +311,10 @@ fn assigned_physical_annotation_asap_input_and_invalid_accuracy_are_rejected() { ] { assert!( matches!( - local_candidates_for_intent(&AggIntent::Count { + computation_candidates_for_intent(&AggIntent::Count { accuracy: target.clone() }), - Err(LogicalCandidateError::InvalidAccuracy { + Err(PerComputationCandidateError::InvalidAccuracy { root: None, measure: None }) @@ -331,11 +336,11 @@ fn assigned_physical_annotation_asap_input_and_invalid_accuracy_are_rejected() { None, ); let error = - enumerate_local_logical_candidates(vec![("q", QueryRoot::Operator(invalid))]).unwrap_err(); + generate_per_computation_candidates(vec![("q", QueryRoot::Operator(invalid))]).unwrap_err(); assert!( matches!( &error, - LogicalCandidateError::InvalidAccuracy { + PerComputationCandidateError::InvalidAccuracy { root: Some(root), measure: Some(1) } if root == "\"q\"" @@ -348,14 +353,14 @@ fn assigned_physical_annotation_asap_input_and_invalid_accuracy_are_rejected() { ); } -/// Local TopK keeps both declared heap substrates without choosing an implementation. +/// TopK keeps both declared heap substrates without choosing an implementation. #[test] fn topk_keeps_both_specialized_heap_choices() { - let choices = alternatives(AggIntent::TopK { + let choices = candidates_for(AggIntent::TopK { k: 10, accuracy: approximate(), }); - assert_eq!(choices[0], LocalCandidate::Exact); + assert_eq!(choices[0], ComputationCandidate::Exact); assert_eq!( algorithms(&choices), vec![ @@ -368,10 +373,10 @@ fn topk_keeps_both_specialized_heap_choices() { /// Summary candidates are unsized, so they do not depend on the requested error. #[test] fn summary_candidates_are_unsized() { - let loose = alternatives(AggIntent::Count { + let loose = candidates_for(AggIntent::Count { accuracy: approximate(), }); - let tight = alternatives(AggIntent::Count { + let tight = candidates_for(AggIntent::Count { accuracy: AccuracyTarget::EpsilonDelta { epsilon: 0.0001, delta: 0.0001, @@ -395,7 +400,7 @@ fn multi_measure_aggregate_has_one_target_per_measure() { vec![], None, ); - let candidates = enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(node))]) + let candidates = generate_per_computation_candidates(vec![(0, QueryRoot::Operator(node))]) .unwrap() .targets; assert_eq!( @@ -406,14 +411,14 @@ fn multi_measure_aggregate_has_one_target_per_measure() { vec![0, 1] ); assert_eq!( - candidates[0].choices.alternatives, - vec![LocalCandidate::ExactAccumulator { + candidates[0].choices.candidates, + vec![ComputationCandidate::ExactAccumulator { kind: ExactKind::Sum, params: ExactParams::Sum }] ); assert_eq!( - algorithms(&candidates[1].choices.alternatives), + algorithms(&candidates[1].choices.candidates), vec![ SketchAlgorithm::Cms, SketchAlgorithm::CountSketch, @@ -430,7 +435,7 @@ fn aggregate_without_realization_is_not_a_target() { population: true, }); assert!( - enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(node))]) + generate_per_computation_candidates(vec![(0, QueryRoot::Operator(node))]) .unwrap() .targets .is_empty() @@ -450,14 +455,14 @@ fn filtered_measure_keeps_candidates_and_records_its_filter() { vec![Some(always()), None], None, ); - let targets = enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(filtered))]) + 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.alternatives, - alternatives(count()), + target.choices.candidates, + candidates_for(count()), "measure {}", target.measure ); @@ -502,7 +507,7 @@ fn equals(column: usize, value: ScalarValue) -> Predicate { 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 = enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(root.clone()))]) + let targets = generate_per_computation_candidates(vec![(0, QueryRoot::Operator(root.clone()))]) .unwrap() .targets; assert_eq!(targets.len(), 1); @@ -516,8 +521,8 @@ fn key_only_having_is_pushed_below_and_keeps_candidates() { }] ); assert_eq!( - targets[0].choices.alternatives, - alternatives(AggIntent::Count { + targets[0].choices.candidates, + candidates_for(AggIntent::Count { accuracy: approximate() }) ); @@ -529,7 +534,7 @@ fn key_only_having_is_pushed_below_and_keeps_candidates() { #[test] fn value_having_stays_above_output_and_keeps_candidates() { let value = || equals(1, ScalarValue::Int64(10)); - let targets = enumerate_local_logical_candidates(vec![ + let targets = generate_per_computation_candidates(vec![ ( 0, QueryRoot::Operator(having_filter(grouped_count(None), value())), @@ -543,8 +548,8 @@ fn value_having_stays_above_output_and_keeps_candidates() { let placements: Vec<_> = target.having.iter().map(|entry| entry.placement).collect(); assert_eq!(placements, vec![HavingPlacement::AboveOutput]); assert_eq!( - target.choices.alternatives, - alternatives(AggIntent::Count { + target.choices.candidates, + candidates_for(AggIntent::Count { accuracy: approximate() }) ); @@ -557,7 +562,7 @@ fn value_having_stays_above_output_and_keeps_candidates() { #[test] fn avg_rewrites_to_sum_and_count_states() { let avg = || vec![AggIntent::Avg { col: None }]; - let targets = enumerate_local_logical_candidates(vec![ + let targets = generate_per_computation_candidates(vec![ ( 0, QueryRoot::Operator(aggregate_over(scan(false), avg(), vec![], None)), @@ -570,13 +575,19 @@ fn avg_rewrites_to_sum_and_count_states() { .unwrap() .targets; assert_eq!( - targets[0].choices.alternatives, - vec![LocalCandidate::AvgAsSumCount] + targets[0].choices.candidates, + vec![ComputationCandidate::AvgAsSumCount] + ); + assert_eq!( + targets[1].choices.candidates, + vec![ComputationCandidate::Exact] ); - assert_eq!(targets[1].choices.alternatives, vec![LocalCandidate::Exact]); assert_eq!( targets[1].choices.pruned, - vec![(LocalCandidate::AvgAsSumCount, PruneReason::NullableAvgInput)] + vec![( + ComputationCandidate::AvgAsSumCount, + PruneReason::NullableAvgInput + )] ); } @@ -587,7 +598,7 @@ 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 = enumerate_local_logical_candidates(vec![ + let targets = generate_per_computation_candidates(vec![ (0, QueryRoot::Operator(filtered.clone())), (1, QueryRoot::Operator(node.clone())), ]) @@ -606,7 +617,7 @@ fn having_on_one_consumer_of_a_shared_aggregate_stays_above() { // 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 = enumerate_local_logical_candidates(vec![ + let targets = generate_per_computation_candidates(vec![ (0, QueryRoot::Operator(first)), (1, QueryRoot::Operator(second)), ]) @@ -629,7 +640,7 @@ fn global_aggregate_having_stays_above() { }] }; let never = Predicate(ScalarExpr::Literal(ScalarValue::Boolean(false))); - let targets = enumerate_local_logical_candidates(vec![ + let targets = generate_per_computation_candidates(vec![ ( 0, QueryRoot::Operator(having_filter( @@ -658,7 +669,7 @@ fn global_aggregate_having_stays_above() { 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 = - enumerate_local_logical_candidates(roots.into_iter().enumerate().collect()).unwrap(); + generate_per_computation_candidates(roots.into_iter().enumerate().collect()).unwrap(); let found: Vec<_> = candidates .targets .iter() @@ -666,23 +677,23 @@ fn example_1_exact_q2_prunes_topk_heaps() { .collect(); let topk = found .iter() - .find(|choices| choices.alternatives == [LocalCandidate::Exact]) + .find(|choices| choices.candidates == [ComputationCandidate::Exact]) .expect("top-k target"); assert_eq!( topk.pruned, vec![ ( - LocalCandidate::Summary(SketchAlgorithm::CmsWithHeap), + ComputationCandidate::Summary(SketchAlgorithm::CmsWithHeap), PruneReason::RequiresExact ), ( - LocalCandidate::Summary(SketchAlgorithm::CountSketchWithHeap), + ComputationCandidate::Summary(SketchAlgorithm::CountSketchWithHeap), PruneReason::RequiresExact ), ] ); - assert!(found.iter().any(|choices| choices.alternatives - == [LocalCandidate::ExactAccumulator { + assert!(found.iter().any(|choices| choices.candidates + == [ComputationCandidate::ExactAccumulator { kind: ExactKind::Sum, params: ExactParams::Sum }] @@ -692,7 +703,7 @@ fn example_1_exact_q2_prunes_topk_heaps() { /// The `HAVING` entries of the target whose measure satisfies `is`. fn having_of( - targets: &[asap_aware_mapping::logical_candidates::LocalLogicalTarget], + targets: &[asap_aware_mapping::per_computation_candidates::ComputationTarget], is: impl Fn(&AggIntent) -> bool, ) -> Vec { let target = targets @@ -726,7 +737,7 @@ fn scalar_reference_filter_is_not_having() { }; let agg = aggregate_over(scan(false), count(), vec![], None); let root = having_filter(scan(false), above_scalar(1, agg.clone())); - let targets = enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(root))]) + let targets = generate_per_computation_candidates(vec![(0, QueryRoot::Operator(root))]) .unwrap() .targets; assert_eq!(targets.len(), 1); @@ -735,7 +746,7 @@ fn scalar_reference_filter_is_not_having() { let agg = aggregate_over(scan(false), count(), vec![], None); let root = having_filter(agg.clone(), above_scalar(0, agg.clone())); - let targets = enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(root.clone()))]) + let targets = generate_per_computation_candidates(vec![(0, QueryRoot::Operator(root.clone()))]) .unwrap() .targets; assert_eq!(targets.len(), 1); @@ -753,7 +764,7 @@ fn scalar_reference_filter_is_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 = enumerate_local_logical_candidates(roots.into_iter().enumerate().collect()) + let targets = generate_per_computation_candidates(roots.into_iter().enumerate().collect()) .unwrap() .targets; assert!(!targets.is_empty()); @@ -762,7 +773,7 @@ fn promql_scalar_operand_is_not_having() { } let roots = implicit_exact_promql("sum by (job) (x) > scalar(count(x))"); - let targets = enumerate_local_logical_candidates(roots.into_iter().enumerate().collect()) + let targets = generate_per_computation_candidates(roots.into_iter().enumerate().collect()) .unwrap() .targets; assert!(having_of(&targets, |intent| matches!(intent, AggIntent::Count { .. })).is_empty()); @@ -786,7 +797,7 @@ fn having_shared_with_a_project_consumer_stays_above() { child: node, })) .unwrap(); - let targets = enumerate_local_logical_candidates(vec![ + let targets = generate_per_computation_candidates(vec![ (0, QueryRoot::Operator(filtered)), (1, QueryRoot::Operator(project)), ]) @@ -823,7 +834,7 @@ fn without_and_per_entity_having_stays_above() { })) .unwrap(); let root = having_filter(node, equals(0, ScalarValue::Utf8("a".into()))); - let targets = enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(root))]) + let targets = generate_per_computation_candidates(vec![(0, QueryRoot::Operator(root))]) .unwrap() .targets; assert_eq!(targets.len(), 1, "{reduction:?}"); diff --git a/crates/integration-tests/tests/logical_candidates_sql.rs b/crates/integration-tests/tests/per_computation_candidates_sql.rs similarity index 80% rename from crates/integration-tests/tests/logical_candidates_sql.rs rename to crates/integration-tests/tests/per_computation_candidates_sql.rs index 3128d9c27..f0d78f526 100644 --- a/crates/integration-tests/tests/logical_candidates_sql.rs +++ b/crates/integration-tests/tests/per_computation_candidates_sql.rs @@ -1,14 +1,14 @@ -//! SQL `HAVING` → Pass 1 local candidates: where each `HAVING` is applied. +//! SQL `HAVING` → Pass 1 per-computation candidates: where each `HAVING` is applied. -use asap_aware_mapping::logical_candidates::{ - enumerate_local_logical_candidates, HavingPlacement, LocalLogicalTarget, +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 { +async fn targets(sql: &str) -> Vec { let catalog = SqlCatalog::new().with_table( "flows", Schema::new(vec![ @@ -23,12 +23,12 @@ async fn targets(sql: &str) -> Vec { let root = lower_sql(sql, &catalog, accuracy) .await .unwrap_or_else(|e| panic!("lower failed for {sql:?}: {e}")); - enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(root))]) + generate_per_computation_candidates(vec![(0, QueryRoot::Operator(root))]) .unwrap() .targets } -fn placements(targets: &[LocalLogicalTarget]) -> Vec { +fn placements(targets: &[ComputationTarget]) -> Vec { targets .iter() .flat_map(|target| target.having.iter().map(|entry| entry.placement)) diff --git a/docs/develop_docs/local-logical-candidates.md b/docs/develop_docs/local-logical-candidates.md deleted file mode 100644 index 11517a048..000000000 --- a/docs/develop_docs/local-logical-candidates.md +++ /dev/null @@ -1,111 +0,0 @@ -# Local logical alternatives (Pass 1) - -`asap_aware_mapping::logical_candidates` enumerates local realization choices over -unified `OperatorNode` and `QueryRoot` inputs. It is the first part of logical -ASAP optimization in [planner layering](../design_docs/proposals/planner-layering.md). - -`enumerate_local_logical_candidates(roots)` returns a `LocalLogicalCandidates` -inventory containing the original named roots and one `LocalLogicalTarget` per -eligible aggregate measure. Discovery includes operator producers read by -scalar roots and expressions. Pointer identity prevents repeated discovery of one -shared producer. Inputs with assigned execution timing or materialization, or -with ASAP operators, are rejected: the input `LogicalDAG` contains no ASAP -summary choices or physical annotations. An approximate accuracy requirement -needs epsilon and delta in (0, 1); otherwise enumeration fails with -`InvalidAccuracy`, naming the root and measure. - -## Eligibility - -A target is one measure of an `Aggregate` whose intent has a realization besides -exact execution: an exact accumulator state, a summary family or a rewrite rule. -Each measure of a multi-measure aggregate is its own target, identified by the -aggregate node and the measure index. A measure with no such realization (for -example `stddev`) is not a target and keeps exact execution. - -Each target retains its original operator, including grouping, filter and input -context, and has an unranked list of `LocalCandidate` descriptors: - -- Each target has exactly one exact option, listed first. A mergeable exact - intent (`count`, `sum`, `min`, `max`, `rate`, `irate`, `increase`) offers - its exact accumulator kind and parameters (`ExactAccumulator`); `avg` offers - `AvgAsSumCount` (below). Plain exact execution of the original sub-DAG - (`Exact`) is offered only when the intent has no exact accumulator form or - that form is pruned. The plain form would duplicate "accumulator, not - materialized", which is the raw plan for such targets; Stage 2/3 decide - materialization and sharing. -- Approximate-capable intents offer all declared specialized/universal summary - algorithms. Summary candidates are unsized: Pass 2 may share a summary and - tighten it to the strictest consumer, so it is sized for the accuracy - requirement later. -- `avg` is rewritten as exact `sum` and `count` accumulator states - (`AvgAsSumCount`). `avg` carries no approximate accuracy requirement, so no - summary family applies to its components. - -A measure filter or `HAVING` does not limit a target to its exact option; -the target keeps every candidate and records how the predicate is applied: - -- A measure's own filter is recorded as `LocalLogicalTarget.filter`; only - rows that pass it update the state (`SummaryAgg.filter`). -- `HAVING` is recorded as `LocalLogicalTarget.having`, one `HavingEntry` - (consumer `Filter`, predicate, placement) each. Only the adjacent form is - recorded for push-down: the aggregate's own `having` (an entry with no - `Filter`) and a `Filter` whose input is the aggregate, which is how the SQL - frontend lowers `HAVING`. A `Filter` that reads the aggregate only from a - scalar expression (for example PromQL `scalar(...)`) is not its `HAVING`, - though it still counts as a consumer. A predicate that reads only `by(...)` - group keys is `HavingPlacement::BelowAggregate`: it is pushed below the - aggregate as a row predicate, like a measure filter. A consumer `Filter`'s - predicate is pushed below only when every consumer of the aggregate applies - the same predicate; otherwise another consumer would lose rows. Any other - `HAVING`, and every `HAVING` of a global aggregate (no group keys, so its - one output row exists even with no input rows), is - `HavingPlacement::AboveOutput`: it stays a `Filter` above the - finalize/estimate output, and approximate candidates of the target carry - `guarantee = None`, since membership near the threshold may change. A value - predicate anywhere above an approximate output, adjacent or not, gets - `guarantee = None` through guarantee propagation (#575); multi-level - push-down is tracked in #664. - -For example, an approximate single-column distinct count offers exact execution, -HLL, Theta, KMV and UnivMon; an approximate `count` offers its exact `Count` -accumulator, CMS, Count Sketch and UnivMon, but not plain `Exact`. These are -candidate choices, not assessed accuracy certificates. Catalog order is stable -and has no cost/preference meaning. - -## Pruning - -Every pruned candidate is recorded on its target with a `PruneReason`: - -- `RequiresExact`: an exact accuracy request does not acquire approximate - summaries. -- `NullableAvgInput`: `count` counts rows, so `avg` over a possibly-null input - is not rewritten. - -When the exact accumulator form (`ExactAccumulator` or `AvgAsSumCount`) is -pruned, it stays in the pruned list with its reason and plain `Exact` becomes -the target's one exact option. - -Distinct-tuple counts do not acquire single-value UnivMon alternatives; that -family is not declared for them, so it is not a pruned candidate. - -TopK offers its per-group heap summaries (`CmsWithHeap`, -`CountSketchWithHeap`). The grouping-level Hydra candidate ("over all groups") -is not generated in Pass 1 here; it is added in #605. - -## Scope - -The API accepts no empirical cost or accuracy model, runtime capabilities, storage -policy or materialization assignment. It does not rank, select, construct runtime -state or claim physical feasibility. Composing each candidate into a replacement -sub-DAG, including applying measure filters and `HAVING` as recorded on the -target, and validating its structure is still Pass 1 work, done later when -candidates are realized (#575); Pass 2 then retains independent and shared -alternatives before physical planning and complete workload selection. The -descriptors are not executable plans, and callers must not execute the first -choice as a selection policy. - -Opaque deployment extensions retain exact execution here; additional local -alternatives require an explicit logical rule rather than a cost model making a -generation decision. The legacy ranked search remains available for the existing -pipeline until its later cutover; this module supplies the new logical-only entry -point without changing production selection prematurely.