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/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 c82baa749..0d1916dd6 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, @@ -24,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; @@ -66,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; @@ -120,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 @@ -134,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; @@ -156,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, @@ -195,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, @@ -235,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/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/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 440eb56f8..4f9f4787c 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,210 @@ 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(OperatorNode::reachable) + .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), + } +} + +/// `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( + 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 +982,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 +1009,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 +1093,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 +2157,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 +2208,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 +2292,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 +2301,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 +2326,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 +2368,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 +2625,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 +3802,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); diff --git a/crates/asap-aware-mapping/src/replacement.rs b/crates/asap-aware-mapping/src/replacement.rs index 139050f5b..f73ed16bc 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 { @@ -1643,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() { @@ -1795,7 +1791,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 +2044,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))?; } } } @@ -3854,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; @@ -6002,14 +5987,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, ), ), @@ -6153,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. @@ -7956,12 +7939,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 +10140,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 +10173,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 +10239,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 +10313,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 +10364,6 @@ mod tests { let strategy = ASAPStrategies::new_with_planning_inputs_and_evidence( &DefaultCostModel, &DefaultAccuracyModel, - &EqualSplitAllocator, &SeparatedTopKEvidence, ); @@ -10441,7 +10416,6 @@ mod tests { let strategy = ASAPStrategies::new_with_planning_inputs_and_evidence( &DefaultCostModel, &DefaultAccuracyModel, - &EqualSplitAllocator, &SeparatedTopKEvidence, ); let (update, reduction) = strategy @@ -10706,11 +10680,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 +10707,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/asap-aware-mapping/src/stage3_plan_selection.rs b/crates/asap-aware-mapping/src/stage3_plan_selection.rs index aa8f56c8b..f277ff444 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,181 @@ 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`): 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) + .map(|series| (*series).max(1) as f64), + QueryRoot::Operator(_) => None, + } +} + +/// 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 `δ`, +/// 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 +529,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 +646,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 +696,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 +2053,168 @@ 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) + ); + } + + /// 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; 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) { 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()