From 5b97aab8badf8160e65d9d327838157c96333cf0 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Thu, 15 Oct 2026 15:06:29 +0000 Subject: [PATCH 1/4] refactor(asap-aware-mapping): equal budget split on AccuracySpec replaces the allocator hook (#669) accuracy/allocation.rs (AccuracyBudgetAllocator, EqualSplitAllocator, CompositionShape, AccuracyAllocation) is deleted. accuracy/budget.rs states the one split the planner makes, an equal share of an AccuracyTarget: eps/n (or (1+eps)^(1/n)-1 for a relative metric) and delta/n. The legacy replacement search calls it directly, and its constructors no longer take an allocator. --- .../src/accuracy/allocation.rs | 139 ---------------- .../asap-aware-mapping/src/accuracy/budget.rs | 93 +++++++++++ crates/asap-aware-mapping/src/accuracy/mod.rs | 9 +- crates/asap-aware-mapping/src/grouping.rs | 11 +- crates/asap-aware-mapping/src/lib.rs | 8 +- crates/asap-aware-mapping/src/replacement.rs | 157 +++++++----------- .../tests/promql_to_post_asap.rs | 13 +- crates/types/src/post_asap/guarantee.rs | 5 +- 8 files changed, 163 insertions(+), 272 deletions(-) delete mode 100644 crates/asap-aware-mapping/src/accuracy/allocation.rs create mode 100644 crates/asap-aware-mapping/src/accuracy/budget.rs diff --git a/crates/asap-aware-mapping/src/accuracy/allocation.rs b/crates/asap-aware-mapping/src/accuracy/allocation.rs deleted file mode 100644 index 30002ed14..000000000 --- a/crates/asap-aware-mapping/src/accuracy/allocation.rs +++ /dev/null @@ -1,139 +0,0 @@ -//! Allocate end-to-end error and failure budgets across approximate layers. -use super::*; - -/// The shape of a composition an allocator splits a budget across. -#[derive(Debug, Clone, PartialEq)] -pub struct CompositionShape { - /// The metric the composed guarantee will carry — decides whether the - /// budget composes additively (`Σ ε_i ≤ ε`) or multiplicatively - /// (`Π(1+ε_i) ≤ 1+ε`). - pub metric: ErrorMetric, - /// How many approximate layers share the budget (≥ 1). - pub approximate_layer_count: usize, -} - -/// One way of splitting an end-to-end target across a composition's -/// approximate layers. `layers[0]` is the outermost layer's local target; -/// the remainder are the inner layers', outermost first. -#[derive(Debug, Clone, PartialEq)] -pub struct AccuracyAllocation { - pub allocator: &'static str, - pub layers: Vec, -} - -impl AccuracyAllocation { - /// The end-to-end budget left for everything below `layers[0]` — what - /// the inner sub-DAG must satisfy as a whole (it re-splits internally). - /// `None` for a single-layer allocation. - pub fn inner_target(&self, shape: &CompositionShape) -> Option { - let inner = &self.layers[1..]; - if inner.is_empty() { - return None; - } - let (eps, delta): (Vec, Vec>) = inner - .iter() - .map(|t| (t.error, t.failure_probability)) - .unzip(); - let epsilon = match shape.metric { - ErrorMetric::Relative => eps.iter().map(|e| 1.0 + e).product::() - 1.0, - _ => eps.iter().sum(), - }; - Some(AccuracyTarget { - error: epsilon, - failure_probability: delta.iter().copied().sum::>(), - ..inner[0].clone() - }) - } -} - -/// Enumerates the finite set of budget splits the search tries for one -/// composition. Exposed as its own hook because equal splitting is rarely -/// cost-optimal; a deployment can return several candidate splits and let -/// cost ranking pick among the legal ones. -pub trait AccuracyBudgetAllocator { - fn allocations( - &self, - target: &AccuracyTarget, - composition: &CompositionShape, - ) -> Vec; -} - -/// The initial deterministic allocator: every approximate layer gets an -/// equal share — `ε_i = ε / n`, `δ_i = δ / n` for an additively composed -/// metric, and `ε_i = (1 + ε)^{1/n} − 1` for a multiplicatively composed -/// one — so the composed bound meets the target exactly with no slack. -/// `AccuracyTarget::exact()` yields no allocation: no approximate layer can -/// meet it. -#[derive(Debug, Default, Clone, Copy)] -pub struct EqualSplitAllocator; - -impl AccuracyBudgetAllocator for EqualSplitAllocator { - fn allocations( - &self, - target: &AccuracyTarget, - composition: &CompositionShape, - ) -> Vec { - let n = composition.approximate_layer_count.max(1); - if target.is_exact() { - return Vec::new(); - } - let (epsilon, delta) = (target.error, target.failure_probability); - if !(epsilon.is_finite() && epsilon > 0.0) { - return Vec::new(); - } - let local_epsilon = match composition.metric { - ErrorMetric::Relative => (1.0 + epsilon).powf(1.0 / n as f64) - 1.0, - _ => epsilon / n as f64, - }; - let layer = AccuracyTarget { - error: local_epsilon, - failure_probability: delta.map(|delta| delta / n as f64), - ..target.clone() - }; - vec![AccuracyAllocation { - allocator: "EqualSplitAllocator", - layers: vec![layer; n], - }] - } -} - -#[cfg(test)] -mod tests { - use super::*; - #[test] - fn equal_split_respects_the_root_epsilon_and_delta() { - let target = AccuracyTarget::epsilon_delta(0.1, 0.02); - let shape = CompositionShape { - metric: ErrorMetric::ABSOLUTE, - approximate_layer_count: 2, - }; - let allocations = EqualSplitAllocator.allocations(&target, &shape); - assert_eq!(allocations.len(), 1); - let layers = &allocations[0].layers; - assert_eq!(layers.len(), 2); - let (eps, deltas): (Vec, Vec) = layers - .iter() - .map(|t| (t.error, t.failure_probability.unwrap())) - .unzip(); - assert!((eps.iter().sum::() - 0.1).abs() < 1e-12); - assert!((deltas.iter().sum::() - 0.02).abs() < 1e-12); - assert_eq!( - allocations[0].inner_target(&shape), - Some(AccuracyTarget::epsilon_delta(0.05, 0.01)) - ); - - // Multiplicative composition: (1+ε_i)^2 = 1+ε, not 2ε_i = ε. - let rel_shape = CompositionShape { - metric: ErrorMetric::Relative, - approximate_layer_count: 2, - }; - let allocations = - EqualSplitAllocator.allocations(&AccuracyTarget::epsilon(0.21), &rel_shape); - let e = allocations[0].layers[0].error; - assert!((e - 0.1).abs() < 1e-12); - - assert!(EqualSplitAllocator - .allocations(&AccuracyTarget::exact(), &shape) - .is_empty()); - } -} diff --git a/crates/asap-aware-mapping/src/accuracy/budget.rs b/crates/asap-aware-mapping/src/accuracy/budget.rs new file mode 100644 index 000000000..b8ca82573 --- /dev/null +++ b/crates/asap-aware-mapping/src/accuracy/budget.rs @@ -0,0 +1,93 @@ +//! Split an end-to-end accuracy target equally among the approximate inputs +//! of one composition (#669, part b). +//! +//! When `n` approximate values compose into one output, each is sized for an +//! equal share of the output's target, so the composed bound meets the +//! target with no slack: `ε_i = ε / n` for an additively composed metric, +//! `ε_i = (1 + ε)^{1/n} − 1` for a relative one (relative errors compose as +//! `Π(1 + ε_i) − 1`), and `δ_i = δ / n` (the union bound). Only equal +//! splits are made. +use super::*; + +/// `shares` of `of` equal shares of `target`, in `metric`'s composition: +/// the target of one approximate input (`shares = 1`), or of a nested +/// sub-DAG holding `shares` of the `of` inputs. `None` for an exact target +/// or an ε that is not positive and finite: no approximate input meets it. +/// `of = 1` (or 0) is `target` itself. +pub fn equal_shares( + target: &AccuracyTarget, + metric: ErrorMetric, + of: usize, + shares: usize, +) -> Option { + if target.is_exact() || !(target.error.is_finite() && target.error > 0.0) { + return None; + } + let fraction = shares.min(of.max(1)) as f64 / of.max(1) as f64; + let error = match metric { + ErrorMetric::Relative => (1.0 + target.error).powf(fraction) - 1.0, + _ => target.error * fraction, + }; + Some(AccuracyTarget { + error, + failure_probability: target.failure_probability.map(|delta| delta * fraction), + ..target.clone() + }) +} + +/// One of `approximate_inputs` equal shares of `target` ([`equal_shares`]). +pub fn equal_share( + target: &AccuracyTarget, + metric: ErrorMetric, + approximate_inputs: usize, +) -> Option { + equal_shares(target, metric, approximate_inputs, 1) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Additive metrics split ε and δ by `n`; the `n` shares add up to the + /// target. + #[test] + fn an_additive_target_splits_epsilon_and_delta_equally() { + let target = AccuracyTarget::epsilon_delta(0.1, 0.02); + let share = equal_share(&target, ErrorMetric::ABSOLUTE, 2).unwrap(); + assert_eq!(share, AccuracyTarget::epsilon_delta(0.05, 0.01)); + let rest = equal_shares(&target, ErrorMetric::ABSOLUTE, 3, 2).unwrap(); + assert!((rest.error - 0.1 * 2.0 / 3.0).abs() < 1e-12); + assert!((rest.failure_probability.unwrap() - 0.02 * 2.0 / 3.0).abs() < 1e-12); + assert_eq!( + equal_share(&target, ErrorMetric::Rank, 1), + Some(target.clone()) + ); + assert_eq!( + equal_share(&AccuracyTarget::epsilon(0.1), ErrorMetric::Rank, 2) + .unwrap() + .failure_probability, + None + ); + } + + /// A relative target splits multiplicatively: `(1 + ε_i)^n = 1 + ε`, + /// not `n ε_i = ε`. + #[test] + fn a_relative_target_splits_multiplicatively() { + let share = equal_share(&AccuracyTarget::epsilon(0.21), ErrorMetric::Relative, 2).unwrap(); + assert!((share.error - 0.1).abs() < 1e-12); + } + + /// No approximate input meets an exact target. + #[test] + fn an_exact_target_has_no_share() { + assert_eq!( + equal_share(&AccuracyTarget::exact(), ErrorMetric::ABSOLUTE, 2), + None + ); + assert_eq!( + equal_share(&AccuracyTarget::epsilon(0.0), ErrorMetric::ABSOLUTE, 2), + None + ); + } +} diff --git a/crates/asap-aware-mapping/src/accuracy/mod.rs b/crates/asap-aware-mapping/src/accuracy/mod.rs index c82baa749..3e8de302c 100644 --- a/crates/asap-aware-mapping/src/accuracy/mod.rs +++ b/crates/asap-aware-mapping/src/accuracy/mod.rs @@ -1,19 +1,18 @@ //! Planner accuracy interfaces and model dispatch. //! //! Estimator models derive local guarantees, composition propagates them, -//! allocation proposes local budgets, and evidence supplies scoped contracts. +//! the budget split sizes each approximate input of a composition for an +//! equal share of its target, and evidence supplies scoped contracts. //! Unknown evidence may retain a candidate but does not authorize selection. //! See `docs/design_docs/concepts/accuracy-models.md` for the design. -pub mod allocation; +pub mod budget; pub mod composition; pub mod estimators; pub mod evidence; pub mod reconciliation; -pub use allocation::{ - AccuracyAllocation, AccuracyBudgetAllocator, CompositionShape, EqualSplitAllocator, -}; +pub use budget::{equal_share, equal_shares}; pub use evidence::{ AccuracyEvidenceProvider, NoAccuracyEvidence, OperandDomain, PropagationStats, WorkloadAccuracyEvidence, diff --git a/crates/asap-aware-mapping/src/grouping.rs b/crates/asap-aware-mapping/src/grouping.rs index 1c9eba931..a4777a764 100644 --- a/crates/asap-aware-mapping/src/grouping.rs +++ b/crates/asap-aware-mapping/src/grouping.rs @@ -80,9 +80,7 @@ use asap_types::post_asap::{ }; use asap_types::pre_asap::agg_intent::AggIntent; -use crate::accuracy::{ - AccuracyBudgetAllocator, AccuracyEvidenceProvider, AccuracyModel, PropagationStats, -}; +use crate::accuracy::{AccuracyEvidenceProvider, AccuracyModel, PropagationStats}; use crate::cost_model::{CostModel, DefaultCostModel}; use crate::realization::{accuracy_target, summary_candidates}; use crate::replacement::{ @@ -156,14 +154,12 @@ impl<'a> HydraGroupingStrategy<'a> { pub fn new_with_planning_inputs_and_evidence( cost_model: &'a dyn CostModel, accuracy_model: &'a dyn AccuracyModel, - allocator: &'a dyn AccuracyBudgetAllocator, evidence: &'a dyn AccuracyEvidenceProvider, ) -> Self { Self { planning_inputs: CandidatePlanningInputs { cost: cost_model, accuracy: accuracy_model, - allocator, evidence, }, } @@ -496,7 +492,7 @@ fn hydra_guarantee(inner: &AccuracyGuarantee, stats: &PropagationStats) -> Accur #[cfg(test)] mod tests { use super::*; - use crate::accuracy::{DefaultAccuracyModel, EqualSplitAllocator}; + use crate::accuracy::DefaultAccuracyModel; use crate::test_support::{agg, agg_per_entity, metric_scan}; use asap_types::post_asap::ErrorMetric; use asap_types::pre_asap::agg_intent::{default_cardinality, default_quantile}; @@ -648,7 +644,6 @@ mod tests { let strategy = HydraGroupingStrategy::new_with_planning_inputs_and_evidence( &DefaultCostModel, &DefaultAccuracyModel, - &EqualSplitAllocator, &ZeroSharedGridEvidence, ); let replacements = strategy.replacements(&TargetSubDAG::new(&q)); @@ -685,7 +680,6 @@ mod tests { let strategy = HydraGroupingStrategy::new_with_planning_inputs_and_evidence( &DefaultCostModel, &DefaultAccuracyModel, - &EqualSplitAllocator, &InvalidEvidence, ); let proposals = strategy.propose(&TargetSubDAG::new(&q)); @@ -732,7 +726,6 @@ mod tests { let strategy = HydraGroupingStrategy::new_with_planning_inputs_and_evidence( &DefaultCostModel, &DefaultAccuracyModel, - &EqualSplitAllocator, &ExcessiveCollision, ); let proposals = strategy.propose(&TargetSubDAG::new(&q)); diff --git a/crates/asap-aware-mapping/src/lib.rs b/crates/asap-aware-mapping/src/lib.rs index 41323238f..8a90da3c3 100644 --- a/crates/asap-aware-mapping/src/lib.rs +++ b/crates/asap-aware-mapping/src/lib.rs @@ -133,9 +133,8 @@ //! `sketch_algebra::capability::Capability`/`is_satisfied_by` is the //! reference downstream implementation. //! -//! - [`accuracy`] — the [`AccuracyModel`](accuracy::AccuracyModel) / -//! [`AccuracyBudgetAllocator`](accuracy::AccuracyBudgetAllocator) -//! extension points (issue #172): the planning-time algebra that derives +//! - [`accuracy`] — the [`AccuracyModel`](accuracy::AccuracyModel) +//! extension point (issue #172): the planning-time algebra that derives //! a machine-readable [`AccuracyGuarantee`](asap_types::post_asap::AccuracyGuarantee) //! for every finalized post-ASAP value, propagates it through //! approximate-over-approximate compositions under conservative rules @@ -172,8 +171,7 @@ pub mod topk_reuse; pub use accuracy::reconciliation::AccuracyReconciliationStrategy; pub use accuracy::{ - AccuracyAllocation, AccuracyBudgetAllocator, AccuracyEvidenceProvider, AccuracyModel, - CompositionShape, DefaultAccuracyModel, EqualSplitAllocator, NoAccuracyEvidence, + AccuracyEvidenceProvider, AccuracyModel, DefaultAccuracyModel, NoAccuracyEvidence, PropagationStats, WorkloadAccuracyEvidence, }; pub use cost_model::{ diff --git a/crates/asap-aware-mapping/src/replacement.rs b/crates/asap-aware-mapping/src/replacement.rs index 139050f5b..83ad667f4 100644 --- a/crates/asap-aware-mapping/src/replacement.rs +++ b/crates/asap-aware-mapping/src/replacement.rs @@ -378,8 +378,7 @@ use std::rc::{Rc, Weak}; use crate::accuracy::reconciliation::AccuracyReconciliationStrategy; use crate::accuracy::{ - AccuracyBudgetAllocator, AccuracyEvidenceProvider, AccuracyModel, CompositionShape, - DefaultAccuracyModel, EqualSplitAllocator, NoAccuracyEvidence, + equal_shares, AccuracyEvidenceProvider, AccuracyModel, DefaultAccuracyModel, NoAccuracyEvidence, }; use crate::cost_model::{ raw_recompute_cost_rate, CostModel, CseCandidate, DefaultCostModel, ExactCompositionCostInputs, @@ -1105,30 +1104,27 @@ pub fn posterior_aware_size_params( /// every caller. static DEFAULT_COST_MODEL: DefaultCostModel = DefaultCostModel; static DEFAULT_ACCURACY_MODEL: DefaultAccuracyModel = DefaultAccuracyModel; -static DEFAULT_ALLOCATOR: EqualSplitAllocator = EqualSplitAllocator; static NO_ACCURACY_EVIDENCE: NoAccuracyEvidence = NoAccuracyEvidence; /// The cost, accuracy, allocation, and evidence inputs consulted during -/// candidate construction, bundled so the construction path threads one argument. `cost` ranks and sizes; `accuracy` and `allocator` decide +/// candidate construction, bundled so the construction path threads one argument. `cost` ranks and sizes; `accuracy` decides /// legality (issue #172) — see [`crate::accuracy`]'s module docs for why /// those are separate from `cost` and run before it. #[derive(Clone, Copy)] pub(crate) struct CandidatePlanningInputs<'a> { pub cost: &'a dyn CostModel, pub accuracy: &'a dyn AccuracyModel, - pub allocator: &'a dyn AccuracyBudgetAllocator, pub evidence: &'a dyn AccuracyEvidenceProvider, } impl<'a> CandidatePlanningInputs<'a> { /// `cost` with the built-in [`DefaultAccuracyModel`]/ - /// [`EqualSplitAllocator`] — what every entry point that only takes a + /// equal budget split — what every entry point that only takes a /// `CostModel` uses. pub(crate) fn with_default_accuracy(cost: &'a dyn CostModel) -> Self { Self { cost, accuracy: &DEFAULT_ACCURACY_MODEL, - allocator: &DEFAULT_ALLOCATOR, evidence: &NO_ACCURACY_EVIDENCE, } } @@ -1173,7 +1169,7 @@ impl<'a> ASAPStrategies<'a> { /// A strategy that ranks/binds via `cost_model` instead of the built-in /// static preference order — the same customization point /// [`realizations_for_intent`] already offers. Accuracy legality stays - /// with the built-in [`DefaultAccuracyModel`]/[`EqualSplitAllocator`]. + /// with the built-in [`DefaultAccuracyModel`]. pub fn new(cost_model: &'a dyn CostModel) -> Self { Self { planning_inputs: CandidatePlanningInputs::with_default_accuracy(cost_model), @@ -1188,13 +1184,11 @@ impl<'a> ASAPStrategies<'a> { pub fn new_with_planning_inputs( cost_model: &'a dyn CostModel, accuracy_model: &'a dyn AccuracyModel, - allocator: &'a dyn AccuracyBudgetAllocator, ) -> Self { Self { planning_inputs: CandidatePlanningInputs { cost: cost_model, accuracy: accuracy_model, - allocator, evidence: &NO_ACCURACY_EVIDENCE, }, } @@ -1205,14 +1199,12 @@ impl<'a> ASAPStrategies<'a> { pub fn new_with_planning_inputs_and_evidence( cost_model: &'a dyn CostModel, accuracy_model: &'a dyn AccuracyModel, - allocator: &'a dyn AccuracyBudgetAllocator, evidence: &'a dyn AccuracyEvidenceProvider, ) -> Self { Self { planning_inputs: CandidatePlanningInputs { cost: cost_model, accuracy: accuracy_model, - allocator, evidence, }, } @@ -1535,64 +1527,59 @@ impl<'a> ASAPStrategies<'a> { else { continue; }; - let shape = CompositionShape { - metric: local.spec.metric, - approximate_layer_count: 1 + child_layers, - }; - let allocations = planning_inputs.allocator.allocations(target, &shape); - if allocations.is_empty() { + let (metric, layer_count) = (local.spec.metric, 1 + child_layers); + let Some(outer_target) = equal_shares(target, metric, layer_count, 1) else { proposals.rejected.push(RejectedCandidate { strategy: "ASAPStrategies", description: rationale.clone(), error: AccuracyError::NoLegalAllocation { target: target.clone(), - layer_count: shape.approximate_layer_count, + layer_count, }, }); continue; - } + }; let declared_child_target = aggregate_child(root) .and_then(|child| bindable_intent(child)) .and_then(accuracy_target); - for allocation in allocations { - let outer_target = &allocation.layers[0]; - let inner_target = allocation.inner_target(&shape); - let (eps, delta) = accuracy_budget(outer_target); - let resized = Realization::Sketch(SketchKind::new( - kind.algorithm().clone(), - planning_inputs - .cost - .size_params(kind.algorithm().clone(), intent, eps, delta), - )); - // Identical to the as-declared composition already recorded - // above — nothing new to propose. - if resized == realization && inner_target.as_ref() == declared_child_target { - continue; - } - let note = GuaranteeSource::BudgetAllocation { - allocator: allocation.allocator.to_string(), - layer: 0, - layer_count: shape.approximate_layer_count, - local_target: outer_target.clone(), - end_to_end_target: target.clone(), - }; - proposals.record( - format!( - "{rationale}; sized under {} budget split of {target:?} across \ - {} approximate layers (this layer {outer_target:?}, child sub-DAG \ - {inner_target:?})", - allocation.allocator, shape.approximate_layer_count - ), - construct_summary_with( - root, - intent, - resized, - planning_inputs, - inner_target.as_ref(), - Some(note), - ), - ); + let inner_target = (layer_count > 1) + .then(|| equal_shares(target, metric, layer_count, layer_count - 1)) + .flatten(); + let (eps, delta) = accuracy_budget(&outer_target); + let resized = Realization::Sketch(SketchKind::new( + kind.algorithm().clone(), + planning_inputs + .cost + .size_params(kind.algorithm().clone(), intent, eps, delta), + )); + // Identical to the as-declared composition already recorded + // above — nothing new to propose. + if resized == realization && inner_target.as_ref() == declared_child_target { + continue; } + let note = GuaranteeSource::BudgetAllocation { + allocator: "equal_split".to_string(), + layer: 0, + layer_count, + local_target: outer_target.clone(), + end_to_end_target: target.clone(), + }; + proposals.record( + format!( + "{rationale}; sized under {} budget split of {target:?} across \ + {} approximate layers (this layer {outer_target:?}, child sub-DAG \ + {inner_target:?})", + "equal_split", layer_count + ), + construct_summary_with( + root, + intent, + resized, + planning_inputs, + inner_target.as_ref(), + Some(note), + ), + ); } if proposals.candidates.is_empty() { if let Some(error) = &proposals.domain_error { @@ -1795,7 +1782,7 @@ pub(crate) fn realize_child( /// [`realize_child`] with every model explicit, plus an optional /// `end_to_end_target` for `root`'s own value (issue #172): when an -/// [`AccuracyBudgetAllocator`] hands an approximate child a share of its +/// equal budget split ([`equal_shares`]) hands an approximate child a share of its /// parent's budget, the child is re-enumerated with that share substituted /// for its declared `AccuracyTarget` — sizing its sketch (and, recursively, /// re-splitting for its own approximate children) under the allocated @@ -2048,25 +2035,14 @@ fn realize_binary( { return Ok(None); } - let shape = CompositionShape { - metric, - approximate_layer_count: approximate.len(), - }; - let Some(allocation) = planning_inputs - .allocator - .allocations(target, &shape) - .into_iter() - .next() - else { + let Some(local_target) = equal_shares(target, metric, approximate.len(), 1) else { return Ok(None); }; - for ((operand_index, _), local_target) in - approximate.into_iter().zip(allocation.layers.iter()) - { + for (operand_index, _) in approximate { if operand_index == 0 { - lhs_node = realize_binary_operand(lhs, planning_inputs, Some(local_target))?; + lhs_node = realize_binary_operand(lhs, planning_inputs, Some(&local_target))?; } else { - rhs_node = realize_binary_operand(rhs, planning_inputs, Some(local_target))?; + rhs_node = realize_binary_operand(rhs, planning_inputs, Some(&local_target))?; } } } @@ -6002,14 +5978,12 @@ pub fn default_strategies_with_evidence<'a>( Box::new(ASAPStrategies::new_with_planning_inputs_and_evidence( cost_model, &DEFAULT_ACCURACY_MODEL, - &DEFAULT_ALLOCATOR, evidence, )), Box::new( HydraGroupingStrategy::new_with_planning_inputs_and_evidence( cost_model, &DEFAULT_ACCURACY_MODEL, - &DEFAULT_ALLOCATOR, evidence, ), ), @@ -7956,12 +7930,9 @@ mod tests { let inner = agg(vec![2], default_quantile(0.5), metric_scan(&["job"])); let outer = agg(vec![], default_quantile(0.99), inner); let target = TargetSubDAG::new(&outer); - let replacements = ASAPStrategies::new_with_planning_inputs( - &DefaultCostModel, - &RankAdditiveModel, - &EqualSplitAllocator, - ) - .replacements(&target); + let replacements = + ASAPStrategies::new_with_planning_inputs(&DefaultCostModel, &RankAdditiveModel) + .replacements(&target); assert_eq!(replacements.len(), 2, "{replacements:?}"); assert!(replacements.iter().all(|candidate| { @@ -10160,7 +10131,6 @@ mod tests { let strategy = ASAPStrategies::new_with_planning_inputs_and_evidence( &DefaultCostModel, &DefaultAccuracyModel, - &EqualSplitAllocator, &SeparatedTopKEvidence, ); let replacements = strategy.replacements(&TargetSubDAG::new(&q)); @@ -10194,7 +10164,6 @@ mod tests { let strategy = ASAPStrategies::new_with_planning_inputs_and_evidence( &DefaultCostModel, &DefaultAccuracyModel, - &EqualSplitAllocator, &SeparatedTopKEvidence, ); let candidates = strategy.replacements(&TargetSubDAG::new(&outer)); @@ -10261,7 +10230,6 @@ mod tests { let strategy = ASAPStrategies::new_with_planning_inputs_and_evidence( &DefaultCostModel, &DefaultAccuracyModel, - &EqualSplitAllocator, &SeparatedTopKEvidence, ); let candidates = strategy.replacements(&TargetSubDAG::new(&outer)); @@ -10336,7 +10304,6 @@ mod tests { let strategy = ASAPStrategies::new_with_planning_inputs_and_evidence( &DefaultCostModel, &DefaultAccuracyModel, - &EqualSplitAllocator, &SeparatedTopKEvidence, ); let candidates = strategy.replacements(&TargetSubDAG::new(&outer)); @@ -10388,7 +10355,6 @@ mod tests { let strategy = ASAPStrategies::new_with_planning_inputs_and_evidence( &DefaultCostModel, &DefaultAccuracyModel, - &EqualSplitAllocator, &SeparatedTopKEvidence, ); @@ -10441,7 +10407,6 @@ mod tests { let strategy = ASAPStrategies::new_with_planning_inputs_and_evidence( &DefaultCostModel, &DefaultAccuracyModel, - &EqualSplitAllocator, &SeparatedTopKEvidence, ); let (update, reduction) = strategy @@ -10706,11 +10671,8 @@ mod tests { // summary levels explicit while preserving the composed guarantee. let inner = agg(vec![2], quantile_eps(0.5, 0.1), metric_scan(&["job"])); let outer = agg(vec![], quantile_eps(0.99, 0.1), inner); - let strategy = ASAPStrategies::new_with_planning_inputs( - &DefaultCostModel, - &RankAdditiveModel, - &EqualSplitAllocator, - ); + let strategy = + ASAPStrategies::new_with_planning_inputs(&DefaultCostModel, &RankAdditiveModel); let proposals = strategy.propose(&TargetSubDAG::new(&outer)); assert!(!proposals.candidates.is_empty()); @@ -10736,12 +10698,9 @@ mod tests { // and global cost ranking. let inner = agg(vec![2], quantile_eps(0.5, 0.1), metric_scan(&["job"])); let outer = agg(vec![], quantile_eps(0.99, 0.1), inner); - let strategies: Vec> = - vec![Box::new(ASAPStrategies::new_with_planning_inputs( - &DefaultCostModel, - &RankAdditiveModel, - &EqualSplitAllocator, - ))]; + let strategies: Vec> = vec![Box::new( + ASAPStrategies::new_with_planning_inputs(&DefaultCostModel, &RankAdditiveModel), + )]; let space = search_workload_with(vec![("q", Rc::clone(&outer))], &strategies).unwrap(); let root = &space.roots[0].1; let group = space.candidates_for_target(root).unwrap(); diff --git a/crates/integration-tests/tests/promql_to_post_asap.rs b/crates/integration-tests/tests/promql_to_post_asap.rs index f14358b94..9ab239347 100644 --- a/crates/integration-tests/tests/promql_to_post_asap.rs +++ b/crates/integration-tests/tests/promql_to_post_asap.rs @@ -9,8 +9,7 @@ use std::rc::Rc; use asap_aware_mapping::accuracy::{ - AccuracyEvidenceProvider, DefaultAccuracyModel, EqualSplitAllocator, OperandDomain, - PropagationStats, + AccuracyEvidenceProvider, DefaultAccuracyModel, OperandDomain, PropagationStats, }; use asap_aware_mapping::cost_model::DefaultCostModel; use asap_aware_mapping::replacement::{retain_exact, RealizationError}; @@ -274,7 +273,6 @@ fn grouped_rate_topk_consumes_finalized_rate_values() { let strategy = ASAPStrategies::new_with_planning_inputs_and_evidence( &DefaultCostModel, &DefaultAccuracyModel, - &EqualSplitAllocator, &SeparatedTopK, ); let plan = strategy @@ -338,7 +336,6 @@ fn weighted_topk_keeps_candidates_with_missing_population_evidence() { let strategy = ASAPStrategies::new_with_planning_inputs_and_evidence( &DefaultCostModel, &DefaultAccuracyModel, - &EqualSplitAllocator, &NoPopulationBound, ); let candidates = strategy.replacements(&TargetSubDAG::new(&root)); @@ -389,7 +386,6 @@ fn weighted_topk_rejects_invalid_population_evidence() { let strategy = ASAPStrategies::new_with_planning_inputs_and_evidence( &DefaultCostModel, &DefaultAccuracyModel, - &EqualSplitAllocator, &InvalidPopulation, ); assert!(strategy.replacements(&TargetSubDAG::new(&root)).is_empty()); @@ -407,7 +403,6 @@ fn rate_and_increase_topk_use_summary_scores_and_grouped_limits() { let strategy = ASAPStrategies::new_with_planning_inputs_and_evidence( &DefaultCostModel, &DefaultAccuracyModel, - &EqualSplitAllocator, &SeparatedTopK, ); let plan = strategy @@ -642,7 +637,6 @@ fn planner_only_e2e_temporal_topk_preserves_query_update_and_evaluation_contract let strategy = ASAPStrategies::new_with_planning_inputs_and_evidence( &DefaultCostModel, &DefaultAccuracyModel, - &EqualSplitAllocator, &SeparatedTopK, ); let candidate = strategy @@ -804,7 +798,6 @@ fn planner_heap_topk_reference_execution_matches_ground_truth() { let strategy = ASAPStrategies::new_with_planning_inputs_and_evidence( &DefaultCostModel, &DefaultAccuracyModel, - &EqualSplitAllocator, &SeparatedTopK, ); // This reference executor consumes keyed heap updates. The inventory @@ -1237,7 +1230,6 @@ fn ddsketch_ratio_rejects_unsafe_domains() { let strategy = ASAPStrategies::new_with_planning_inputs_and_evidence( &DefaultCostModel, &DefaultAccuracyModel, - &EqualSplitAllocator, &evidence, ); let replacements = strategy.replacements(&TargetSubDAG::new(&pre)); @@ -1278,7 +1270,6 @@ fn ddsketch_ratio_rejects_one_invalid_domain_when_the_other_is_missing() { let strategy = ASAPStrategies::new_with_planning_inputs_and_evidence( &DefaultCostModel, &DefaultAccuracyModel, - &EqualSplitAllocator, &PartialUnsafeDomain, ); assert!(strategy.replacements(&TargetSubDAG::new(&pre)).is_empty()); @@ -1300,7 +1291,6 @@ fn ddsketch_ratio_bound_holds_for_signed_pinned_sketch_evaluations() { let strategy = ASAPStrategies::new_with_planning_inputs_and_evidence( &DefaultCostModel, &DefaultAccuracyModel, - &EqualSplitAllocator, &evidence, ); let candidates = strategy.replacements(&TargetSubDAG::new(&pre)); @@ -1383,7 +1373,6 @@ fn ddsketch_ratio_requires_a_supported_population_size() { let strategy = ASAPStrategies::new_with_planning_inputs_and_evidence( &DefaultCostModel, &DefaultAccuracyModel, - &EqualSplitAllocator, &evidence, ); assert!(strategy.replacements(&TargetSubDAG::new(&pre)).is_empty()); diff --git a/crates/types/src/post_asap/guarantee.rs b/crates/types/src/post_asap/guarantee.rs index 42c45e62b..00cc73f84 100644 --- a/crates/types/src/post_asap/guarantee.rs +++ b/crates/types/src/post_asap/guarantee.rs @@ -364,7 +364,7 @@ pub enum GuaranteeSource { rule: String, }, /// The budget split that produced this layer's local target — present - /// only when an `AccuracyBudgetAllocator` re-sized a layer. + /// only when an equal budget split re-sized a layer. BudgetAllocation { allocator: String, layer: usize, @@ -467,8 +467,7 @@ impl AccuracyGuarantee { /// How many approximate sketch evaluations contributed to this value — /// `1` for a plain evaluation, `0` for an exact value, and the transitive /// count through every [`GuaranteeSource::ChildGuarantee`] for a - /// composition. An `AccuracyBudgetAllocator` uses this as the number - /// of layers a budget must be split across. + /// composition. pub fn approximate_layer_count(&self) -> usize { self.provenance .iter() From 0186669f6518798edb16044985bccbd944f7b137 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Thu, 15 Oct 2026 15:21:44 +0000 Subject: [PATCH 2/4] feat(planner): split a query's error budget among its approximate inputs at Stage 1 composition (#669) When a query's value composes several approximate inputs (every sketch summary below its root, when the composed guarantee is not exact), compose_logical_candidate sizes each for an equal share of the query's target: eps/n (relative: (1+eps)^(1/n)-1) and delta/n. Only inputs approximate in that candidate count; a summary several queries read takes the largest n. The summaries are first sized for the whole budget, then the candidate is composed again with the shares. --- .../src/per_computation_candidates.rs | 322 ++++++++++++++++-- 1 file changed, 289 insertions(+), 33 deletions(-) diff --git a/crates/asap-aware-mapping/src/per_computation_candidates.rs b/crates/asap-aware-mapping/src/per_computation_candidates.rs index 440eb56f8..f63e98d18 100644 --- a/crates/asap-aware-mapping/src/per_computation_candidates.rs +++ b/crates/asap-aware-mapping/src/per_computation_candidates.rs @@ -52,9 +52,16 @@ //! groups. An `AboveOutput` `HAVING` stays a `Filter` above the output. //! //! Summaries are sized here, each for its own measure's accuracy requirement. -//! Sizing belongs after Pass 2, where a shared summary takes the strictest -//! requirement among its consumers; Pass 2 sharing is not implemented yet, so -//! this is that post-Pass-2 sizing step with no sharing. +//! When a query's value composes several approximate inputs (every sketch +//! summary below its root, when its guarantee is not exact), each is sized +//! for an equal share of that requirement: `ε/n` (a relative error +//! `(1 + ε)^{1/n} − 1`) and `δ/n` for `n` inputs (#669b). Only the inputs +//! approximate in this candidate count. Stage 1 sizes for ε and δ per output +//! value; plan selection enlarges the parameters that depend on δ for a +//! target over the whole result. Sizing belongs after Pass 2, where a shared +//! summary takes the strictest requirement among its consumers; Pass 2 +//! sharing is not implemented yet, so this is that post-Pass-2 sizing step +//! with no sharing. //! //! Every node of the candidate gets a theoretical guarantee, composed with //! the built-in accuracy model through operators (#669): exact over exact @@ -96,7 +103,7 @@ use crate::accuracy::composition::{ PRODUCT_OPERAND_DOMAINS, QUOTIENT_OPERAND_DOMAINS, }; use crate::accuracy::estimators::{hll, univmon}; -use crate::accuracy::{AccuracyModel, DefaultAccuracyModel, PropagationStats}; +use crate::accuracy::{equal_share, AccuracyModel, DefaultAccuracyModel, PropagationStats}; use crate::function_rules::function_rules; use crate::realization::{ accuracy_budget, accuracy_target, default_size_params, summary_candidates, @@ -718,9 +725,11 @@ fn with_samples_requirement(mut update: SummaryUpdate, metric: Option<&str>) -> /// accumulators divided, and any other measure by its own single-measure /// aggregate; several measures are joined back on the group keys. A summary /// is sized here for its measure's accuracy requirement; Pass 1 leaves it -/// unsized. A measure filter becomes `SummaryAgg.filter`; `HAVING` is -/// applied as its target records. A sub-DAG shared between roots stays shared. -/// Every node carries the guarantee [`composed_guarantee`] gives it. +/// unsized; several approximate inputs of one query share its budget +/// equally ([`budget_shares`]). A measure filter becomes `SummaryAgg.filter`; +/// `HAVING` is applied as its target records. A sub-DAG shared between roots +/// stays shared. Every node carries the guarantee [`composed_guarantee`] +/// gives it. pub fn compose_logical_candidate( inventory: &PerComputationCandidates, choice: &[usize], @@ -749,12 +758,120 @@ pub fn compose_logical_candidate( .filter(|entry| entry.placement == HavingPlacement::BelowAggregate) .filter_map(|entry| entry.filter.as_ref().map(Rc::as_ptr)) .collect(); + // Each summary is first sized for its measure's whole budget. A query + // whose value composes several approximate inputs is then composed + // again with each sized for an equal share ([`budget_shares`]). + let whole = Sizing::default(); + let roots = compose_roots(inventory, &chosen, &pushed_below, &whole)?; + let shares = budget_shares(&roots, &whole.built.into_inner()); + if shares.is_empty() { + return Ok(roots); + } + let shared = Sizing { + shares, + ..Sizing::default() + }; + compose_roots(inventory, &chosen, &pushed_below, &shared) +} + +/// One measure of an aggregate: the aggregate and the measure's index. +type MeasureKey = (*const OperatorNode, usize); + +/// How [`compose_logical_candidate`] sizes summaries in one composition. +#[derive(Default)] +struct Sizing { + /// How many approximate inputs each measure's summary shares its + /// query's budget with; a measure not listed has the whole budget. + shares: HashMap, + /// Every sketch summary state built, with the measure it sizes. + built: std::cell::RefCell, MeasureKey)>>, +} + +/// The equal budget split among the approximate inputs of each query +/// (#669b): a query whose value carries a guarantee that is not exact +/// composes every sketch summary state below its root +/// ([`approximate_inputs`]), so with `n ≥ 2` of them each is sized for one +/// of `n` equal shares of the query's target. A state several queries read +/// takes the largest `n`. Only states `built` lists (by the measure they +/// size) are split. Empty when no query composes two approximate inputs. +fn budget_shares( + roots: &[(impl Sized, QueryRoot)], + built: &[(Rc, MeasureKey)], +) -> HashMap { + let measure: HashMap<_, _> = built + .iter() + .map(|(state, key)| (Rc::as_ptr(state), *key)) + .collect(); + let mut shares = HashMap::new(); + for (_, root) in roots { + let inputs = approximate_inputs(root); + if inputs.len() < 2 { + continue; + } + for state in &inputs { + if let Some(key) = measure.get(&Rc::as_ptr(state)) { + let share = shares.entry(*key).or_insert(1); + *share = inputs.len().max(*share); + } + } + } + shares +} + +/// The approximate inputs a query's value composes: every sketch summary +/// state reachable from `root`, each once, when the value carries a +/// guarantee that is not exact ([`root_guarantee`]). Empty when it carries +/// none (no rule composes the inputs) or an exact one. Plan selection sizes +/// each for its share of the query's failure probability too. +pub(crate) fn approximate_inputs(root: &QueryRoot) -> Vec> { + if root_guarantee(root).is_none_or(|g| g.is_exact()) { + return Vec::new(); + } + let operators = match root { + QueryRoot::Operator(node) => vec![node], + QueryRoot::Scalar(expr) => expr.operator_refs(), + }; + let mut seen = HashSet::new(); + operators + .into_iter() + .flat_map(|node| OperatorNode::reachable(node)) + .filter(|node| { + matches!( + node.asap(), + Some(ASAPOp::SummaryAgg { + family: FieldDataType::Sketch(..), + .. + }) + ) + }) + .filter(|node| seen.insert(Rc::as_ptr(node))) + .collect() +} + +/// The theoretical guarantee of a query's output, as composed: the root +/// node's own, or a scalar root's composed over the operators it reads +/// ([`scalar_guarantee`]). +pub(crate) fn root_guarantee(root: &QueryRoot) -> Option { + match root { + QueryRoot::Operator(node) => node.guarantee.clone(), + QueryRoot::Scalar(expr) => scalar_guarantee(expr, &|_| None), + } +} + +/// Every root of `inventory` with the chosen candidates substituted, each +/// summary sized as `sizing` says, and every node with its guarantee. +fn compose_roots( + inventory: &PerComputationCandidates, + chosen: &Chosen<'_>, + pushed_below: &HashSet<*const OperatorNode>, + sizing: &Sizing, +) -> Result, PerComputationCandidateError> { let (mut memo, mut guaranteed) = (HashMap::new(), HashMap::new()); let mut rewrite = |node: &Rc| { - let node = rewrite(node, &chosen, &pushed_below, &mut memo)?; + let node = rewrite(node, chosen, pushed_below, sizing, &mut memo)?; with_guarantees(&node, &mut guaranteed) }; - inventory + let roots = inventory .roots .iter() .map(|(id, root)| { @@ -775,7 +892,14 @@ pub fn compose_logical_candidate( root.validate_structure()?; Ok((id.clone(), root)) }) - .collect() + .collect::, PerComputationCandidateError>>()?; + // The built states, as the roots hold them: with their guarantees. + for (state, _) in sizing.built.borrow_mut().iter_mut() { + if let Some(with_guarantee) = guaranteed.get(&Rc::as_ptr(state)) { + *state = with_guarantee.clone(); + } + } + Ok(roots) } /// Each aggregate's targets with their chosen candidates. @@ -795,20 +919,21 @@ fn rewrite( node: &Rc, chosen: &Chosen<'_>, pushed_below: &HashSet<*const OperatorNode>, + sizing: &Sizing, memo: &mut Memo, ) -> Result, PerComputationCandidateError> { if let Some(done) = memo.get(&Rc::as_ptr(node)) { return Ok(done.clone()); } for child in node.children() { - rewrite(child, chosen, pushed_below, memo)?; + rewrite(child, chosen, pushed_below, sizing, memo)?; } let changed = node .children() .iter() .any(|child| !Rc::ptr_eq(child, &memo[&Rc::as_ptr(child)])); let rebuilt = match (chosen.get(&Rc::as_ptr(node)), node.non_asap()) { - (Some(targets), _) if substituted(targets) => substitute(node, targets, memo)?, + (Some(targets), _) if substituted(targets) => substitute(node, targets, sizing, memo)?, (_, Some(NonASAPOp::Filter { child, .. })) if pushed_below.contains(&Rc::as_ptr(node)) => { memo[&Rc::as_ptr(child)].clone() } @@ -878,8 +1003,10 @@ fn with_guarantees( /// keeps its child's guarantee. A summary state carries none itself. /// /// A guarantee holds per output point (per series or group), with its -/// failure probability per point. It is not scaled by the number of -/// output points (`G·δ` over `G` groups); that is #669 part b. +/// failure probability per point ([`AccuracyScope::PerValue`]). Plan +/// selection states it over a whole result of `G` points as `G·δ`. +/// +/// [`AccuracyScope::PerValue`]: asap_types::types::AccuracyScope::PerValue fn composed_guarantee(node: &OperatorNode) -> Option { let exact = |node: &OperatorNode| { node.guarantee @@ -1940,8 +2067,9 @@ fn estimate_bound_over_samples( /// measure that is no target) by its substitution, with the same output /// columns. fn substitute( - target: &OperatorNode, + target: &Rc, targets: &[(&ComputationTarget, &ComputationCandidate)], + sizing: &Sizing, memo: &Memo, ) -> Result, PerComputationCandidateError> { let Some(NonASAPOp::Aggregate { @@ -1990,10 +2118,26 @@ fn substitute( |(target, candidate)| (*candidate, target.samples_of.as_deref()), ); let filter = filters.get(measure).and_then(Option::as_ref).map(rewritten); + let key = (Rc::as_ptr(target), measure); if let ComputationCandidate::WholeExpression(algorithm) = candidate { - return whole_expression(target, intent, algorithm, samples_of, memo); + return whole_expression( + target, + intent, + algorithm, + samples_of, + (sizing, key), + memo, + ); } - substitute_measure(target, intent, candidate, filter, &input, samples_of) + substitute_measure( + target, + intent, + candidate, + filter, + &input, + samples_of, + (sizing, key), + ) }) .collect::, _>>()?; let mut output = match <[_; 1]>::try_from(parts) { @@ -2058,6 +2202,7 @@ fn whole_expression( intent: &AggIntent, algorithm: &SketchAlgorithm, samples_of: Option<&str>, + (sizing, key): (&Sizing, MeasureKey), memo: &Memo, ) -> Result, PerComputationCandidateError> { let (input, update, partitions) = whole_expression_input(target).ok_or( @@ -2066,11 +2211,15 @@ fn whole_expression( let state = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryAgg { child: memo[&Rc::as_ptr(&input)].clone(), input: with_samples_requirement(update, samples_of), - family: FieldDataType::Sketch(sized(algorithm, intent), GroupingStrategy::default()), + family: FieldDataType::Sketch( + sized(algorithm, intent, sizing.share(key)), + GroupingStrategy::default(), + ), reduction: partitions, grouping: GroupingStrategy::default(), filter: None, }))?; + sizing.built.borrow_mut().push((state.clone(), key)); Ok(OperatorNode::new_shared(Operator::ASAP( ASAPOp::SummaryEstimate { summary_input: state, @@ -2087,6 +2236,7 @@ fn substitute_measure( filter: Option, input: &Rc, samples_of: Option<&str>, + (sizing, key): (&Sizing, MeasureKey), ) -> Result, PerComputationCandidateError> { let Some(NonASAPOp::Aggregate { reduction, .. }) = target.non_asap() else { unreachable!("substitute checked the aggregate") @@ -2128,14 +2278,18 @@ fn substitute_measure( finalized(kind.clone(), params.clone(), intent, read)? } ComputationCandidate::Summary(algorithm) => { - let family = - FieldDataType::Sketch(sized(algorithm, intent), GroupingStrategy::default()); + let family = FieldDataType::Sketch( + sized(algorithm, intent, sizing.share(key)), + GroupingStrategy::default(), + ); + let summary_input = state( + family, + intent, + measure_input(intent, input, filter.as_ref())?, + )?; + sizing.built.borrow_mut().push((summary_input.clone(), key)); OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryEstimate { - summary_input: state( - family, - intent, - measure_input(intent, input, filter.as_ref())?, - )?, + summary_input, query: statistic(intent)?, }))? } @@ -2381,14 +2535,42 @@ fn key_items(keys: usize) -> Vec { .collect() } -/// `algorithm` sized for `intent`'s accuracy requirement. -fn sized(algorithm: &SketchAlgorithm, intent: &AggIntent) -> SketchKind { - let (epsilon, delta) = - accuracy_budget(accuracy_target(intent).unwrap_or(&AccuracyTarget::exact())); - SketchKind::new( - algorithm.clone(), - default_size_params(algorithm.clone(), intent, epsilon, delta), - ) +impl Sizing { + /// How many approximate inputs `key`'s summary shares its query's + /// budget with. + fn share(&self, key: MeasureKey) -> usize { + self.shares.get(&key).copied().unwrap_or(1) + } +} + +/// `algorithm` sized for one of `share` equal shares of `intent`'s accuracy +/// requirement ([`equal_share`]), split in the metric of the summary's own +/// bound: a relative error splits multiplicatively. +fn sized(algorithm: &SketchAlgorithm, intent: &AggIntent, share: usize) -> SketchKind { + let target = accuracy_target(intent) + .cloned() + .unwrap_or_else(AccuracyTarget::exact); + let size = |target: &AccuracyTarget| { + let (epsilon, delta) = accuracy_budget(target); + SketchKind::new( + algorithm.clone(), + default_size_params(algorithm.clone(), intent, epsilon, delta), + ) + }; + let whole = size(&target); + if share <= 1 { + return whole; + } + let metric = statistic(intent) + .ok() + .and_then(|query| { + DefaultAccuracyModel.local_guarantee( + &FieldDataType::Sketch(whole.clone(), GroupingStrategy::default()), + &query, + ) + }) + .map_or(target.metric, |guarantee| guarantee.spec.metric); + equal_share(&target, metric, share).map_or(whole, |share| size(&share)) } /// What a measure's state reads from each row of `input`, resolvable there. @@ -3530,6 +3712,80 @@ mod tests { assert_eq!(stats.sign_statistic, None); } + /// The sketch parameters of every summary state below `root`. + fn sketch_params(root: &Rc) -> Vec { + OperatorNode::reachable(root) + .into_iter() + .filter_map(|node| match node.asap() { + Some(ASAPOp::SummaryAgg { + family: FieldDataType::Sketch(kind, _), + .. + }) => Some(kind.params().clone()), + _ => None, + }) + .collect() + } + + /// `query` composed with target `t` at the candidate `pick(t, its + /// candidates)` returns the position of. + fn composed_picking( + query: &str, + pick: impl Fn(usize, &[ComputationCandidate]) -> usize, + ) -> Rc { + let inventory = inventory(query); + let choice: Vec = inventory + .targets + .iter() + .enumerate() + .map(|(t, target)| pick(t, &target.choices.candidates)) + .collect(); + let composed = compose_logical_candidate(&inventory, &choice).unwrap(); + let QueryRoot::Operator(root) = &composed[0].1 else { + panic!("operator root") + }; + root.clone() + } + + fn position_of(candidates: &[ComputationCandidate], candidate: ComputationCandidate) -> usize { + candidates + .iter() + .position(|c| *c == candidate) + .expect("the candidate is offered") + } + + fn cms(eps: f64, delta: f64) -> SketchParams { + SketchParams::Cms { + width: crate::accuracy::estimators::cms::cms_width(eps), + depth: crate::accuracy::estimators::cms::cms_depth(delta), + } + } + + /// A combination of approximate values splits the query's budget + /// equally among them (#669b): the difference of two Count-Min counts + /// sizes each for ε/2 and δ/2 of the query's ε = 0.01, δ = 0.001. + #[test] + fn approximate_inputs_of_a_combination_share_the_budget_equally() { + let both = composed_picking("count by (job) (a) - count by (job) (b)", |_, c| { + position_of(c, ComputationCandidate::Summary(SketchAlgorithm::Cms)) + }); + assert!(both.guarantee.as_ref().is_some_and(|g| !g.is_exact())); + assert_eq!(sketch_params(&both), vec![cms(0.005, 0.0005); 2]); + } + + /// Only the inputs that are approximate in the combination share the + /// budget: with one count exact, the other gets all of it. + #[test] + fn an_exact_input_takes_no_share_of_the_budget() { + let one = composed_picking("count by (job) (a) - count by (job) (b)", |t, c| { + if t == 0 { + position_of(c, ComputationCandidate::Summary(SketchAlgorithm::Cms)) + } else { + 0 + } + }); + assert_eq!(sketch_params(&one), vec![cms(0.01, 0.001)]); + } + /// The first composed candidate of `query`. fn inventory_root(query: &str) -> Rc { let inventory = inventory(query); From deaf87e0036e81bcd46c048083299908702e4a3f Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Thu, 15 Oct 2026 15:47:33 +0000 Subject: [PATCH 3/4] feat(planner): whole-result delta scaling and Stage 3 failure-probability sizing (#669) A per-value guarantee no longer meets a whole-result target directly. Plan selection states it over the G values one evaluation returns: delta_whole = G * delta_per (whole_result_guarantee). G is 1 for a scalar root and the declared input cardinality for a PromQL instant vector; a SQL relation or a range vector is unbound, which leaves a whole-result delta target unmet. A whole-result guarantee meets a per-value target directly, and the scope does not matter without a delta requirement or for a guarantee that never fails. Before pricing, Stage 3 enlarges each summary for delta / (n * G) of every query reading it (n approximate inputs): Count-Min and Count Sketch depth, KLL's failure probability, a distinct-count UnivMon's heap, and classic HLL's precision from its source's distinct bound. The guarantees above are composed again, the selected plan and the selected logical roots carry the enlarged sketches, and Stage 2's placement is kept. The legacy replacement search checks each node per value, as it sizes it. The PromQL stand-in of Example 2 states its targets per value: its declared input cardinality counts source IPs, not series. --- .../src/accuracy/estimators/cms.rs | 1 + .../src/accuracy/estimators/count_sketch.rs | 4 +- .../src/accuracy/estimators/hll.rs | 9 + .../src/accuracy/estimators/kll.rs | 38 ++- .../src/accuracy/estimators/mod.rs | 153 +++++++++ .../src/accuracy/estimators/univmon.rs | 29 ++ crates/asap-aware-mapping/src/accuracy/mod.rs | 132 ++++++-- .../src/pass/stage1_to_stage3.rs | 14 +- .../src/per_computation_candidates.rs | 92 ++++- crates/asap-aware-mapping/src/replacement.rs | 13 +- .../src/stage3_plan_selection.rs | 313 ++++++++++++++++-- .../tests/planner_layering_example2.rs | 41 ++- 12 files changed, 770 insertions(+), 69 deletions(-) diff --git a/crates/asap-aware-mapping/src/accuracy/estimators/cms.rs b/crates/asap-aware-mapping/src/accuracy/estimators/cms.rs index ddcc4afb4..9dbaed597 100644 --- a/crates/asap-aware-mapping/src/accuracy/estimators/cms.rs +++ b/crates/asap-aware-mapping/src/accuracy/estimators/cms.rs @@ -59,6 +59,7 @@ mod tests { &g, &AccuracyTarget::epsilon_delta(0.01, 0.001) .with_metric(ErrorMetric::ABSOLUTE_STREAM_L1) + .with_scope(asap_types::types::AccuracyScope::PerValue) )); } diff --git a/crates/asap-aware-mapping/src/accuracy/estimators/count_sketch.rs b/crates/asap-aware-mapping/src/accuracy/estimators/count_sketch.rs index f0293dcd8..b97db7159 100644 --- a/crates/asap-aware-mapping/src/accuracy/estimators/count_sketch.rs +++ b/crates/asap-aware-mapping/src/accuracy/estimators/count_sketch.rs @@ -90,7 +90,9 @@ mod tests { assert_eq!(guarantee.spec.metric, ErrorMetric::ABSOLUTE_STREAM_L2); assert!(DefaultAccuracyModel.satisfies( &guarantee, - &AccuracyTarget::epsilon_delta(0.01, 0.01).with_metric(ErrorMetric::ABSOLUTE_STREAM_L2) + &AccuracyTarget::epsilon_delta(0.01, 0.01) + .with_metric(ErrorMetric::ABSOLUTE_STREAM_L2) + .with_scope(asap_types::types::AccuracyScope::PerValue) )); } diff --git a/crates/asap-aware-mapping/src/accuracy/estimators/hll.rs b/crates/asap-aware-mapping/src/accuracy/estimators/hll.rs index 51ff31969..1e3d1d711 100644 --- a/crates/asap-aware-mapping/src/accuracy/estimators/hll.rs +++ b/crates/asap-aware-mapping/src/accuracy/estimators/hll.rs @@ -192,6 +192,15 @@ impl ClassicHllConfidence { } } +/// The smallest precision from `precision` up whose classic failure +/// probability over a source of at most `max_distinct` distinct items is at +/// most `delta` (#669b); `precision` when none up to 18 is. +pub(crate) fn precision_for(precision: u8, max_distinct: u64, delta: f64) -> u8 { + (precision..=18) + .find(|&p| classic_failure_probability(p, max_distinct) <= delta) + .unwrap_or(precision) +} + /// HLL RSE-magnitude inversion: sized from ε alone. Its confidence comes /// from the source's distinct bound ([`classic_guarantee`]). pub(crate) fn hll_precision(eps: f64) -> u8 { diff --git a/crates/asap-aware-mapping/src/accuracy/estimators/kll.rs b/crates/asap-aware-mapping/src/accuracy/estimators/kll.rs index ca84f5532..c0f117013 100644 --- a/crates/asap-aware-mapping/src/accuracy/estimators/kll.rs +++ b/crates/asap-aware-mapping/src/accuracy/estimators/kll.rs @@ -150,6 +150,36 @@ pub(crate) fn size_params(eps: f64, delta: f64) -> SketchParams { } } +/// `params` resized to fail with probability at most `delta` at no larger +/// rank error: `k` for its current error at `delta`'s scale, never smaller. +/// Unchanged when it already fails no more often. +pub(crate) fn with_failure_probability(params: &SketchParams, delta: f64) -> SketchParams { + let SketchParams::Kll { + k, + failure_probability, + } = params + else { + return params.clone(); + }; + let current = failure_probability.unwrap_or(KLL_CALIBRATED_FAILURE_PROBABILITY); + let Some(scale) = delta_scale(current) else { + return params.clone(); + }; + if delta >= current { + return params.clone(); + } + match size_params(kll_rank_error_99(*k) * scale, delta) { + SketchParams::Kll { + k: resized, + failure_probability, + } => SketchParams::Kll { + k: resized.max(*k), + failure_probability, + }, + other => other, + } +} + #[cfg(test)] mod tests { use super::*; @@ -178,7 +208,9 @@ mod tests { assert_eq!(g.spec.failure_probability.evaluate(), Some(0.01)); assert!(DefaultAccuracyModel.satisfies( &g, - &AccuracyTarget::epsilon_delta(0.01, 0.01).with_metric(ErrorMetric::Rank) + &AccuracyTarget::epsilon_delta(0.01, 0.01) + .with_metric(ErrorMetric::Rank) + .with_scope(asap_types::types::AccuracyScope::PerValue) )); assert_eq!(g.approximate_layer_count(), 1); assert!(g.provenance.iter().any(|source| matches!( @@ -209,7 +241,9 @@ mod tests { .unwrap(); assert!(DefaultAccuracyModel.satisfies( &g, - &AccuracyTarget::epsilon_delta(0.01, 0.001).with_metric(ErrorMetric::Rank) + &AccuracyTarget::epsilon_delta(0.01, 0.001) + .with_metric(ErrorMetric::Rank) + .with_scope(asap_types::types::AccuracyScope::PerValue) )); assert_eq!(g.spec.failure_probability.evaluate(), Some(0.001)); } diff --git a/crates/asap-aware-mapping/src/accuracy/estimators/mod.rs b/crates/asap-aware-mapping/src/accuracy/estimators/mod.rs index 3f6dfcec5..67aca9271 100644 --- a/crates/asap-aware-mapping/src/accuracy/estimators/mod.rs +++ b/crates/asap-aware-mapping/src/accuracy/estimators/mod.rs @@ -70,6 +70,52 @@ pub(super) fn local_guarantee( } } } +/// `params` with the parameters its per-value failure probability depends +/// on enlarged, never reduced, so that each value it answers fails with +/// probability at most `delta` and its error bound is unchanged or smaller +/// (#669b): Count-Min's depth, Count Sketch's depth (the binomial tail of +/// its median), KLL's failure probability (its δ scale) and a distinct +/// count UnivMon's heap. Other parameters are returned as they are: +/// DDSketch never fails; classic HLL's failure probability comes from its +/// source's distinct bound ([`hll::precision_for`]); KMV and Theta are +/// calibrated at a fixed 99%; an L2 or entropy UnivMon is not enlarged. +pub(crate) fn with_failure_probability(params: &SketchParams, delta: f64) -> SketchParams { + if !(delta > 0.0 && delta < 1.0) { + return params.clone(); + } + match params { + SketchParams::Cms { width, depth } => SketchParams::Cms { + width: *width, + depth: (*depth).max(cms::cms_depth(delta)), + }, + SketchParams::CmsWithHeap { + width, + depth, + heap_size, + } => SketchParams::CmsWithHeap { + width: *width, + depth: (*depth).max(cms::cms_depth(delta)), + heap_size: *heap_size, + }, + SketchParams::CountSketch { width, depth } => SketchParams::CountSketch { + width: *width, + depth: (*depth).max(count_sketch::count_sketch_depth(*width, delta)), + }, + SketchParams::CountSketchWithHeap { + width, + depth, + heap_size, + } => SketchParams::CountSketchWithHeap { + width: *width, + depth: (*depth).max(count_sketch::count_sketch_depth(*width, delta)), + heap_size: *heap_size, + }, + SketchParams::Kll { .. } => kll::with_failure_probability(params, delta), + SketchParams::UnivMon { .. } => univmon::with_failure_probability(params, delta), + _ => params.clone(), + } +} + pub(crate) fn size_params( kind: SketchAlgorithm, intent: &AggIntent, @@ -405,3 +451,110 @@ mod shared_size_tests { ); } } + +#[cfg(test)] +mod tests { + use super::*; + use asap_types::pre_asap::agg_intent::{default_cardinality, default_quantile}; + + fn delta_and_error(params: &SketchParams, query: &SketchStatistic) -> (f64, f64) { + let algorithm = match params { + SketchParams::Cms { .. } => SketchAlgorithm::Cms, + SketchParams::CountSketch { .. } => SketchAlgorithm::CountSketch, + SketchParams::Kll { .. } => SketchAlgorithm::Kll, + other => panic!("{other:?}"), + }; + let g = sketch_guarantee(&algorithm, params, query).unwrap(); + ( + g.spec.failure_probability.evaluate().unwrap(), + g.spec.error.evaluate().unwrap(), + ) + } + + /// Enlarging for a smaller failure probability deepens Count-Min and + /// Count Sketch and grows KLL until each meets it, at no larger error; + /// it never shrinks a sketch that already meets it. (Count Sketch's + /// depth is capped at 255 rows, which a non-power-of-two width reaches + /// before 1e-6.) + #[test] + fn enlarging_meets_the_failure_probability_at_no_larger_error() { + let point = SketchStatistic::PointCount { + key: asap_types::pre_asap::expr_ir::ColumnRef::SampleValue, + value: None, + }; + let quantile = SketchStatistic::Quantile { q: 0.99 }; + for (params, query, delta) in [ + ( + size_params(SketchAlgorithm::Cms, &default_cardinality(), 0.01, 0.01), + &point, + 1e-6, + ), + ( + size_params( + SketchAlgorithm::CountSketch, + &default_cardinality(), + 0.01, + 0.01, + ), + &point, + 1e-3, + ), + ( + size_params(SketchAlgorithm::Kll, &default_quantile(0.99), 0.01, 0.01), + &quantile, + 1e-6, + ), + ] { + let (_, error) = delta_and_error(¶ms, query); + let enlarged = with_failure_probability(¶ms, delta); + let (new_delta, new_error) = delta_and_error(&enlarged, query); + assert!(new_delta <= delta, "{enlarged:?}"); + assert!(new_error <= error * (1.0 + 1e-9), "{enlarged:?}"); + assert_ne!(enlarged, params); + assert_eq!(with_failure_probability(&enlarged, 0.01), enlarged); + } + let ddsketch = SketchParams::DDSketch { alpha: 0.01 }; + assert_eq!(with_failure_probability(&ddsketch, 1e-6), ddsketch); + } + + /// A distinct-count UnivMon grows its heap for a smaller failure + /// probability at no larger relative error. + #[test] + fn a_distinct_count_univmon_grows_its_heap() { + let params = size_params(SketchAlgorithm::UnivMon, &default_cardinality(), 0.05, 0.01); + let enlarged = with_failure_probability(¶ms, 1e-4); + let error = |p: &SketchParams| { + univmon::distinct_guarantee(p, "m") + .unwrap() + .spec + .error + .evaluate() + .unwrap() + }; + let ( + SketchParams::UnivMon { heap_size: k, .. }, + SketchParams::UnivMon { + heap_size: grown, + distinct_failure_probability, + .. + }, + ) = (¶ms, &enlarged) + else { + panic!("{enlarged:?}") + }; + assert!(grown > k); + assert_eq!(*distinct_failure_probability, Some(1e-4)); + assert!(error(&enlarged) <= error(¶ms) * (1.0 + 1e-9)); + } + + /// Classic HLL's precision grows until its failure probability over the + /// source's distinct bound meets the target, up to 18. + #[test] + fn classic_hll_precision_grows_for_its_failure_probability() { + let p = hll::precision_for(4, 64, 0.01); + assert!(p > 4); + assert!(hll::classic_failure_probability(p, 64) <= 0.01); + assert!(hll::classic_failure_probability(p - 1, 64) > 0.01); + assert_eq!(hll::precision_for(12, 4096, 1e-30), 12); + } +} diff --git a/crates/asap-aware-mapping/src/accuracy/estimators/univmon.rs b/crates/asap-aware-mapping/src/accuracy/estimators/univmon.rs index 39537cb34..8dffc8a62 100644 --- a/crates/asap-aware-mapping/src/accuracy/estimators/univmon.rs +++ b/crates/asap-aware-mapping/src/accuracy/estimators/univmon.rs @@ -289,6 +289,35 @@ pub(crate) fn distinct_guarantee(params: &SketchParams, source: &str) -> Option< )) } +/// A distinct-count `params` resized to fail with probability at most +/// `delta` at no larger relative error: the heap for its current error at +/// `delta` ([`distinct_heap`]), never smaller. Unchanged for a shape not +/// sized for a distinct count, or one that already fails no more often. +pub(crate) fn with_failure_probability(params: &SketchParams, delta: f64) -> SketchParams { + let SketchParams::UnivMon { + heap_size, + sketch_rows, + sketch_cols, + layers, + distinct_failure_probability: Some(current), + } = params + else { + return params.clone(); + }; + let chebyshev = current - incomplete_top_layer_bound(*heap_size); + if delta >= *current || chebyshev <= 0.0 { + return params.clone(); + } + let relative_error = (DISTINCT_VARIANCE / (f64::from(*heap_size) * chebyshev)).sqrt(); + SketchParams::UnivMon { + heap_size: distinct_heap(relative_error, delta).max(*heap_size), + sketch_rows: *sketch_rows, + sketch_cols: *sketch_cols, + layers: *layers, + distinct_failure_probability: Some(delta), + } +} + /// Size a UnivMon for one readout's `(eps, delta)`: /// - L2: the smallest rows and columns meeting the target ([`l2_shape`]), /// one layer (L2 reads only layer 0) and a one-key heap; diff --git a/crates/asap-aware-mapping/src/accuracy/mod.rs b/crates/asap-aware-mapping/src/accuracy/mod.rs index 3e8de302c..0d1916dd6 100644 --- a/crates/asap-aware-mapping/src/accuracy/mod.rs +++ b/crates/asap-aware-mapping/src/accuracy/mod.rs @@ -23,7 +23,7 @@ use asap_types::post_asap::{ AccuracyError, AccuracyGuarantee, BoundExpr, CompositionOperator, ErrorMetric, FieldDataType, GuaranteeSource, ProbabilityExpr, SketchAlgorithm, SketchParams, SketchStatistic, }; -use asap_types::types::AccuracyTarget; +use asap_types::types::{AccuracyScope, AccuracyTarget}; use crate::exact_composition::ExactOperation; @@ -65,7 +65,8 @@ pub trait AccuracyModel { /// Whether `guarantee` meets `target`: a bound in the target's metric /// (or one with a sound direct comparison) within its ε, and within its - /// δ when it states one. A bound in another metric does not satisfy, + /// δ when it states one, over the values the target's scope covers (a + /// per-value δ does not meet a whole-result target). A bound in another metric does not satisfy, /// however small. Unknown required dimensions fail; selection /// separately excludes missing accuracy evidence. fn satisfies(&self, guarantee: &AccuracyGuarantee, target: &AccuracyTarget) -> bool; @@ -119,6 +120,7 @@ impl AccuracyModel for DefaultAccuracyModel { return guarantee.is_exact(); } metric_meets(guarantee.spec.metric, target.metric) + && scope_meets(guarantee, target) && within(guarantee.spec.error.evaluate(), target.error) && target .failure_probability @@ -133,21 +135,73 @@ impl AccuracyModel for DefaultAccuracyModel { /// `‖f‖₂ ≤ ‖f‖₁`. Any other pair needs a conversion backed by evidence /// (Absolute ↔ Relative needs a bound on the true value's magnitude; a rank /// error has no distribution-free value error), which the default model -/// does not make: it does not satisfy. The scope is not compared yet: a -/// per-value guarantee is checked against a whole-result target directly -/// until the union bound over the output values (#669b). +/// does not make: it does not satisfy. The scope is compared separately +/// ([`scope_meets`]). fn metric_meets(guarantee: ErrorMetric, target: ErrorMetric) -> bool { guarantee == target || (guarantee == ErrorMetric::ABSOLUTE_STREAM_L2 && target == ErrorMetric::ABSOLUTE_STREAM_L1) } -/// `target` restated in `guarantee`'s metric: the check of one summary -/// against the target it was sized for. A summary is sized in its own -/// native terms (#690 §4) and an aggregate intent's metric is not checked: -/// only a query root's metric is, by plan selection. +/// Whether `guarantee`'s failure probability covers the values `target`'s +/// does. A whole-result guarantee covers a per-value target (each value is +/// in the result), but a per-value one does not cover a whole-result +/// target: it must first be stated over the result's values +/// ([`whole_result_guarantee`]). The scope bounds only the failure +/// probability, so it does not matter for a target without a δ requirement +/// or for a guarantee that never fails. +fn scope_meets(guarantee: &AccuracyGuarantee, target: &AccuracyTarget) -> bool { + target.scope == AccuracyScope::PerValue + || guarantee.spec.scope == AccuracyScope::WholeResult + || target.failure_probability.is_none() + || guarantee.spec.failure_probability.is_zero() +} + +/// `target` restated in `guarantee`'s metric, per value: the check of one +/// summary against the target it was sized for. A summary is sized in its +/// own native terms (#690 §4) and for each value it answers; an aggregate +/// intent's metric and scope are not checked: only a query root's are, by +/// plan selection. pub fn native_target(target: &AccuracyTarget, guarantee: &AccuracyGuarantee) -> AccuracyTarget { - target.clone().with_metric(guarantee.spec.metric) + target + .clone() + .with_metric(guarantee.spec.metric) + .with_scope(AccuracyScope::PerValue) +} + +/// The statistic an unbound number of output values per evaluation is left +/// as ([`whole_result_guarantee`]); nothing binds it, so the failure +/// probability stays unknown. +pub const OUTPUT_VALUES: &str = "output_values_per_evaluation"; + +/// `guarantee` stated over the whole result of one evaluation, of +/// `output_values` values (`G`): a per-value failure probability `δ` becomes +/// `G · δ` by the union bound over the values, whatever their dependence. +/// The error is unchanged. A guarantee that never fails (`δ = 0`) holds for +/// every value at once whatever `G`; an unbound `G` (`None`) leaves the +/// failure probability symbolic in [`OUTPUT_VALUES`]. A whole-result +/// guarantee is returned as it is. +pub fn whole_result_guarantee( + guarantee: &AccuracyGuarantee, + output_values: Option, +) -> AccuracyGuarantee { + let mut whole = guarantee.clone(); + if guarantee.spec.scope == AccuracyScope::WholeResult { + return whole; + } + whole.spec.scope = AccuracyScope::WholeResult; + if !guarantee.spec.failure_probability.is_zero() { + whole.spec.failure_probability = ProbabilityExpr::Scaled { + count: output_values.map_or_else( + || BoundExpr::Unknown { + statistic: OUTPUT_VALUES.into(), + }, + |value| BoundExpr::Constant { value }, + ), + inner: Box::new(guarantee.spec.failure_probability.clone()), + }; + } + whole } pub(crate) use estimators::topk_capacity; @@ -155,7 +209,6 @@ pub(crate) use estimators::topk_capacity; #[cfg(test)] mod tests { use super::*; - use asap_types::types::AccuracyScope; fn abs(bound: f64, delta: f64) -> AccuracyGuarantee { AccuracyGuarantee::new( ErrorMetric::ABSOLUTE, @@ -194,7 +247,11 @@ mod tests { #[test] fn satisfies_requires_the_targets_metric() { let model = DefaultAccuracyModel; - let target = |metric| AccuracyTarget::epsilon_delta(0.5, 0.5).with_metric(metric); + let target = |metric| { + AccuracyTarget::epsilon_delta(0.5, 0.5) + .with_metric(metric) + .with_scope(AccuracyScope::PerValue) + }; let (l1, l2) = ( ErrorMetric::ABSOLUTE_STREAM_L1, ErrorMetric::ABSOLUTE_STREAM_L2, @@ -234,16 +291,49 @@ mod tests { assert!(!DefaultAccuracyModel.satisfies(&g, &AccuracyTarget::epsilon_delta(0.01, 0.5))); } - /// Pins the current scope behavior: a per-value guarantee is compared - /// with a whole-result target directly. The union bound over the output - /// values (δ_whole = G · δ_per) arrives with #669b, which changes this - /// test. + /// A per-value guarantee does not meet a whole-result δ target + /// directly: over `G` output values it holds for all of them with + /// δ_whole = G · δ_per ([`whole_result_guarantee`]), and an unbound `G` + /// leaves it unmet. A whole-result guarantee meets a per-value target + /// directly. #[test] - fn per_value_guarantees_meet_whole_result_targets_directly_until_669b() { - let g = guarantee(ErrorMetric::ABSOLUTE, 0.01, 0.01); + fn per_value_guarantees_meet_whole_result_targets_over_g_values() { + let model = DefaultAccuracyModel; + let g = guarantee(ErrorMetric::ABSOLUTE, 0.01, 0.001); assert_eq!(g.spec.scope, AccuracyScope::PerValue); - let target = AccuracyTarget::epsilon_delta(0.01, 0.01); - assert_eq!(target.scope, AccuracyScope::WholeResult); - assert!(DefaultAccuracyModel.satisfies(&g, &target)); + let whole = AccuracyTarget::epsilon_delta(0.01, 0.01); + assert_eq!(whole.scope, AccuracyScope::WholeResult); + assert!(!model.satisfies(&g, &whole)); + let per_value = whole.clone().with_scope(AccuracyScope::PerValue); + assert!(model.satisfies(&g, &per_value)); + let over = |values| whole_result_guarantee(&g, values); + assert_eq!(over(Some(10.0)).spec.scope, AccuracyScope::WholeResult); + assert_eq!( + over(Some(10.0)).spec.failure_probability.evaluate(), + Some(0.01) + ); + assert!(model.satisfies(&over(Some(10.0)), &whole)); + assert!(!model.satisfies(&over(Some(11.0)), &whole)); + assert!(!model.satisfies(&over(None), &whole)); + assert!(model.satisfies(&over(Some(10.0)), &per_value)); + } + + /// The scope bounds only the failure probability: a target without a δ + /// requirement, or a guarantee that never fails, meets a whole-result + /// target whatever the number of values. + #[test] + fn scope_matters_only_with_a_failure_probability() { + let model = DefaultAccuracyModel; + let g = guarantee(ErrorMetric::ABSOLUTE, 0.01, 0.001); + assert!(model.satisfies(&g, &AccuracyTarget::epsilon(0.01))); + let never_fails = AccuracyGuarantee::new( + ErrorMetric::Relative, + BoundExpr::Constant { value: 0.01 }, + ProbabilityExpr::Zero, + vec![], + ); + let whole = AccuracyTarget::epsilon_delta(0.01, 0.01).with_metric(ErrorMetric::Relative); + assert!(model.satisfies(&never_fails, &whole)); + assert!(model.satisfies(&whole_result_guarantee(&never_fails, None), &whole)); } } diff --git a/crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs b/crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs index bb1c2fa20..b7c533af6 100644 --- a/crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs +++ b/crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs @@ -417,8 +417,13 @@ pub fn run_stage2_and_stage3( roots: candidate.roots, }); } - let selection = resolve_targets(&per_computation.roots, targets) - .and_then(|targets| select_plan(&physical, &targets, recurrence, data, models)); + let (resolved, selection) = match resolve_targets(&per_computation.roots, targets) { + Ok(targets) => { + let selection = select_plan(&physical, &targets, recurrence, data, models); + (Some(targets), selection) + } + Err(error) => (None, Err(error)), + }; let selected_logical = selection.as_ref().ok().and_then(|selection| { let from = &physical .iter() @@ -430,8 +435,11 @@ pub fn run_stage2_and_stage3( .roots .as_ref() .ok()?; + // As Stage 3 sized it for its queries' failure probabilities. + let sized = + crate::stage3_plan_selection::sized_roots(roots, resolved.as_deref()?, data).ok()?; Some(crate::stage3_plan_selection::with_confirmed_weights( - roots, data, + &sized, data, )) }); Stage1ToStage3Run { diff --git a/crates/asap-aware-mapping/src/per_computation_candidates.rs b/crates/asap-aware-mapping/src/per_computation_candidates.rs index f63e98d18..4f9f4787c 100644 --- a/crates/asap-aware-mapping/src/per_computation_candidates.rs +++ b/crates/asap-aware-mapping/src/per_computation_candidates.rs @@ -834,7 +834,7 @@ pub(crate) fn approximate_inputs(root: &QueryRoot) -> Vec> { let mut seen = HashSet::new(); operators .into_iter() - .flat_map(|node| OperatorNode::reachable(node)) + .flat_map(OperatorNode::reachable) .filter(|node| { matches!( node.asap(), @@ -858,6 +858,96 @@ pub(crate) fn root_guarantee(root: &QueryRoot) -> Option { } } +/// `roots` with every sketch summary state rebuilt with the parameters +/// `resize` gives it (same algorithm; `None` keeps them), and every node +/// above one rebuilt over its new inputs with its guarantee composed again +/// ([`composed_guarantee`]; a node that had none keeps none). Timing, +/// materialization and sharing are kept. Plan selection enlarges summaries +/// for a failure probability this way. +pub(crate) fn with_resized_summaries( + roots: &[QueryRoot], + resize: &dyn Fn(&OperatorNode, &SketchParams) -> Option, +) -> Result, SchemaDerivationError> { + fn walk( + node: &Rc, + resize: &dyn Fn(&OperatorNode, &SketchParams) -> Option, + memo: &mut Memo, + ) -> Result, SchemaDerivationError> { + if let Some(done) = memo.get(&Rc::as_ptr(node)) { + return Ok(done.clone()); + } + for child in node.children() { + walk(child, resize, memo)?; + } + let mut changed = node + .children() + .iter() + .any(|child| !Rc::ptr_eq(child, &memo[&Rc::as_ptr(child)])); + let mut operator = node.operator.clone(); + if let Operator::ASAP(ASAPOp::SummaryAgg { + family: FieldDataType::Sketch(kind, _), + .. + }) = &mut operator + { + if let Some(params) = resize(node, kind.params()).filter(|p| p != kind.params()) { + *kind = SketchKind::new(kind.algorithm().clone(), params); + changed = true; + } + } + let rebuilt = if changed { + let template = OperatorNode { + operator, + ..(**node).clone() + }; + let mut rebuilt = + template.with_new_children(|child| memo[&Rc::as_ptr(child)].clone())?; + rebuilt.guarantee = node + .guarantee + .as_ref() + .and_then(|_| composed_guarantee(&rebuilt)); + Rc::new( + rebuilt + .with_timing(node.timing) + .with_materialization(node.materialization), + ) + } else { + node.clone() + }; + memo.insert(Rc::as_ptr(node), rebuilt.clone()); + Ok(rebuilt) + } + let mut memo = HashMap::new(); + roots + .iter() + .map(|root| { + Ok(match root { + QueryRoot::Operator(node) => QueryRoot::Operator(walk(node, resize, &mut memo)?), + QueryRoot::Scalar(expr) => { + let mut rebuilt = HashMap::new(); + for node in expr.operator_refs() { + rebuilt.insert(Rc::as_ptr(node), walk(node, resize, &mut memo)?); + } + QueryRoot::Scalar( + expr.map_operator_refs(&mut |node| rebuilt[&Rc::as_ptr(node)].clone()), + ) + } + }) + }) + .collect() +} + +/// The one source a summary state counts the distinct items of its own +/// rows of, as [`source_distinct_guarantee`] reads it: the source whose +/// distinct bound gives its failure probability. `None` for a state over +/// computed rows or several sources. +pub(crate) fn summary_source(state: &OperatorNode) -> Option { + let Some(ASAPOp::SummaryAgg { child, .. }) = state.asap() else { + return None; + }; + source_scan(child)?; + source(child) +} + /// Every root of `inventory` with the chosen candidates substituted, each /// summary sized as `sizing` says, and every node with its guarantee. fn compose_roots( diff --git a/crates/asap-aware-mapping/src/replacement.rs b/crates/asap-aware-mapping/src/replacement.rs index 83ad667f4..f73ed16bc 100644 --- a/crates/asap-aware-mapping/src/replacement.rs +++ b/crates/asap-aware-mapping/src/replacement.rs @@ -1630,6 +1630,15 @@ impl Proposals { } } +/// `target` over each value separately: this search sizes and checks every +/// node per value; a target over a whole result is checked only at a query +/// root, by plan selection. +fn per_value(target: &AccuracyTarget) -> AccuracyTarget { + target + .clone() + .with_scope(asap_types::types::AccuracyScope::PerValue) +} + /// The `child` of a [`bindable_intent`]-shaped `Aggregate`. fn aggregate_child(node: &OperatorNode) -> Option<&Rc> { match node.non_asap() { @@ -3830,7 +3839,7 @@ impl CandidateLogicalASAPDAGs { || !requirements.iter().all(|target| { plan.guarantee .as_ref() - .is_some_and(|g| accuracy.satisfies(g, target)) + .is_some_and(|g| accuracy.satisfies(g, &per_value(target))) }) { continue; @@ -6127,7 +6136,7 @@ pub fn search_workload_with_targets<'s, Id>( Replacement::SubDAG(_) if candidate.is_logical_rewrite() => true, Replacement::SubDAG(node) => node.guarantee.as_ref().map_or_else( || !target.is_exact(), - |g| accuracy_model.satisfies(&g.optimistic_floor(), &target), + |g| accuracy_model.satisfies(&g.optimistic_floor(), &per_value(&target)), ), // A composition's guarantee depends on the concrete child; // prepare_compositions checks those pairs after all roots. diff --git a/crates/asap-aware-mapping/src/stage3_plan_selection.rs b/crates/asap-aware-mapping/src/stage3_plan_selection.rs index aa8f56c8b..922cf37b9 100644 --- a/crates/asap-aware-mapping/src/stage3_plan_selection.rs +++ b/crates/asap-aware-mapping/src/stage3_plan_selection.rs @@ -68,19 +68,22 @@ use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; use std::rc::Rc; use asap_types::ir::operator_properties::BinaryOpKind; +use asap_types::ir::physical_export::compile_physical_asap_workload_with_node_ids; use asap_types::ir::physical_export::{ PhysicalASAPDAG, PhysicalASAPNodeId, PhysicalASAPOperatorPayload as Payload, }; use asap_types::ir::{ - ASAPOp, BinaryOperator, NonASAPOp, Operator, OperatorNode, QueryRoot, ScalarExpr, + ASAPOp, BinaryOperator, NonASAPOp, Operator, OperatorNode, OperatorResultKind, QueryRoot, + ScalarExpr, SchemaDerivationError, }; use asap_types::post_asap::sketch::KLL_MAX_K; +use asap_types::post_asap::ExecutionDataStateError; use asap_types::post_asap::{ AccuracyGuarantee, CompositionOperator, ErrorMetric, ExecutionTiming, FieldDataType, Materialization, SketchAlgorithm, SketchParams, SketchStatistic, WeightDomain, }; use asap_types::pre_asap::{AggIntent, DataType, Reduction, ScalarValue, Schema}; -use asap_types::types::AccuracyTarget; +use asap_types::types::{AccuracyScope, AccuracyTarget}; use asap_types::workload::{DataWorkload, MetricType, QueryRecurrence, RepeatedDemand}; use thiserror::Error; @@ -88,14 +91,15 @@ use crate::accuracy::composition::{ parse_samples_sign_statistic, parse_source_norm_statistic, DIVISION_OPERAND_DOMAINS, INPUT_ROW_COUNT, PRODUCT_OPERAND_DOMAINS, STREAM_L1_NORM, }; +use crate::accuracy::estimators::with_failure_probability; use crate::accuracy::estimators::{hll, univmon}; -use crate::accuracy::{OperandDomain, PropagationStats}; +use crate::accuracy::{whole_result_guarantee, OperandDomain, PropagationStats}; use crate::analytical_cost::{ estimate_operator, AnalyticalCostError, PhysicalOperator, ResourceCalibration, ResourceEstimate, }; use crate::per_computation_candidates::{ - operand_domain_statistic, operand_guarantee, operand_of, project_column_guarantee, - scalar_guarantee, + approximate_inputs, operand_domain_statistic, operand_guarantee, operand_of, + project_column_guarantee, root_guarantee, summary_source, with_resized_summaries, }; use crate::physical_operator_statistics::{ EdgeStatistics, OperatorStatistics, PartitionStatistics, UnaryEdgeStatistics, @@ -194,6 +198,13 @@ pub enum SelectionError { }, #[error("no valid candidate: {0:?}")] NoValidCandidate(Vec), + /// A candidate's summaries could not be rebuilt with the parameters its + /// queries' failure probabilities need. + #[error("candidate {candidate}: sizing for the failure probability: {error}")] + Sizing { + candidate: String, + error: SizingError, + }, /// A query's requested accuracy cannot be resolved for its root, e.g. a /// quantile query's target names no metric. #[error("q{query}: {error}")] @@ -216,7 +227,7 @@ pub fn select_plan( ) -> Result { let mut costs = BTreeMap::new(); let mut rejected = Vec::new(); - let mut best: Option<(&str, f64)> = None; + let mut best: Option<(String, f64, Vec)> = None; for candidate in cands { if candidate.roots.len() != targets.len() || candidate.roots.len() != recurrence.len() { return Err(SelectionError::TargetCount { @@ -226,6 +237,13 @@ pub fn select_plan( recurrences: recurrence.len(), }); } + let candidate = + &sized_for_failure_probability(candidate, targets, data).map_err(|error| { + SelectionError::Sizing { + candidate: candidate.id.clone(), + error, + } + })?; if let Some(reason) = accuracy_violation(candidate, targets, data, &models) { rejected.push(Rejection { id: candidate.id.clone(), @@ -243,12 +261,15 @@ pub fn select_plan( error, } })?; - if best.is_none_or(|(_, total)| cheaper(cost.total, total)) { - best = Some((&candidate.id, cost.total)); + if best + .as_ref() + .is_none_or(|(_, total, _)| cheaper(cost.total, *total)) + { + best = Some((candidate.id.clone(), cost.total, candidate.roots.clone())); } costs.insert(candidate.id.clone(), cost); } - let Some((selected, best_total)) = best else { + let Some((selected, best_total, selected_roots)) = best else { return Err(SelectionError::NoValidCandidate(rejected)); }; for candidate in cands { @@ -269,13 +290,9 @@ pub fn select_plan( }); } } - let plan = cands - .iter() - .find(|c| c.id == selected) - .map(|c| with_confirmed_weights(&c.roots, data)) - .unwrap_or_default(); + let plan = with_confirmed_weights(&selected_roots, data); Ok(Selection { - selected: selected.to_string(), + selected, plan, costs, rejected, @@ -289,6 +306,126 @@ fn cheaper(cost: f64, best: f64) -> bool { cost < best - best.abs() * 1e-9 } +/// Why a candidate's summaries could not be rebuilt for a failure +/// probability ([`sized_for_failure_probability`]). +#[derive(Debug, Error)] +pub enum SizingError { + #[error(transparent)] + Structure(#[from] SchemaDerivationError), + #[error(transparent)] + Export(#[from] ExecutionDataStateError), +} + +/// Without a planning time, only evidence that cannot expire is fresh. +const ANY_TIME: u64 = u64::MAX; + +/// How many values one evaluation of `root` returns at most (`G`), from +/// `data`: one for a scalar root, at most the input cardinality (the +/// series) for a PromQL instant vector, whose groups and series never +/// outnumber its input series. `None` when unbound: a SQL result, until a +/// bound on the groups of its source is an input, and a range vector. +fn output_values(root: &QueryRoot, data: &DataWorkload) -> Option { + match root { + QueryRoot::Scalar(_) => Some(1.0), + QueryRoot::Operator(node) if node.result_kind == OperatorResultKind::InstantVector => data + .input_cardinality + .value_at(ANY_TIME) + .map(|series| (*series).max(1) as f64), + QueryRoot::Operator(_) => None, + } +} + +/// `candidate` with each summary's per-value failure probability enlarged +/// for its queries' targets (#669b), before pricing, so that cost reflects +/// the deeper sketch. A query with a target with failure probability `δ`, +/// over `n` approximate inputs ([`approximate_inputs`]) and `G` values per +/// evaluation ([`output_values`]; 1 for a per-value target), needs each +/// input to fail with probability at most `δ / (n · G)`: `n · G` such +/// failures union-bound to `δ` over the whole result. A summary several +/// queries read takes the smallest. The parameters that depend on it are +/// enlarged ([`with_failure_probability`]; classic HLL's precision from its +/// source's distinct bound), and the guarantees above are composed again. +/// A query whose `G` is unbound enlarges nothing: its whole-result target +/// stays unmet. Stage 1 sizes for ε and reads no data workload; only this +/// stage binds `G`. +fn sized_for_failure_probability( + candidate: &PhysicalCandidate, + targets: &[Option], + data: &DataWorkload, +) -> Result { + let roots = sized_roots(&candidate.roots, targets, data)?; + if roots + .iter() + .zip(&candidate.roots) + .all(|(sized, root)| sized == root) + { + return Ok(candidate.clone()); + } + let mut dag = compile_physical_asap_workload_with_node_ids(&roots)?.dag; + // The DAG has the same nodes, so each keeps the placement Stage 2 gave it. + if dag.nodes.len() == candidate.dag.nodes.len() { + for (node, placed) in dag.nodes.iter_mut().zip(&candidate.dag.nodes) { + if node.id == placed.id { + node.output_state = placed.output_state; + node.materialization = placed.materialization; + } + } + } + Ok(PhysicalCandidate { + roots, + dag, + ..candidate.clone() + }) +} + +/// `roots` with their summaries enlarged as [`sized_for_failure_probability`] +/// says: the plan plan selection prices and runs, for the logical roots of +/// the selected candidate too. +pub fn sized_roots( + roots: &[QueryRoot], + targets: &[Option], + data: &DataWorkload, +) -> Result, SchemaDerivationError> { + let mut required: HashMap<*const OperatorNode, f64> = HashMap::new(); + for (root, target) in roots.iter().zip(targets) { + let Some((target, delta)) = target + .as_ref() + .and_then(|t| t.failure_probability.map(|delta| (t, delta))) + .filter(|(t, delta)| !t.is_exact() && *delta > 0.0) + else { + continue; + }; + let values = match target.scope { + AccuracyScope::PerValue => Some(1.0), + AccuracyScope::WholeResult => output_values(root, data), + }; + let Some(values) = values else { continue }; + let inputs = approximate_inputs(root); + let share = delta / (inputs.len() as f64 * values); + for state in &inputs { + let entry = required.entry(Rc::as_ptr(state)).or_insert(share); + *entry = entry.min(share); + } + } + if required.is_empty() { + return Ok(roots.to_vec()); + } + let resize = |state: &OperatorNode, params: &SketchParams| { + let delta = *required.get(&std::ptr::from_ref(state))?; + Some(match params { + SketchParams::Hll { precision } => { + let source = summary_source(state)?; + let bound = data.max_distinct_per_evaluation.value_at(ANY_TIME)?; + SketchParams::Hll { + precision: hll::precision_for(*precision, *bound.get(&source)?, delta), + } + } + params => with_failure_probability(params, delta), + }) + }; + with_resized_summaries(roots, &resize) +} + /// The first query whose output is not shown to meet its target, as a /// reason. An output meets its target when its theoretical root guarantee /// (composed in Stage 1 with the built-in model) satisfies it, or when the @@ -337,18 +474,40 @@ fn accuracy_violation( )); } } - let guarantee = bound_root_guarantee(root, data, models); + // A guarantee per value is stated over the whole result's `G` + // values for a whole-result target. + let values = output_values(root, data); + let over_scope = |g: AccuracyGuarantee| match target.scope { + AccuracyScope::WholeResult => whole_result_guarantee(&g, values), + AccuracyScope::PerValue => g, + }; + let guarantee = bound_root_guarantee(root, data, models).map(over_scope); let meets = |g: &AccuracyGuarantee| models.accuracy.satisfies(g, target); if guarantee.as_ref().is_some_and(meets) || models .evidence .output_guarantee(root) + .map(over_scope) .as_ref() .is_some_and(meets) { continue; } return Some(match guarantee { + Some(g) + if values.is_none() + && target.scope == AccuracyScope::WholeResult + && target.failure_probability.is_some() + && g.spec.failure_probability.evaluate().is_none() => + { + format!( + "q{query}: the output's guarantee holds for each value, and the number \ + of values one evaluation returns is not bound (no declared input \ + cardinality for a PromQL vector; no bound on a SQL result's groups yet), \ + so it does not meet whole-result target {target:?}, and the deployment \ + has no evidence that it meets it" + ) + } Some(g) => format!( "q{query}: the output guarantees a {:?} bound {:?}, failure probability {:?}, \ which misses target {target:?}, and the deployment has no evidence that it \ @@ -432,16 +591,6 @@ fn root_operators(root: &QueryRoot) -> Vec<&Rc> { } } -/// The theoretical guarantee of a query's output, as Stage 1 composed it: -/// the root node's own, or a scalar root's composed over the operators it -/// reads ([`scalar_guarantee`]). -fn root_guarantee(root: &QueryRoot) -> Option { - match root { - QueryRoot::Operator(node) => node.guarantee.clone(), - QueryRoot::Scalar(expr) => scalar_guarantee(expr, &|_| None), - } -} - /// [`root_guarantee`] with the statistics Stage 1 left symbolic bound: /// - the input row count, from the data workload's input cardinality; /// - each source's stream norm ([`source_norm_statistic`]), from the data @@ -492,8 +641,6 @@ fn bind_statistics( /// bound, from evidence that holds at any planning time, or the sign of /// one metric's samples, from its declared type. fn workload_statistic(statistic: &str, data: &DataWorkload) -> Option { - // Without a planning time, only evidence that cannot expire is fresh. - const ANY_TIME: u64 = u64::MAX; // The sign of one metric's samples: proven (1) when the metric is // declared a counter; no type is inferred from its name. if let Some(metric) = parse_samples_sign_statistic(statistic) { @@ -1851,6 +1998,116 @@ mod tests { .any(|n| matches!(n.payload, Payload::ASAP(ASAPOp::SummaryEstimate { .. }))) } + /// The KLL candidates of `quantile_over_time(0.99, x[5m])`, per series, + /// within rank error 0.01 at δ = 0.01 over `scope`. + fn per_series_kll(scope: AccuracyScope) -> (Vec, AccuracyTarget) { + let requested = RequestedAccuracy::epsilon_delta(0.01, 0.01).with_metric(ErrorMetric::Rank); + let candidates = all_candidates("quantile_over_time(0.99, x[5m])", &requested) + .into_iter() + .filter(|c| !kll_params(&c.roots).is_empty()) + .collect::>(); + assert!(!candidates.is_empty()); + let target = requested.target_in(ErrorMetric::Rank).with_scope(scope); + (candidates, target) + } + + fn kll_params(roots: &[QueryRoot]) -> Vec { + roots + .iter() + .flat_map(root_operators) + .flat_map(OperatorNode::reachable) + .filter_map(|node| match node.asap() { + Some(ASAPOp::SummaryAgg { + family: FieldDataType::Sketch(kind, _), + .. + }) if *kind.algorithm() == SketchAlgorithm::Kll => Some(kind.params().clone()), + _ => None, + }) + .collect() + } + + /// A whole-result target over `G` = 10 000 series (the declared input + /// cardinality) needs each series' KLL to fail with probability at most + /// δ / G = 1e-6 (#669b): Stage 3 enlarges it before pricing, and the + /// selected plan runs the enlarged sketch. A per-value target keeps the + /// Stage 1 size. + #[test] + fn a_whole_result_target_enlarges_each_sketch_for_g_values() { + let (candidates, whole) = per_series_kll(AccuracyScope::WholeResult); + let select = |target: &AccuracyTarget| { + select_plan( + &candidates, + &[Some(target.clone())], + &every_10s(1), + &data(), + PlanningModels::builtin(), + ) + .unwrap() + }; + let stage1 = kll_params(&candidates[0].roots); + let enlarged = kll_params(&select(&whole).plan); + assert_eq!(enlarged.len(), 1); + let ( + SketchParams::Kll { k: k1, .. }, + SketchParams::Kll { + k, + failure_probability, + }, + ) = (&stage1[0], &enlarged[0]) + else { + panic!("{stage1:?} {enlarged:?}") + }; + assert_eq!(*failure_probability, Some(0.01 / 10_000.0)); + assert!(k > k1, "{k} vs {k1}"); + let per_value = whole.clone().with_scope(AccuracyScope::PerValue); + assert_eq!(kll_params(&select(&per_value).plan), stage1); + } + + /// Without a bound on the values one evaluation returns, a per-value + /// guarantee does not meet a whole-result target: a PromQL vector with + /// no declared input cardinality, and a SQL relation (until a bound on + /// its groups is an input). A scalar returns one value. + #[test] + fn an_unbound_number_of_values_leaves_a_whole_result_target_unmet() { + let (candidates, whole) = per_series_kll(AccuracyScope::WholeResult); + let no_cardinality = DataWorkload { + input_cardinality: Evidence::default(), + ..data() + }; + let Err(SelectionError::NoValidCandidate(rejected)) = select_plan( + &candidates, + &[Some(whole)], + &every_10s(1), + &no_cardinality, + PlanningModels::builtin(), + ) else { + panic!("the KLL candidates are invalid") + }; + assert!( + rejected[0].reason.contains("is not bound"), + "{}", + rejected[0].reason + ); + let table = crate::test_support::scan_from( + asap_types::pre_asap::Source::Table { + table_ref: "t".into(), + }, + Schema::new(vec![asap_types::pre_asap::schema::Field::plain( + "v", + DataType::Float64, + false, + )]), + ); + assert_eq!(output_values(&QueryRoot::Operator(table), &data()), None); + assert_eq!( + output_values( + &QueryRoot::Scalar(ScalarExpr::literal_f64(1.0)), + &DataWorkload::default() + ), + Some(1.0) + ); + } + /// Deployment evidence that every output was measured exact. struct EveryOutputMeasuredExact; diff --git a/crates/integration-tests/tests/planner_layering_example2.rs b/crates/integration-tests/tests/planner_layering_example2.rs index c6440f1a9..6c356740a 100644 --- a/crates/integration-tests/tests/planner_layering_example2.rs +++ b/crates/integration-tests/tests/planner_layering_example2.rs @@ -24,7 +24,7 @@ use asap_types::post_asap::{ Field, ProbabilityExpr, Schema, SketchAlgorithm, SketchParams, SketchStatistic, }; use asap_types::pre_asap::DataType; -use asap_types::types::RequestedAccuracy; +use asap_types::types::{AccuracyScope, RequestedAccuracy}; use asap_types::workload::{ AccuracyRequirement, DataArrival, DataDistribution, DataWorkload, DurationMs, LatencyRequirement, PlanningWorkload, Predictability, Query, QueryLanguage, QueryRequirements, @@ -54,14 +54,15 @@ const PROMQL_Q3: &str = "l2_over_time(flows_src_ip[1m])"; /// (ε, δ) of Q1, Q2, Q3. Q3 is the strictest. const ACCURACY: [(f64, f64); 3] = [(0.02, 0.01), (0.05, 0.01), (0.01, 0.01)]; -fn panel(query: &str, (epsilon, delta): (f64, f64)) -> RepeatingEntry { +fn panel(query: &str, (epsilon, delta): (f64, f64), scope: AccuracyScope) -> RepeatingEntry { RepeatingEntry { query: Query(query.into()), demand: RepeatedDemand::FixedInterval(RepetitionInterval(10_000)), requirements: QueryRequirements { - accuracy: AccuracyRequirement::Explicit(RequestedAccuracy::epsilon_delta( - epsilon, delta, - )), + accuracy: AccuracyRequirement::Explicit(RequestedAccuracy { + scope, + ..RequestedAccuracy::epsilon_delta(epsilon, delta) + }), response_latency: LatencyRequirement::Unspecified, }, predictability: Predictability::Predictable { known_at: None }, @@ -92,8 +93,18 @@ fn data_workload(ingestion_interval: bool) -> DataWorkload { } } +/// The SQL queries' targets cover their whole result, as the doc's do. The +/// PromQL stand-in states them per value: it reads one series, but the +/// data workload's input cardinality (the 10M source IPs) is what plan +/// selection takes as the number of values a PromQL vector returns, so a +/// whole-result target would be checked over 10M values (#669b). fn workload(language: QueryLanguage, queries: [&str; 3]) -> PlanningWorkload { let sql = matches!(language, QueryLanguage::SQL(_)); + let scope = if sql { + AccuracyScope::WholeResult + } else { + AccuracyScope::PerValue + }; PlanningWorkload { query_workload: QueryWorkload { language, @@ -102,7 +113,7 @@ fn workload(language: QueryLanguage, queries: [&str; 3]) -> PlanningWorkload { queries .iter() .zip(ACCURACY) - .map(|(q, accuracy)| panel(q, accuracy)) + .map(|(q, accuracy)| panel(q, accuracy, scope)) .collect(), ), }, @@ -615,12 +626,20 @@ fn stage3_plan_selection_accepts_hll_given_a_distinct_bound() { } } -/// A bound too large for the precision Pass 1 chose leaves δ over its -/// target: one inside classic HLL's certified range gives a failure -/// probability above 0.01, and one outside it certifies nothing (δ = 1). +/// A bound too large for the precision Pass 1 chose (12 certifies δ = 0.01 +/// for at most 9 distinct items) is met by a larger precision: plan +/// selection enlarges HLL until its failure probability over the bound +/// meets the target, up to precision 18 (#669b): a bound of 10 is met. The +/// model certifies a bound of 100 at no precision (a larger precision also +/// shrinks the relative error it certifies), and one outside its range (at +/// most 4096) certifies nothing. #[test] -fn stage3_plan_selection_rejects_hll_whose_distinct_bound_is_too_large() { - for bound in [10, 100, 10_000_000] { +fn stage3_plan_selection_enlarges_hll_for_a_larger_distinct_bound() { + let run = pipeline_with_distinct_bound(Some(10)); + for id in hll_only_candidates(&run) { + assert!(run.cost(id).is_some(), "{id}: {:?}", run.invalid().get(id)); + } + for bound in [100, 10_000_000] { let run = pipeline_with_distinct_bound(Some(bound)); let invalid = run.invalid(); for id in hll_only_candidates(&run) { From 79ceacada76d503eb1a24ea1645aa691328fa949 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Thu, 15 Oct 2026 16:09:16 +0000 Subject: [PATCH 4/4] feat(planner): a root of at most one row returns one value (#669) G is 1 for a root that returns at most one row whatever its input, provable from the query: a global aggregate or summary (no group keys, not a top-k or count_values), and above it only operators that keep or drop rows, or joins and binary operations of two such inputs (not a full join). A grouped SQL root stays unbound. --- .../src/stage3_plan_selection.rs | 117 +++++++++++++++++- 1 file changed, 112 insertions(+), 5 deletions(-) diff --git a/crates/asap-aware-mapping/src/stage3_plan_selection.rs b/crates/asap-aware-mapping/src/stage3_plan_selection.rs index 922cf37b9..f277ff444 100644 --- a/crates/asap-aware-mapping/src/stage3_plan_selection.rs +++ b/crates/asap-aware-mapping/src/stage3_plan_selection.rs @@ -319,14 +319,17 @@ pub enum SizingError { /// Without a planning time, only evidence that cannot expire is fresh. const ANY_TIME: u64 = u64::MAX; -/// How many values one evaluation of `root` returns at most (`G`), from -/// `data`: one for a scalar root, at most the input cardinality (the -/// series) for a PromQL instant vector, whose groups and series never -/// outnumber its input series. `None` when unbound: a SQL result, until a -/// bound on the groups of its source is an input, and a range vector. +/// How many values one evaluation of `root` returns at most (`G`): one for +/// a scalar root and for a root of at most one row ([`at_most_one_row`]), +/// provable from the query; otherwise, from `data`, at most the input +/// cardinality (the series) for a PromQL instant vector, whose groups and +/// series never outnumber its input series. `None` when unbound: any other +/// SQL result, until a bound on the groups of its source is an input, and a +/// range vector. fn output_values(root: &QueryRoot, data: &DataWorkload) -> Option { match root { QueryRoot::Scalar(_) => Some(1.0), + QueryRoot::Operator(node) if at_most_one_row(node) => Some(1.0), QueryRoot::Operator(node) if node.result_kind == OperatorResultKind::InstantVector => data .input_cardinality .value_at(ANY_TIME) @@ -335,6 +338,58 @@ fn output_values(root: &QueryRoot, data: &DataWorkload) -> Option { } } +/// Whether `node` returns at most one row whatever its input: a global +/// aggregate or summary (a reduction with no group keys), and above one +/// only operators that keep or drop its rows, or join two such inputs +/// without adding unmatched rows of both sides. +fn at_most_one_row(node: &OperatorNode) -> bool { + let global = |reduction: &asap_types::pre_asap::Reduction| { + reduction + .group_keys() + .is_some_and(|by| !by.is_without() && by.keys().is_empty()) + }; + match &node.operator { + // A top-k or `count_values` returns several rows per group. + Operator::NonASAP(NonASAPOp::Aggregate { + reduction, + measures, + .. + }) => { + global(reduction) + && !measures + .iter() + .any(|m| matches!(m, AggIntent::TopK { .. } | AggIntent::CountValues { .. })) + } + Operator::ASAP(ASAPOp::SummaryAgg { reduction, .. }) => global(reduction), + Operator::ASAP(ASAPOp::SummaryEstimate { + summary_input, + query, + }) => !matches!(query, SketchStatistic::TopK { .. }) && at_most_one_row(summary_input), + Operator::ASAP(ASAPOp::FinalizeExactAccumulator { child }) => at_most_one_row(child), + Operator::NonASAP( + NonASAPOp::Project { child, .. } + | NonASAPOp::Filter { child, .. } + | NonASAPOp::Sort { child, .. } + | NonASAPOp::Limit { child, .. } + | NonASAPOp::Dedup { child, .. }, + ) => at_most_one_row(child), + Operator::NonASAP(NonASAPOp::Join { + kind, left, right, .. + }) => { + use asap_types::ir::operator_properties::JoinKind; + match kind { + JoinKind::Semi | JoinKind::Anti => at_most_one_row(left), + JoinKind::Full => false, + _ => at_most_one_row(left) && at_most_one_row(right), + } + } + Operator::NonASAP(NonASAPOp::BinaryOp { lhs, rhs, .. }) => { + at_most_one_row(lhs) && at_most_one_row(rhs) + } + _ => false, + } +} + /// `candidate` with each summary's per-value failure probability enlarged /// for its queries' targets (#669b), before pricing, so that cost reflects /// the deeper sketch. A query with a target with failure probability `δ`, @@ -2108,6 +2163,58 @@ mod tests { ); } + /// A root that returns at most one row has `G` = 1, provable from the + /// query with no data workload: a global aggregate, PromQL or SQL, and + /// what selects, renames or joins single rows above it. A grouped SQL + /// aggregate stays unbound, and a grouped PromQL one, or a global top-k + /// (k rows), is bound by the input cardinality. + #[test] + fn a_root_of_at_most_one_row_returns_one_value() { + let none = DataWorkload::default(); + let root = |query: &str| { + QueryRoot::Operator(lower_promql(query, RequestedAccuracy::epsilon(0.01))) + }; + assert_eq!(output_values(&root("sum(rate(x[5m]))"), &none), Some(1.0)); + assert_eq!(output_values(&root("topk(5, rate(x[5m]))"), &none), None); + assert_eq!( + output_values(&root("sum by (job) (rate(x[5m]))"), &none), + None + ); + assert_eq!( + output_values(&root("sum by (job) (rate(x[5m]))"), &data()), + Some(10_000.0) + ); + let table = || { + crate::test_support::scan_from( + asap_types::pre_asap::Source::Table { + table_ref: "t".into(), + }, + Schema::new(vec![ + asap_types::pre_asap::schema::Field::plain("k", DataType::Utf8, false), + asap_types::pre_asap::schema::Field::plain("v", DataType::Float64, false), + ]), + ) + }; + let count = || AggIntent::Count { + accuracy: AccuracyTarget::exact(), + }; + let global = crate::test_support::agg(vec![], count(), table()); + let grouped = crate::test_support::agg(vec![0], count(), table()); + assert_eq!( + output_values(&QueryRoot::Operator(global.clone()), &none), + Some(1.0) + ); + assert_eq!(output_values(&QueryRoot::Operator(grouped), &data()), None); + let cross = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Join { + kind: asap_types::ir::operator_properties::JoinKind::Cross, + pred: asap_types::ir::Predicate(ScalarExpr::Literal(ScalarValue::Boolean(true))), + left: global.clone(), + right: global, + })) + .unwrap(); + assert_eq!(output_values(&QueryRoot::Operator(cross), &none), Some(1.0)); + } + /// Deployment evidence that every output was measured exact. struct EveryOutputMeasuredExact;