From d21c7c6844303dde72b3f370a0f754b9e047f214 Mon Sep 17 00:00:00 2001 From: zz_y Date: Wed, 26 Aug 2026 17:40:45 -0600 Subject: [PATCH] feat(accuracy): propagate end-to-end accuracy guarantees for nested summaries (#172) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR 1 + PR 2 of the stacked plan in #172. Represent guarantees and fail closed: - asap-types: `post_asap::guarantee` — `ErrorMetric` (#[non_exhaustive]), symbolic `BoundExpr`/`ProbabilityExpr` expression vocabularies (evaluate to `None`, never 0, on an unknown statistic), `GuaranteeSource` provenance, `ResultGuarantee`, `CompositionOperator`, and typed `AccuracyError::{UnsupportedComposition, MissingInputGuarantee, TargetNotSatisfied, NoLegalAllocation}`. - `SummaryNode.guarantee: Option` on every finalized value (readouts, exact accumulators, kept pre-ASAP subtrees); raw sketch state carries none. Exact values are zero-error; sketch readouts get a family-specific local guarantee by inverting `default_size_params`. - `construct_summary_agg` detects approximate-child -> approximate-parent composition and rejects it unless the `AccuracyModel` has a rule — never treating the child as exact. Rejections are typed `RejectedCandidate`s on `MemoGroup::rejected`, via a new `ReplacementStrategy::propose` hook (default delegates to `replacements`). - DAG export: `SummaryDagNode.guarantee`, guarantee in merged-graph `detail`, and `NamedGraph.rejections` — all additive/omitted when empty. Conservative same-metric propagation and budget allocation: - `asap_aware_mapping::accuracy` — `AccuracyModel` trait (`local_guarantee`/`propagate`/`satisfies`) and `DefaultAccuracyModel`: exact-input, additive absolute (B_in + B_out, delta by union bound), relative with cross term (needs known sign), explicitly registered L-Lipschitz, exact sum (sum of bounds) and max/min over approximate inputs; incompatible metrics -> `UnsupportedComposition`. No independence assumptions anywhere. - `AccuracyBudgetAllocator` trait + `EqualSplitAllocator` (eps_i = eps/n, delta_i = delta/n; (1+eps)^(1/n)-1 for relative); each allocation re-sizes the outer layer through the existing `CostModel::size_params` path and re-enumerates the child under its share. Only legal allocations become candidates. - `search_workload_with_targets` checks a root `QueryRequirements` target against the root group's bound candidates before any cost ranking; `CostModel` never sees a rejected candidate. - Precedence between root and per-node targets documented in `accuracy.rs` and the design doc. Deferred to PR 3: TopK margin certificate and #239 posterior refinement. Co-Authored-By: Claude Fable 5 --- crates/asap-aware-mapping/src/accuracy.rs | 1088 ++++++++++++++++++ crates/asap-aware-mapping/src/cost_model.rs | 2 + crates/asap-aware-mapping/src/grouping.rs | 6 + crates/asap-aware-mapping/src/lib.rs | 25 +- crates/asap-aware-mapping/src/replacement.rs | 1000 +++++++++++++++- crates/devtools/src/bin/dag_export.rs | 45 +- crates/types/src/dag_export.rs | 187 ++- crates/types/src/post_asap/expr.rs | 13 + crates/types/src/post_asap/guarantee.rs | 450 ++++++++ crates/types/src/post_asap/mod.rs | 5 + crates/types/src/post_asap/sketch.rs | 6 +- docs/design_docs/asap_aware_mapping.md | 31 + 12 files changed, 2799 insertions(+), 59 deletions(-) create mode 100644 crates/asap-aware-mapping/src/accuracy.rs create mode 100644 crates/types/src/post_asap/guarantee.rs diff --git a/crates/asap-aware-mapping/src/accuracy.rs b/crates/asap-aware-mapping/src/accuracy.rs new file mode 100644 index 0000000..9393404 --- /dev/null +++ b/crates/asap-aware-mapping/src/accuracy.rs @@ -0,0 +1,1088 @@ +//! Planning-time accuracy algebra (issue #172): the [`AccuracyModel`] +//! extension point, its conservative default, and end-to-end +//! accuracy-budget allocation. +//! +//! ## Why a second trait next to `CostModel` +//! +//! Accuracy legality and cost ranking are different responsibilities. +//! [`crate::cost_model::CostModel`] answers "which legal candidate is +//! cheapest"; this module answers "which candidates are legal at all". The +//! pipeline [`crate::replacement`] runs is, in order: +//! +//! ```text +//! candidate generation +//! -> guarantee propagation (AccuracyModel::propagate) +//! -> AccuracyTarget satisfaction (AccuracyModel::satisfies) +//! -> legal candidates only (illegal ones become MemoGroup::rejected) +//! -> cost ranking / global selection (CostModel) +//! ``` +//! +//! A `CostModel` only ever sees the survivors, so it cannot override a +//! legality decision — the same "permutation only, never prune" contract +//! `CostModel::rank_candidates` already has, applied one stage earlier. +//! +//! ## What the default model admits +//! +//! [`DefaultAccuracyModel`] is deliberately conservative and fail-closed: +//! +//! | operator | rule | result | +//! |---|---|---| +//! | any, all inputs exact | exact input | the local guarantee (or exact) | +//! | `ApproximateAggregate`, all `AbsoluteValue` | additive | `Σ B`, `δ` by union bound | +//! | `ApproximateAggregate`, all `RelativeValue`, values known non-negative | multiplicative | `ε_in + ε_out + ε_in·ε_out`, `δ` by union bound | +//! | `Lipschitz { L }`, one `AbsoluteValue` input | Lipschitz | `L·B_in + B_local`, `δ` by union bound | +//! | `ExactSum`, value-like inputs | sum | `Σ B_i` (`AbsoluteValue`), `δ` by union bound over inputs | +//! | `ExactExtremum`, same-metric inputs | max/min | `max B_i`, `δ` by union bound over inputs | +//! | anything else | — | [`AccuracyError::UnsupportedComposition`] | +//! +//! Cross-metric compositions (a `Rank` error under a value-additive rule, +//! a `Cardinality` error under a `Frequency` sketch, …) have no registered +//! rule and are rejected. The child is **never** treated as exact. Nothing +//! assumes independence: every probability combinator is the union bound. +//! A statistic the rule needs but [`PropagationStats`] does not supply +//! (an input row count, a stream's L1 norm) stays a +//! [`BoundExpr::Unknown`] leaf — the guarantee is still produced, but it +//! cannot satisfy any target until something instantiates the statistic. +//! +//! ## Precedence between root and per-node targets +//! +//! - A root `QueryRequirements.accuracy`, when supplied to +//! [`crate::replacement::search_workload_with_targets`], is the +//! end-to-end target for that query's root value. It is checked against +//! the root group's candidates *before* cost ranking; a candidate whose +//! guarantee is unknown, or misses the target, is moved to +//! `MemoGroup::rejected`. +//! - For an approximate node over an **exact** child, the node's own +//! `AggIntent.accuracy` sizes its sketch, exactly as before this module +//! existed, and the readout's guarantee is that sketch's local guarantee. +//! - For an approximate node over an **approximate** child, the outer +//! node's `AggIntent.accuracy` is the end-to-end target *for that value*. +//! The inner node's `AggIntent.accuracy` is only its declared local +//! requirement: the as-declared composition is evaluated and kept only if +//! it satisfies the outer target, and the [`AccuracyBudgetAllocator`] +//! additionally proposes re-sized splits of the outer target. A front end +//! that copied the same target onto every node has therefore *not* +//! produced a valid end-to-end allocation — the composed guarantee is +//! what decides. +//! - `AccuracyTarget::Exact` on a node admits only exact realizations +//! (unchanged), and an approximate layer can never satisfy it. + +use asap_types::post_asap::{ + AccuracyError, BoundExpr, CompositionOperator, ErrorMetric, GuaranteeSource, ProbabilityExpr, + ResultGuarantee, SketchAlgorithm, SketchParams, SketchQuery, SummaryFamilyType, +}; +use asap_types::types::AccuracyTarget; + +/// Statistics a propagation rule may consult. Every field is optional and +/// defaults to "unknown": a rule that needs a missing statistic emits a +/// [`BoundExpr::Unknown`] leaf (or rejects) rather than guessing. +#[derive(Debug, Clone, Default, PartialEq)] +pub struct PropagationStats { + /// Whether every input value is known to be non-negative — required by + /// the multiplicative relative-error rule, which is unsound across a + /// sign change. + pub values_non_negative: Option, + /// Number of input rows an exact aggregation consumes (e.g. the number + /// of groups a `sum` folds), for `ExactSum`/`ExactExtremum`'s union + /// bound over per-input failures. + pub input_row_count: Option, +} + +/// The deployment-extensible accuracy algebra. `asap-aware-mapping` ships +/// [`DefaultAccuracyModel`]; a deployment with a proof for a composition the +/// default rejects (a registered cross-metric conversion, say) implements +/// this trait and passes it to +/// [`crate::replacement::SketchAlgorithmStrategy::with_models`]. +pub trait AccuracyModel { + /// The guarantee of reading `query` out of a summary of family `family` + /// built over an **exact** input — derived from the family's committed + /// parameters by inverting the same sizing formulas + /// [`crate::replacement::default_size_params`] uses. `None` when this + /// model has no error model for the family (the default has none for + /// `Sample`/`Wavelet`/`StatModel`). + fn local_guarantee( + &self, + family: &SummaryFamilyType, + query: &SketchQuery, + ) -> Option; + + /// Compose `inputs`' guarantees (in the parent's child order) with the + /// parent's own `local` guarantee under `op`. `Err` is the fail-closed + /// answer: no registered rule, or a missing input guarantee. + fn propagate( + &self, + op: &CompositionOperator, + inputs: &[ResultGuarantee], + local: Option<&ResultGuarantee>, + stats: &PropagationStats, + ) -> Result; + + /// Does `guarantee` meet `target`? An unevaluable bound or probability + /// never satisfies anything. + fn satisfies(&self, guarantee: &ResultGuarantee, target: &AccuracyTarget) -> bool; +} + +/// The conservative, fail-closed default — see the module docs' table. +#[derive(Debug, Default, Clone, Copy)] +pub struct DefaultAccuracyModel; + +/// Failure probability this crate attributes to a KLL bound sized by +/// `k = ⌈2/ε⌉`: the 99%-confidence convention the `k`-to-rank-error tables +/// of the reference implementation (Apache DataSketches) quote that formula +/// at. KLL's sizing ignores δ, so this is the confidence level the sizing +/// implicitly claims, not something re-derived from `k`. +pub const KLL_FAILURE_PROBABILITY: f64 = 0.01; + +/// Failure probability this crate attributes to an HLL/Theta/KMV bound +/// sized by its relative *standard error* (`1.04/√m`, `1/√k`): a +/// one-standard-deviation bound of an approximately Gaussian estimator is +/// exceeded with probability ≈ 0.3173. Those families' sizing ignores δ, so +/// an `EpsilonDelta` target with a tighter δ is honestly not met by the +/// parameters the sizing picked; the guarantee says so instead of hiding it. +pub const STANDARD_ERROR_FAILURE_PROBABILITY: f64 = 0.3173; + +/// Small relative tolerance for comparing an evaluated bound against a +/// target, so a parameter sized by `⌈·⌉` to *exactly* meet ε is not rejected +/// by floating-point noise. +const SATISFACTION_TOLERANCE: f64 = 1e-9; + +impl DefaultAccuracyModel { + /// The local guarantee of one sketch `(algorithm, params)` for `query` + /// — each arm inverts the matching formula in + /// [`crate::replacement::default_size_params`]. + pub fn sketch_guarantee( + algorithm: &SketchAlgorithm, + params: &SketchParams, + query: &SketchQuery, + ) -> Option { + let (metric, bound, delta) = match params { + // KLL: rank error ε ≈ 2/k. + SketchParams::Kll { k } => ( + ErrorMetric::Rank, + 2.0 / f64::from(*k), + ProbabilityExpr::Constant { + value: KLL_FAILURE_PROBABILITY, + }, + ), + // DDSketch: deterministic relative value error α. + SketchParams::DDSketch { alpha } => { + (ErrorMetric::RelativeValue, *alpha, ProbabilityExpr::Zero) + } + // HLL: relative standard error 1.04/√(2^p). + SketchParams::Hll { precision } => ( + ErrorMetric::Cardinality, + 1.04 / 2f64.powi(i32::from(*precision)).sqrt(), + ProbabilityExpr::Constant { + value: STANDARD_ERROR_FAILURE_PROBABILITY, + }, + ), + // KMV / Theta: relative standard error 1/√k. + SketchParams::Kmv { k } | SketchParams::Theta { k } => ( + ErrorMetric::Cardinality, + 1.0 / f64::from(*k).sqrt(), + ProbabilityExpr::Constant { + value: STANDARD_ERROR_FAILURE_PROBABILITY, + }, + ), + // CMS family: over-count ≤ (e/w)·‖f‖₁ with probability ≥ 1 − e^{−d}. + // Count-Sketch is sized with the same placeholder formula (see + // `default_size_params`), so it gets the same placeholder bound. + SketchParams::Cms { width, depth } + | SketchParams::CountSketch { width, depth } + | SketchParams::CmsWithHeap { width, depth, .. } + | SketchParams::CountSketchWithHeap { width, depth, .. } => ( + ErrorMetric::Frequency, + std::f64::consts::E / f64::from(*width), + ProbabilityExpr::Constant { + value: (-f64::from(*depth)).exp(), + }, + ), + }; + let mut provenance = vec![GuaranteeSource::SketchReadout { + algorithm: format!("{algorithm:?}"), + params: serde_json::to_value(params).unwrap_or(serde_json::Value::Null), + query: format!("{query:?}"), + }]; + // A `TopK` readout's count for each *reported* key carries the + // sketch's frequency bound; which keys are reported is not certified + // by anything here (issue #172, PR 3) — say so in the trail rather + // than claiming `TopKMembership`. + if matches!(query, SketchQuery::TopK { .. }) { + provenance.push(GuaranteeSource::UnavailableStatistic { + statistic: "topk_membership_margin_certificate".into(), + }); + } + Some(ResultGuarantee { + metric, + bound: BoundExpr::Constant { value: bound }, + failure_probability: delta, + provenance, + }) + } + + fn additive( + op: &CompositionOperator, + inputs: &[ResultGuarantee], + local: &ResultGuarantee, + rule: &str, + ) -> ResultGuarantee { + let mut terms: Vec = inputs.iter().map(|g| g.bound.clone()).collect(); + terms.push(local.bound.clone()); + let mut deltas: Vec = inputs + .iter() + .map(|g| g.failure_probability.clone()) + .collect(); + deltas.push(local.failure_probability.clone()); + ResultGuarantee { + metric: local.metric, + bound: BoundExpr::Sum { terms }, + failure_probability: ProbabilityExpr::UnionBound { terms: deltas }, + provenance: composed_provenance(op, inputs, local, rule), + } + } + + /// `(1 + ε_total) = Π (1 + ε_i)` ⇒ for two factors + /// `ε_in + ε_out + ε_in·ε_out`; written out as the sum of all + /// cross-products so the expression tree is exact for any input count. + fn multiplicative( + op: &CompositionOperator, + inputs: &[ResultGuarantee], + local: &ResultGuarantee, + ) -> ResultGuarantee { + let factors: Vec<&BoundExpr> = inputs + .iter() + .map(|g| &g.bound) + .chain(std::iter::once(&local.bound)) + .collect(); + // Every non-empty subset's product: Π(1+ε_i) − 1 = Σ_{S≠∅} Π_{i∈S} ε_i. + let mut terms = Vec::new(); + for mask in 1..(1u32 << factors.len()) { + let subset: Vec = factors + .iter() + .enumerate() + .filter(|(i, _)| mask & (1 << i) != 0) + .map(|(_, b)| (*b).clone()) + .collect(); + terms.push(if subset.len() == 1 { + subset.into_iter().next().expect("one element") + } else { + BoundExpr::Product { factors: subset } + }); + } + let mut deltas: Vec = inputs + .iter() + .map(|g| g.failure_probability.clone()) + .collect(); + deltas.push(local.failure_probability.clone()); + ResultGuarantee { + metric: ErrorMetric::RelativeValue, + bound: BoundExpr::Sum { terms }, + failure_probability: ProbabilityExpr::UnionBound { terms: deltas }, + provenance: composed_provenance(op, inputs, local, "relative_cross_term_union_bound"), + } + } + + fn lipschitz( + op: &CompositionOperator, + constant: f64, + inputs: &[ResultGuarantee], + local: Option<&ResultGuarantee>, + ) -> ResultGuarantee { + let input = &inputs[0]; + let scaled = BoundExpr::Scaled { + factor: constant, + inner: Box::new(input.bound.clone()), + }; + let (bound, delta) = match local { + Some(local) => ( + BoundExpr::Sum { + terms: vec![scaled, local.bound.clone()], + }, + ProbabilityExpr::UnionBound { + terms: vec![ + input.failure_probability.clone(), + local.failure_probability.clone(), + ], + }, + ), + None => (scaled, input.failure_probability.clone()), + }; + let exact_local = ResultGuarantee::exact("deterministic Lipschitz transformation"); + ResultGuarantee { + metric: ErrorMetric::AbsoluteValue, + bound, + failure_probability: delta, + provenance: composed_provenance( + op, + inputs, + local.unwrap_or(&exact_local), + "lipschitz_union_bound", + ), + } + } + + /// Exact `sum` over approximate inputs: `B ≤ Σ B_i`, `δ ≤ Σ δ_i`. The + /// planner composes one *per-value* child guarantee over an unknown + /// number of input rows, so both the bound and the union bound scale by + /// `stats.input_row_count` — an [`BoundExpr::Unknown`] leaf when it is + /// not supplied. Each input's normalized bound is first converted to + /// absolute units via the statistic its metric is normalized by (also + /// unknown unless supplied); a `Rank` input has no such conversion. + fn exact_sum( + op: &CompositionOperator, + inputs: &[ResultGuarantee], + stats: &PropagationStats, + ) -> Result { + let mut terms = Vec::with_capacity(inputs.len()); + let mut deltas = Vec::with_capacity(inputs.len()); + let mut provenance = Vec::new(); + for (i, input) in inputs.iter().enumerate() { + let absolute = + absolute_bound(input).ok_or_else(|| AccuracyError::UnsupportedComposition { + operator: op.clone(), + input_metrics: inputs.iter().map(|g| g.metric).collect(), + local_metric: None, + reason: format!( + "input {i} carries a {:?} guarantee, which has no registered \ + conversion to an absolute value error", + input.metric + ), + })?; + if let BoundExpr::Product { factors } = &absolute { + for f in factors { + if let BoundExpr::Unknown { statistic } = f { + provenance.push(GuaranteeSource::UnavailableStatistic { + statistic: statistic.clone(), + }); + } + } + } + terms.push(absolute); + deltas.push(input.failure_probability.clone()); + } + let count = row_count(stats, &mut provenance); + let exact_local = ResultGuarantee::exact("ExactAggregate(Sum)"); + provenance.extend(composed_provenance( + op, + inputs, + &exact_local, + "exact_sum_union_bound", + )); + Ok(ResultGuarantee { + metric: ErrorMetric::AbsoluteValue, + bound: BoundExpr::Product { + factors: vec![count.clone(), BoundExpr::Sum { terms }], + }, + failure_probability: ProbabilityExpr::Scaled { + count, + inner: Box::new(ProbabilityExpr::UnionBound { terms: deltas }), + }, + provenance, + }) + } + + /// Exact `max`/`min` over approximate inputs of one shared metric: the + /// returned value's error is at most the largest input bound (order + /// statistics are monotone under a uniform perturbation), with + /// probability by the union bound over every input row. This bounds the + /// returned *value*; it does not identify the true winning key. + fn exact_extremum( + op: &CompositionOperator, + inputs: &[ResultGuarantee], + stats: &PropagationStats, + ) -> Result { + let metric = inputs[0].metric; + if inputs.iter().any(|g| g.metric != metric) || metric == ErrorMetric::TopKMembership { + return Err(AccuracyError::UnsupportedComposition { + operator: op.clone(), + input_metrics: inputs.iter().map(|g| g.metric).collect(), + local_metric: None, + reason: "exact max/min needs every input under one value-like metric".into(), + }); + } + let mut provenance = Vec::new(); + let count = row_count(stats, &mut provenance); + let exact_local = ResultGuarantee::exact("ExactAggregate(MinMax)"); + provenance.extend(composed_provenance( + op, + inputs, + &exact_local, + "exact_extremum_union_bound", + )); + Ok(ResultGuarantee { + metric, + bound: BoundExpr::Max { + terms: inputs.iter().map(|g| g.bound.clone()).collect(), + }, + failure_probability: ProbabilityExpr::Scaled { + count, + inner: Box::new(ProbabilityExpr::UnionBound { + terms: inputs + .iter() + .map(|g| g.failure_probability.clone()) + .collect(), + }), + }, + provenance, + }) + } +} + +/// `stats.input_row_count` as a bound factor, or an `Unknown` leaf (recorded +/// in `provenance`) when absent. +fn row_count(stats: &PropagationStats, provenance: &mut Vec) -> BoundExpr { + match stats.input_row_count { + Some(n) => BoundExpr::Constant { value: n as f64 }, + None => { + provenance.push(GuaranteeSource::UnavailableStatistic { + statistic: "input_row_count".into(), + }); + BoundExpr::Unknown { + statistic: "input_row_count".into(), + } + } + } +} + +/// `input`'s bound converted to absolute value units, multiplying a +/// normalized metric by the (unknown) statistic it is normalized by. `None` +/// for a metric with no such conversion (`Rank`, `TopKMembership`). +fn absolute_bound(input: &ResultGuarantee) -> Option { + let normalizer = match input.metric { + ErrorMetric::AbsoluteValue => return Some(input.bound.clone()), + ErrorMetric::RelativeValue => "true_value_magnitude", + ErrorMetric::Cardinality => "true_cardinality", + ErrorMetric::Frequency => "stream_l1_norm", + // `Rank` has no distribution-free conversion to a value error; a + // metric this crate does not know has no registered conversion. + ErrorMetric::Rank | ErrorMetric::TopKMembership | _ => return None, + }; + if input.bound.is_zero() { + return Some(BoundExpr::Zero); + } + Some(BoundExpr::Product { + factors: vec![ + input.bound.clone(), + BoundExpr::Unknown { + statistic: normalizer.into(), + }, + ], + }) +} + +fn composed_provenance( + op: &CompositionOperator, + inputs: &[ResultGuarantee], + local: &ResultGuarantee, + rule: &str, +) -> Vec { + let mut provenance: Vec = inputs + .iter() + .enumerate() + .map(|(input_index, g)| GuaranteeSource::ChildGuarantee { + input_index, + guarantee: Box::new(g.clone()), + }) + .collect(); + provenance.extend(local.provenance.iter().cloned()); + provenance.push(GuaranteeSource::CompositionStep { + operator: op.clone(), + rule: rule.into(), + }); + provenance +} + +impl AccuracyModel for DefaultAccuracyModel { + fn local_guarantee( + &self, + family: &SummaryFamilyType, + query: &SketchQuery, + ) -> Option { + match family { + SummaryFamilyType::Plain(_) => Some(ResultGuarantee::exact("Plain value")), + SummaryFamilyType::ExactAggregate(kind, _) => { + Some(ResultGuarantee::exact(format!("ExactAggregate({kind:?})"))) + } + SummaryFamilyType::Sketch(kind, _) => { + Self::sketch_guarantee(kind.algorithm(), kind.params(), query) + } + // No error model is registered for these families. + SummaryFamilyType::Sample(..) + | SummaryFamilyType::Wavelet(..) + | SummaryFamilyType::StatModel(..) => None, + } + } + + fn propagate( + &self, + op: &CompositionOperator, + inputs: &[ResultGuarantee], + local: Option<&ResultGuarantee>, + stats: &PropagationStats, + ) -> Result { + // Exact input: only the local guarantee remains (or the value is exact). + if inputs.iter().all(ResultGuarantee::is_exact) { + return Ok(match local { + Some(local) => { + let mut out = local.clone(); + out.provenance + .extend(inputs.iter().enumerate().map(|(input_index, g)| { + GuaranteeSource::ChildGuarantee { + input_index, + guarantee: Box::new(g.clone()), + } + })); + out.provenance.push(GuaranteeSource::CompositionStep { + operator: op.clone(), + rule: "exact_input".into(), + }); + out + } + None => { + let mut out = ResultGuarantee::exact(format!("{op:?} over exact inputs")); + out.provenance.push(GuaranteeSource::CompositionStep { + operator: op.clone(), + rule: "exact_input".into(), + }); + out + } + }); + } + + let input_metrics: Vec = inputs.iter().map(|g| g.metric).collect(); + let unsupported = |reason: String| AccuracyError::UnsupportedComposition { + operator: op.clone(), + input_metrics: input_metrics.clone(), + local_metric: local.map(|g| g.metric), + reason, + }; + // An exact input is compatible with every metric; only approximate + // inputs constrain the rule. + let approximate: Vec<&ResultGuarantee> = inputs.iter().filter(|g| !g.is_exact()).collect(); + let same_metric = |metric: ErrorMetric| approximate.iter().all(|g| g.metric == metric); + + match op { + CompositionOperator::ApproximateAggregate => { + let local = local.ok_or_else(|| { + unsupported("approximate operator has no local guarantee to compose".into()) + })?; + if !same_metric(local.metric) { + return Err(unsupported(format!( + "no registered cross-metric rule from {input_metrics:?} to {:?}", + local.metric + ))); + } + match local.metric { + ErrorMetric::AbsoluteValue => { + Ok(Self::additive(op, inputs, local, "additive_union_bound")) + } + ErrorMetric::RelativeValue => { + if stats.values_non_negative != Some(true) { + return Err(unsupported( + "relative-error composition needs values of known sign \ + (PropagationStats::values_non_negative)" + .into(), + )); + } + Ok(Self::multiplicative(op, inputs, local)) + } + ErrorMetric::Rank + | ErrorMetric::Cardinality + | ErrorMetric::Frequency + | ErrorMetric::TopKMembership + | _ => Err(unsupported(format!( + "no registered same-metric composition rule for {:?} over {:?}", + local.metric, local.metric + ))), + } + } + CompositionOperator::Lipschitz { constant } => { + if !(constant.is_finite() && *constant >= 0.0) { + return Err(unsupported(format!( + "Lipschitz constant {constant} is not a finite non-negative number" + ))); + } + if inputs.len() != 1 || !same_metric(ErrorMetric::AbsoluteValue) { + return Err(unsupported( + "Lipschitz rule is registered for exactly one AbsoluteValue input".into(), + )); + } + if local.is_some_and(|g| g.metric != ErrorMetric::AbsoluteValue) { + return Err(unsupported( + "Lipschitz rule needs an AbsoluteValue local guarantee".into(), + )); + } + Ok(Self::lipschitz(op, *constant, inputs, local)) + } + CompositionOperator::ExactSum => Self::exact_sum(op, inputs, stats), + CompositionOperator::ExactExtremum => Self::exact_extremum(op, inputs, stats), + CompositionOperator::TopKSelection => Err(unsupported( + "top-k membership over approximate inputs needs a margin certificate \ + (issue #172, PR 3)" + .into(), + )), + // An operator this crate does not know has no registered rule. + _ => Err(unsupported("no registered rule for this operator".into())), + } + } + + fn satisfies(&self, guarantee: &ResultGuarantee, target: &AccuracyTarget) -> bool { + let within = |value: Option, limit: f64| { + value.is_some_and(|v| v <= limit * (1.0 + SATISFACTION_TOLERANCE) + f64::EPSILON) + }; + match target { + AccuracyTarget::Exact => guarantee.is_exact(), + AccuracyTarget::Epsilon(eps) => within(guarantee.bound.evaluate(), *eps), + AccuracyTarget::EpsilonDelta { epsilon, delta } => { + within(guarantee.bound.evaluate(), *epsilon) + && within(guarantee.failure_probability.evaluate(), *delta) + } + } + } +} + +// ── Budget allocation ─────────────────────────────────────────────────────── + +/// 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 subtree 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| match t { + AccuracyTarget::Exact => (0.0, Some(0.0)), + AccuracyTarget::Epsilon(e) => (*e, None), + AccuracyTarget::EpsilonDelta { epsilon, delta } => (*epsilon, Some(*delta)), + }) + .unzip(); + let epsilon = match shape.metric { + ErrorMetric::RelativeValue => eps.iter().map(|e| 1.0 + e).product::() - 1.0, + _ => eps.iter().sum(), + }; + Some(match delta.iter().copied().sum::>() { + Some(delta) => AccuracyTarget::EpsilonDelta { epsilon, delta }, + None => AccuracyTarget::Epsilon(epsilon), + }) + } +} + +/// 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); + let (epsilon, delta) = match target { + AccuracyTarget::Exact => return Vec::new(), + AccuracyTarget::Epsilon(e) => (*e, None), + AccuracyTarget::EpsilonDelta { epsilon, delta } => (*epsilon, Some(*delta)), + }; + if !(epsilon.is_finite() && epsilon > 0.0) { + return Vec::new(); + } + let local_epsilon = match composition.metric { + ErrorMetric::RelativeValue => (1.0 + epsilon).powf(1.0 / n as f64) - 1.0, + _ => epsilon / n as f64, + }; + let layer = match delta { + Some(delta) => AccuracyTarget::EpsilonDelta { + epsilon: local_epsilon, + delta: delta / n as f64, + }, + None => AccuracyTarget::Epsilon(local_epsilon), + }; + vec![AccuracyAllocation { + allocator: "EqualSplitAllocator", + layers: vec![layer; n], + }] + } +} + +#[cfg(test)] +mod tests { + use super::*; + use asap_types::post_asap::{GroupingStrategy, SketchKind}; + + fn abs(bound: f64, delta: f64) -> ResultGuarantee { + ResultGuarantee { + metric: ErrorMetric::AbsoluteValue, + bound: BoundExpr::Constant { value: bound }, + failure_probability: ProbabilityExpr::Constant { value: delta }, + provenance: vec![], + } + } + + fn rel(bound: f64) -> ResultGuarantee { + ResultGuarantee { + metric: ErrorMetric::RelativeValue, + bound: BoundExpr::Constant { value: bound }, + failure_probability: ProbabilityExpr::Zero, + provenance: vec![], + } + } + + fn with_metric(metric: ErrorMetric, bound: f64) -> ResultGuarantee { + ResultGuarantee { + metric, + ..abs(bound, 0.0) + } + } + + #[test] + fn exact_child_contributes_zero_error() { + let local = abs(0.05, 0.01); + let out = DefaultAccuracyModel + .propagate( + &CompositionOperator::ApproximateAggregate, + &[ResultGuarantee::exact("sum")], + Some(&local), + &PropagationStats::default(), + ) + .unwrap(); + assert_eq!(out.bound.evaluate(), Some(0.05)); + assert_eq!(out.failure_probability.evaluate(), Some(0.01)); + assert_eq!(out.metric, ErrorMetric::AbsoluteValue); + } + + #[test] + fn additive_bounds_and_delta_union_bound_compose() { + let out = DefaultAccuracyModel + .propagate( + &CompositionOperator::ApproximateAggregate, + &[abs(0.02, 0.01)], + Some(&abs(0.03, 0.02)), + &PropagationStats::default(), + ) + .unwrap(); + assert!((out.bound.evaluate().unwrap() - 0.05).abs() < 1e-12); + // Union bound, not 1 − (1−0.01)(1−0.02) = 0.0298. + assert!((out.failure_probability.evaluate().unwrap() - 0.03).abs() < 1e-12); + assert!(out.provenance.iter().any(|s| matches!( + s, + GuaranteeSource::CompositionStep { rule, .. } if rule == "additive_union_bound" + ))); + } + + #[test] + fn relative_error_includes_the_cross_term() { + let stats = PropagationStats { + values_non_negative: Some(true), + ..Default::default() + }; + let out = DefaultAccuracyModel + .propagate( + &CompositionOperator::ApproximateAggregate, + &[rel(0.1)], + Some(&rel(0.2)), + &stats, + ) + .unwrap(); + // 0.1 + 0.2 + 0.1·0.2 = 0.32, not 0.3. + assert!((out.bound.evaluate().unwrap() - 0.32).abs() < 1e-12); + assert_eq!(out.metric, ErrorMetric::RelativeValue); + } + + #[test] + fn relative_error_without_sign_knowledge_is_rejected() { + let err = DefaultAccuracyModel + .propagate( + &CompositionOperator::ApproximateAggregate, + &[rel(0.1)], + Some(&rel(0.2)), + &PropagationStats::default(), + ) + .unwrap_err(); + assert!(matches!(err, AccuracyError::UnsupportedComposition { .. })); + } + + #[test] + fn incompatible_metrics_are_rejected_not_treated_as_exact() { + // HLL cardinality error under a CMS frequency guarantee. + let err = DefaultAccuracyModel + .propagate( + &CompositionOperator::ApproximateAggregate, + &[with_metric(ErrorMetric::Cardinality, 0.01)], + Some(&with_metric(ErrorMetric::Frequency, 0.01)), + &PropagationStats::default(), + ) + .unwrap_err(); + assert!(matches!( + err, + AccuracyError::UnsupportedComposition { + input_metrics, + local_metric: Some(ErrorMetric::Frequency), + .. + } if input_metrics == vec![ErrorMetric::Cardinality] + )); + // Quantile rank error under value-additive logic. + let err = DefaultAccuracyModel + .propagate( + &CompositionOperator::ApproximateAggregate, + &[with_metric(ErrorMetric::Rank, 0.01)], + Some(&abs(0.01, 0.0)), + &PropagationStats::default(), + ) + .unwrap_err(); + assert!(matches!(err, AccuracyError::UnsupportedComposition { .. })); + } + + #[test] + fn same_metric_rank_over_rank_has_no_registered_rule() { + let err = DefaultAccuracyModel + .propagate( + &CompositionOperator::ApproximateAggregate, + &[with_metric(ErrorMetric::Rank, 0.01)], + Some(&with_metric(ErrorMetric::Rank, 0.01)), + &PropagationStats::default(), + ) + .unwrap_err(); + assert!(matches!(err, AccuracyError::UnsupportedComposition { .. })); + } + + #[test] + fn lipschitz_scales_the_input_bound() { + let out = DefaultAccuracyModel + .propagate( + &CompositionOperator::Lipschitz { constant: 3.0 }, + &[abs(0.1, 0.01)], + Some(&abs(0.05, 0.02)), + &PropagationStats::default(), + ) + .unwrap(); + assert!((out.bound.evaluate().unwrap() - 0.35).abs() < 1e-12); + assert!((out.failure_probability.evaluate().unwrap() - 0.03).abs() < 1e-12); + } + + #[test] + fn exact_sum_over_approximate_sums_bounds_and_keeps_unknown_row_count_unknown() { + let out = DefaultAccuracyModel + .propagate( + &CompositionOperator::ExactSum, + &[abs(0.1, 0.01)], + None, + &PropagationStats::default(), + ) + .unwrap(); + assert_eq!(out.metric, ErrorMetric::AbsoluteValue); + assert_eq!( + out.bound.evaluate(), + None, + "unknown row count stays unknown" + ); + assert!(out.provenance.iter().any(|s| matches!( + s, + GuaranteeSource::UnavailableStatistic { statistic } if statistic == "input_row_count" + ))); + assert!(!DefaultAccuracyModel.satisfies(&out, &AccuracyTarget::Epsilon(1.0))); + + let known = PropagationStats { + input_row_count: Some(4), + ..Default::default() + }; + let out = DefaultAccuracyModel + .propagate( + &CompositionOperator::ExactSum, + &[abs(0.1, 0.01)], + None, + &known, + ) + .unwrap(); + assert!((out.bound.evaluate().unwrap() - 0.4).abs() < 1e-12); + assert!((out.failure_probability.evaluate().unwrap() - 0.04).abs() < 1e-12); + } + + #[test] + fn exact_extremum_takes_the_max_bound() { + let known = PropagationStats { + input_row_count: Some(2), + ..Default::default() + }; + let out = DefaultAccuracyModel + .propagate( + &CompositionOperator::ExactExtremum, + &[abs(0.1, 0.01), abs(0.3, 0.01)], + None, + &known, + ) + .unwrap(); + assert!((out.bound.evaluate().unwrap() - 0.3).abs() < 1e-12); + assert!((out.failure_probability.evaluate().unwrap() - 0.04).abs() < 1e-12); + } + + #[test] + fn topk_selection_is_unsupported() { + let err = DefaultAccuracyModel + .propagate( + &CompositionOperator::TopKSelection, + &[abs(0.1, 0.01)], + None, + &PropagationStats::default(), + ) + .unwrap_err(); + assert!(matches!(err, AccuracyError::UnsupportedComposition { .. })); + } + + #[test] + fn local_guarantee_inverts_the_sizing_formulas() { + use crate::replacement::default_size_params; + use asap_types::pre_asap::agg_intent::{default_cardinality, default_quantile}; + + let q = default_quantile(0.99); + let params = default_size_params(SketchAlgorithm::Kll, &q, 0.01, 0.01); + let g = DefaultAccuracyModel + .local_guarantee( + &SummaryFamilyType::Sketch( + SketchKind::new(SketchAlgorithm::Kll, params), + GroupingStrategy::default(), + ), + &SketchQuery::Quantile { q: 0.99 }, + ) + .unwrap(); + assert_eq!(g.metric, ErrorMetric::Rank); + assert!(DefaultAccuracyModel.satisfies(&g, &AccuracyTarget::Epsilon(0.01))); + assert_eq!(g.approximate_layer_count(), 1); + + let c = default_cardinality(); + let params = default_size_params(SketchAlgorithm::Hll, &c, 0.01, 0.01); + let g = DefaultAccuracyModel + .local_guarantee( + &SummaryFamilyType::Sketch( + SketchKind::new(SketchAlgorithm::Hll, params), + GroupingStrategy::default(), + ), + &SketchQuery::Cardinality, + ) + .unwrap(); + assert_eq!(g.metric, ErrorMetric::Cardinality); + assert!(DefaultAccuracyModel.satisfies(&g, &AccuracyTarget::Epsilon(0.01))); + + let params = default_size_params(SketchAlgorithm::Cms, &c, 0.01, 0.001); + let g = DefaultAccuracyModel + .local_guarantee( + &SummaryFamilyType::Sketch( + SketchKind::new(SketchAlgorithm::Cms, params), + GroupingStrategy::default(), + ), + &SketchQuery::Cardinality, + ) + .unwrap(); + assert_eq!(g.metric, ErrorMetric::Frequency); + assert!(DefaultAccuracyModel.satisfies( + &g, + &AccuracyTarget::EpsilonDelta { + epsilon: 0.01, + delta: 0.001 + } + )); + } + + #[test] + fn satisfies_is_fail_closed_on_unknowns_and_exact() { + let unknown = ResultGuarantee { + bound: BoundExpr::Unknown { + statistic: "x".into(), + }, + ..abs(0.0, 0.0) + }; + assert!(!DefaultAccuracyModel.satisfies(&unknown, &AccuracyTarget::Epsilon(1.0))); + assert!(!DefaultAccuracyModel.satisfies(&abs(0.0, 0.01), &AccuracyTarget::Exact)); + assert!( + DefaultAccuracyModel.satisfies(&ResultGuarantee::exact("x"), &AccuracyTarget::Exact) + ); + } + + #[test] + fn equal_split_respects_the_root_epsilon_and_delta() { + let target = AccuracyTarget::EpsilonDelta { + epsilon: 0.1, + delta: 0.02, + }; + let shape = CompositionShape { + metric: ErrorMetric::AbsoluteValue, + 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| match t { + AccuracyTarget::EpsilonDelta { epsilon, delta } => (*epsilon, *delta), + other => panic!("unexpected {other:?}"), + }) + .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::EpsilonDelta { + epsilon: 0.05, + delta: 0.01 + }) + ); + + // Multiplicative composition: (1+ε_i)^2 = 1+ε, not 2ε_i = ε. + let rel_shape = CompositionShape { + metric: ErrorMetric::RelativeValue, + approximate_layer_count: 2, + }; + let allocations = + EqualSplitAllocator.allocations(&AccuracyTarget::Epsilon(0.21), &rel_shape); + let AccuracyTarget::Epsilon(e) = allocations[0].layers[0] else { + panic!() + }; + assert!((e - 0.1).abs() < 1e-12); + + assert!(EqualSplitAllocator + .allocations(&AccuracyTarget::Exact, &shape) + .is_empty()); + } +} diff --git a/crates/asap-aware-mapping/src/cost_model.rs b/crates/asap-aware-mapping/src/cost_model.rs index 82a3e97..0e40d65 100644 --- a/crates/asap-aware-mapping/src/cost_model.rs +++ b/crates/asap-aware-mapping/src/cost_model.rs @@ -681,6 +681,7 @@ mod tests { fields: vec![], time_index: None, }, + guarantee: None, }), family: family.clone(), col: asap_types::pre_asap::expr_ir::ColumnRef::Named("value".into()), @@ -695,6 +696,7 @@ mod tests { }], time_index: None, }, + guarantee: None, } } diff --git a/crates/asap-aware-mapping/src/grouping.rs b/crates/asap-aware-mapping/src/grouping.rs index c5485f4..0e3f294 100644 --- a/crates/asap-aware-mapping/src/grouping.rs +++ b/crates/asap-aware-mapping/src/grouping.rs @@ -275,6 +275,11 @@ fn with_grouping(node: Rc, grouping: GroupingStrategy) -> Rc, grouping: GroupingStrategy) -> Rc, + pub rejected: Vec, +} + /// A replacement strategy: given a [`TargetSubDAG`], does this strategy have /// an opinion on it at all (`matches`), and if so, every semantically valid /// replacement (`replacements`)? @@ -498,6 +535,19 @@ pub trait ReplacementStrategy { /// Reporting "every valid candidate" is this method's whole job; picking /// the best one is a [`CostModel`]'s job, out of scope here. fn replacements(&self, target: &TargetSubDAG<'_>) -> Vec; + + /// [`replacements`](Self::replacements) plus the accuracy-illegal + /// candidates this strategy refused to propose (issue #172). Default: + /// every candidate from `replacements`, no rejections — a strategy that + /// never performs an accuracy check need not override this. + /// [`search_workload_with`] calls this (not `replacements`) so the + /// rejections land in [`MemoGroup::rejected`]. + fn propose(&self, target: &TargetSubDAG<'_>) -> Proposals { + Proposals { + candidates: self.replacements(target), + rejected: Vec::new(), + } + } } // ── Implementation: how one AggIntent may be realised ─────────────────────── @@ -1055,6 +1105,33 @@ fn saturating_ceil(x: f64, lo: u32, hi: u32) -> u32 { /// `DefaultCostModel` is a unit struct with no state, so one instance serves /// every caller. static DEFAULT_COST_MODEL: DefaultCostModel = DefaultCostModel; +static DEFAULT_ACCURACY_MODEL: DefaultAccuracyModel = DefaultAccuracyModel; +static DEFAULT_ALLOCATOR: EqualSplitAllocator = EqualSplitAllocator; + +/// The three deployment-pluggable models one candidate construction +/// consults, bundled so the construction path threads one argument rather +/// than three. `cost` ranks and sizes; `accuracy` and `allocator` decide +/// 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 Models<'a> { + pub cost: &'a dyn CostModel, + pub accuracy: &'a dyn AccuracyModel, + pub allocator: &'a dyn AccuracyBudgetAllocator, +} + +impl<'a> Models<'a> { + /// `cost` with the built-in [`DefaultAccuracyModel`]/ + /// [`EqualSplitAllocator`] — 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, + } + } +} /// Wraps [`implementations_for_with`]'s exhaustive, ranked list directly: for /// a bindable `Aggregate`, every valid candidate summary realization as its @@ -1065,8 +1142,17 @@ static DEFAULT_COST_MODEL: DefaultCostModel = DefaultCostModel; /// [`SketchAlgorithmStrategy::new`] — so a deployment-specific cost model's /// other hooks (`size_params`, `realize_extension`, `readout_extension`) are /// still consulted while binding each candidate. +/// +/// The one thing that *does* drop a candidate is accuracy legality (issue +/// #172), decided by the [`AccuracyModel`] — never by the cost model: a +/// sketch over an approximate child is proposed only if its composed +/// guarantee has a sound propagation rule and satisfies the node's own +/// `AccuracyTarget`; otherwise it is reported through +/// [`ReplacementStrategy::propose`] as a [`RejectedCandidate`]. See +/// [`crate::accuracy`]'s module docs for the rules and the precedence +/// between root and per-node targets. pub struct SketchAlgorithmStrategy<'a> { - cost_model: &'a dyn CostModel, + models: Models<'a>, } impl SketchAlgorithmStrategy<'static> { @@ -1074,7 +1160,7 @@ impl SketchAlgorithmStrategy<'static> { /// what a deployment gets with no custom cost model plugged in. pub fn default_cost_model() -> Self { Self { - cost_model: &DEFAULT_COST_MODEL, + models: Models::with_default_accuracy(&DEFAULT_COST_MODEL), } } } @@ -1082,39 +1168,199 @@ impl SketchAlgorithmStrategy<'static> { impl<'a> SketchAlgorithmStrategy<'a> { /// A strategy that ranks/binds via `cost_model` instead of the built-in /// static preference order — the same customization point - /// [`implementations_for_with`] already offers. + /// [`implementations_for_with`] already offers. Accuracy legality stays + /// with the built-in [`DefaultAccuracyModel`]/[`EqualSplitAllocator`]. pub fn new(cost_model: &'a dyn CostModel) -> Self { - Self { cost_model } + Self { + models: Models::with_default_accuracy(cost_model), + } } -} -impl ReplacementStrategy for SketchAlgorithmStrategy<'_> { - fn matches(&self, target: &TargetSubDAG<'_>) -> bool { - bindable_intent(target.root).is_some() + /// A strategy with every model plugged in explicitly: `cost_model` for + /// ranking/sizing, `accuracy_model` for guarantee derivation/propagation/ + /// satisfaction, `allocator` for end-to-end budget splits. One model + /// never overrides another: legality is settled by `accuracy_model` + /// before `cost_model` ranks what is left. + pub fn with_models( + cost_model: &'a dyn CostModel, + accuracy_model: &'a dyn AccuracyModel, + allocator: &'a dyn AccuracyBudgetAllocator, + ) -> Self { + Self { + models: Models { + cost: cost_model, + accuracy: accuracy_model, + allocator, + }, + } } - fn replacements(&self, target: &TargetSubDAG<'_>) -> Vec { - let Some(intent) = bindable_intent(target.root) else { - return Vec::new(); + pub(crate) fn from_models(models: Models<'a>) -> Self { + Self { models } + } + + /// The whole enumeration for one target, with `intent_override` + /// substituting the target's own intent (only ever its `AccuracyTarget` + /// differs — see [`realize_child_with`]). + fn propose_with(&self, root: &Rc, intent_override: Option<&AggIntent>) -> Proposals { + let mut proposals = Proposals::default(); + let Some(declared) = bindable_intent(root) else { + return proposals; }; + let intent = intent_override.unwrap_or(declared); + let models = self.models; + + // Is the child approximate? Probed once, up front: a candidate over + // an approximate child needs the end-to-end budget split across both + // layers, which changes which candidates exist at all. + let child_layers = aggregate_child(root) + .and_then(|child| realize_child_with(child, models, None).ok()) + .and_then(|child| { + child + .guarantee + .as_ref() + .filter(|g| !g.is_exact()) + .map(ResultGuarantee::approximate_layer_count) + }); + // `implementations_for_with` is already exhaustive and ranked — no // separate dispatch needed here. Only `Sketch` has more than one // candidate in practice (every other variant's own dispatch produces // exactly one `Implementation`), but this loop doesn't need to know // that; it just constructs whatever the list contains. - implementations_for_with(intent, self.cost_model) - .into_iter() - .filter_map(|implementation| { - let rationale = describe_implementation(intent, &implementation); - let node = construct_summary(target.root, implementation, self.cost_model).ok()?; - Some(ReplacementSubDAG { + for implementation in implementations_for_with(intent, models.cost) { + let rationale = describe_implementation(intent, &implementation); + // The as-declared composition: every layer sized to its own + // declared `AccuracyTarget`. Legal iff the composed guarantee + // satisfies this node's target — a front end copying one target + // onto every node does not make that so. + proposals.record( + rationale.clone(), + construct_summary_with(root, intent, implementation.clone(), models, None, None), + ); + + // Budget-split alternatives (issue #172, PR 2): re-size this + // layer and the approximate child under each allocation of this + // node's target across every approximate layer. + let (Some(child_layers), Implementation::Sketch(kind), Some(target)) = + (child_layers, &implementation, accuracy_target(intent)) + else { + continue; + }; + let Some(readout_query) = aggregate_child(root) + .and_then(|child| child.output_schema().ok()) + .map(|schema| readout(intent, &summarised_column(intent, &schema), models.cost)) + else { + continue; + }; + let family = SummaryFamilyType::Sketch(kind.clone(), GroupingStrategy::default()); + let Some(local) = models.accuracy.local_guarantee(&family, &readout_query) else { + continue; + }; + let shape = CompositionShape { + metric: local.metric, + approximate_layer_count: 1 + child_layers, + }; + let allocations = models.allocator.allocations(target, &shape); + if allocations.is_empty() { + proposals.rejected.push(RejectedCandidate { strategy: "SketchAlgorithmStrategy", - replacement: Replacement::Summary(node), - provenance: ReplacementProvenance::SummaryImplementation, - rationale, - }) - }) - .collect() + description: rationale.clone(), + error: AccuracyError::NoLegalAllocation { + target: target.clone(), + layer_count: shape.approximate_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 = Implementation::Sketch(SketchKind::new( + kind.algorithm().clone(), + models + .cost + .size_params(kind.algorithm().clone(), intent, eps, delta), + )); + // Identical to the as-declared composition already recorded + // above — nothing new to propose. + if resized == implementation && 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 subtree \ + {inner_target:?})", + allocation.allocator, shape.approximate_layer_count + ), + construct_summary_with( + root, + intent, + resized, + models, + inner_target.as_ref(), + Some(note), + ), + ); + } + } + proposals + } +} + +impl Proposals { + /// File one construction attempt: a legal node becomes a candidate, an + /// [`ImplementError::Accuracy`] becomes a [`RejectedCandidate`], and a + /// schema-derivation failure is skipped exactly as it always was. + fn record(&mut self, rationale: String, built: Result, ImplementError>) { + match built { + Ok(node) => self.candidates.push(ReplacementSubDAG { + strategy: "SketchAlgorithmStrategy", + replacement: Replacement::Summary(node), + provenance: ReplacementProvenance::SummaryImplementation, + rationale, + }), + Err(ImplementError::Accuracy(error)) => self.rejected.push(RejectedCandidate { + strategy: "SketchAlgorithmStrategy", + description: rationale, + error, + }), + Err(ImplementError::Schema(_)) => {} + } + } +} + +/// The `child` of a [`bindable_intent`]-shaped `Aggregate`. +fn aggregate_child(node: &QueryExpr) -> Option<&Rc> { + match node { + QueryExpr::Aggregate { child, .. } => Some(child), + _ => None, + } +} + +impl ReplacementStrategy for SketchAlgorithmStrategy<'_> { + fn matches(&self, target: &TargetSubDAG<'_>) -> bool { + bindable_intent(target.root).is_some() + } + + fn replacements(&self, target: &TargetSubDAG<'_>) -> Vec { + self.propose(target).candidates + } + + fn propose(&self, target: &TargetSubDAG<'_>) -> Proposals { + self.propose_with(target.root, None) } } @@ -1205,9 +1451,32 @@ pub(crate) fn realize_child( root: &Rc, cost_model: &dyn CostModel, ) -> Result, ImplementError> { - let target = TargetSubDAG::new(root); - match SketchAlgorithmStrategy::new(cost_model) - .replacements(&target) + realize_child_with(root, Models::with_default_accuracy(cost_model), None) +} + +/// [`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 +/// 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 +/// budget. A child whose declared target is `Exact` keeps it: an allocation +/// never approximates something the caller declared exact. +pub(crate) fn realize_child_with( + root: &Rc, + models: Models<'_>, + end_to_end_target: Option<&AccuracyTarget>, +) -> Result, ImplementError> { + let overridden = end_to_end_target.and_then(|target| { + let declared = bindable_intent(root)?; + match accuracy_target(declared) { + Some(AccuracyTarget::Exact) | None => None, + Some(_) => Some(override_accuracy(declared, target)), + } + }); + match SketchAlgorithmStrategy::from_models(models) + .propose_with(root, overridden.as_ref()) + .candidates .into_iter() .next() { @@ -1223,12 +1492,28 @@ pub(crate) fn realize_child( } // No candidate at all: `root` isn't `bindable_intent` shape (or its // intent has no realization `implementations_for_with` can't - // produce — never happens, that match is exhaustive) — the same - // conservative fallback `SketchAlgorithmStrategy::matches` uses. + // produce — never happens, that match is exhaustive), or every + // candidate was accuracy-illegal — either way the same conservative + // fallback `SketchAlgorithmStrategy::matches` uses: keep the + // pre-ASAP subtree, executed exactly. None => keep_pre_asap(root), } } +/// `intent` with its `AccuracyTarget` replaced by `target` — a no-op for an +/// intent that carries none (see [`accuracy_target`]). +fn override_accuracy(intent: &AggIntent, target: &AccuracyTarget) -> AggIntent { + let mut out = intent.clone(); + match &mut out { + AggIntent::Quantile { accuracy, .. } + | AggIntent::Cardinality { accuracy, .. } + | AggIntent::Count { accuracy } + | AggIntent::TopK { accuracy, .. } => *accuracy = target.clone(), + _ => {} + } + out +} + /// Wrap an unrewritten pre-ASAP subtree, lifting its schema with every column /// `SummaryFamilyType::Plain`. `pub` so a caller can fall back to this /// explicitly — e.g. when `SketchAlgorithmStrategy::replacements()` returns no @@ -1244,6 +1529,9 @@ fn keep_pre_asap_rc(expr: Rc) -> Result, ImplementErr Ok(Rc::new(SummaryNode { expr: SummaryExpr::KeepPreAsap(expr), schema: lift(&schema), + // A kept pre-ASAP subtree is executed exactly by the runtime + // (`Implementation::PassThrough`'s contract) — zero error. + guarantee: Some(ResultGuarantee::exact("KeepPreAsap")), })) } @@ -1293,21 +1581,48 @@ pub(crate) fn construct_summary( expr: &QueryExpr, implementation: Implementation, cost_model: &dyn CostModel, +) -> Result, ImplementError> { + let models = Models::with_default_accuracy(cost_model); + match bindable_intent(expr) { + Some(intent) => construct_summary_with(expr, intent, implementation, models, None, None), + None => keep_pre_asap_rc(Rc::new(expr.clone())), + } +} + +/// [`construct_summary`] with every model explicit (issue #172). `intent` +/// is `expr`'s own [`bindable_intent`], or a copy of it with an allocated +/// `AccuracyTarget` substituted (see [`realize_child_with`]). +/// `child_target`, when set, is the end-to-end budget the child subtree is +/// re-enumerated under; `allocation` is the provenance note recording the +/// split that produced both. `Err(ImplementError::Accuracy)` is the +/// fail-closed answer for a composition with no sound rule or one that +/// misses `intent`'s target. +pub(crate) fn construct_summary_with( + expr: &QueryExpr, + intent: &AggIntent, + implementation: Implementation, + models: Models<'_>, + child_target: Option<&AccuracyTarget>, + allocation: Option, ) -> Result, ImplementError> { if let QueryExpr::Aggregate { - reduction, - measures, - having, - child, - .. + reduction, child, .. } = expr { - // The bindable shape: exactly one intent, no HAVING. (Multi-intent - // nodes and HAVING stay logical — see `bindable_intent`.) - if let ([intent], None) = (measures.as_slice(), having) { + // `bindable_intent` already established the shape: exactly one + // intent, no HAVING. (Multi-intent nodes and HAVING stay logical.) + if bindable_intent(expr).is_some() { if let Some((family, estimate)) = summary_family(implementation) { return construct_summary_agg( - expr, reduction, intent, child, family, estimate, cost_model, + expr, + reduction, + intent, + child, + family, + estimate, + models, + child_target, + allocation, ); } } @@ -1352,7 +1667,9 @@ fn construct_summary_agg( child: &Rc, family: SummaryFamilyType, estimate: bool, - cost_model: &dyn CostModel, + models: Models<'_>, + child_target: Option<&AccuracyTarget>, + allocation: Option, ) -> Result, ImplementError> { let child_schema = child.output_schema()?; // The single canonical pre-ASAP derivation (per-series vs cross-series, @@ -1367,13 +1684,29 @@ fn construct_summary_agg( let state_idx = summary_col_index(&out_schema, &by, per_series); let col = summarised_column(intent, &child_schema); - let query = estimate.then(|| readout(intent, &col, cost_model)); + let query = estimate.then(|| readout(intent, &col, models.cost)); let mut state_schema = lift(&out_schema); if let Some(field) = state_schema.fields.get_mut(state_idx) { field.dtype = family.clone(); } + let bound_child = realize_child_with(child, models, child_target)?; + + // ── Guarantee (issue #172) ────────────────────────────────────────── + // Derived *before* the node exists, so an illegal composition is never + // materialized: the local guarantee of this family's readout (or exact + // accumulator) composed over the child's, under the operator this + // family applies to the child's values. + let guarantee = compose_guarantee( + &family, + query.as_ref(), + &bound_child, + intent, + models.accuracy, + allocation, + )?; + // `reduction` is carried onto `SummaryAgg` verbatim — not flattened to a // bare `Vec` — so `SummaryExecutor::find_candidates` can tell // a genuine empty-`by` reduction apart from a per-entity shape with no @@ -1381,13 +1714,16 @@ fn construct_summary_agg( // single place that decides this; nothing downstream re-derives it. let agg = Rc::new(SummaryNode { expr: SummaryExpr::SummaryAgg { - child: realize_child(child, cost_model)?, + child: bound_child, family, col, reduction: reduction.clone(), grouping: GroupingStrategy::default(), }, schema: state_schema, + // Summary *state* carries no caller-visible guarantee; only a + // finalized value does. An exact accumulator's state is its value. + guarantee: if estimate { None } else { guarantee.clone() }, }); match query { // The readout: downstream of the estimate the schema is the plain @@ -1399,11 +1735,102 @@ fn construct_summary_agg( query, }, schema: lift(&out_schema), + guarantee, })), None => Ok(agg), } } +/// The guarantee of the value a `family` node produces over `child` — +/// [`AccuracyModel::propagate`] under the [`CompositionOperator`] this family +/// applies to its child's values — checked against `intent`'s own +/// `AccuracyTarget` whenever the child is approximate (an approximate +/// parent over an exact child is sized to that target by construction and +/// is not re-checked here, so single-layer behavior is unchanged; see +/// [`crate::accuracy`]'s precedence rules). `Ok(None)` is "no error model" +/// (an approximate family the model has no local guarantee for, over an +/// exact child) — unknown, never exact. +fn compose_guarantee( + family: &SummaryFamilyType, + query: Option<&PostAsapSketchQuery>, + child: &SummaryNode, + intent: &AggIntent, + accuracy: &dyn AccuracyModel, + allocation: Option, +) -> Result, AccuracyError> { + let (op, local) = match (family, query) { + (SummaryFamilyType::ExactAggregate(kind, _), _) => { + let op = match kind { + ExactKind::Sum => CompositionOperator::ExactSum, + ExactKind::MinMax => CompositionOperator::ExactExtremum, + // A row count does not depend on the rows' values: exact + // regardless of the child's own error. + ExactKind::Count => { + return Ok(Some(ResultGuarantee::exact( + "ExactAggregate(Count): row count is independent of input values", + ))) + } + // Counter-reset detection over perturbed values has no finite + // Lipschitz constant — over an approximate child this is a + // deterministic transform with no registered rule. + ExactKind::Increase | ExactKind::Rate => CompositionOperator::Lipschitz { + constant: f64::INFINITY, + }, + }; + ( + op, + Some(ResultGuarantee::exact(format!("ExactAggregate({kind:?})"))), + ) + } + (_, Some(query)) => ( + CompositionOperator::ApproximateAggregate, + accuracy.local_guarantee(family, query), + ), + (_, None) => (CompositionOperator::ApproximateAggregate, None), + }; + let Some(input) = child.guarantee.clone() else { + // A child with no guarantee at all is an unknown quantity, which + // nothing can be composed over (a `Sample` readout, say) — unless + // this node is itself the unknown family, in which case it inherits + // "unknown" rather than fabricating a guarantee for its child. + return match local { + Some(_) => Err(AccuracyError::MissingInputGuarantee { + operator: op, + input_index: 0, + }), + None => Ok(None), + }; + }; + if local.is_none() && input.is_exact() { + return Ok(None); + } + let mut guarantee = accuracy.propagate( + &op, + std::slice::from_ref(&input), + local.as_ref(), + &PropagationStats::default(), + )?; + if let Some(note) = allocation { + guarantee.provenance.push(note); + } + if let Some(target) = accuracy_target(intent) { + guarantee.provenance.push(GuaranteeSource::AccuracyTarget { + target: target.clone(), + }); + // Approximate over approximate: the composed guarantee must meet the + // target this node's value was requested at. + if !input.is_exact() && !accuracy.satisfies(&guarantee, target) { + return Err(AccuracyError::TargetNotSatisfied { + metric: guarantee.metric, + bound: guarantee.bound.evaluate(), + failure_probability: guarantee.failure_probability.evaluate(), + target: target.clone(), + }); + } + } + Ok(Some(guarantee)) +} + /// Index of the summary-state column in the aggregate's output schema: /// cross-series output is `by ++ [agg]` (the column after the keys); /// a per-series reduction keeps every label and replaces the sample value @@ -1580,6 +2007,12 @@ pub struct MemoGroup { /// order (not ranked — see [`PlanSpace::cost_sorted`] for the ranked /// view). pub candidates: Vec, + /// Every candidate a strategy considered for `target` but refused on + /// accuracy-legality grounds (issue #172), plus any `candidates` entry + /// the root-target check ([`search_workload_with_targets`]) moved here. + /// Never ranked — [`PlanSpace::cost_sorted`]/[`PlanSpace::global_selection`] + /// read only `candidates`, so a [`CostModel`] cannot resurrect one. + pub rejected: Vec, } impl MemoGroup { @@ -1588,6 +2021,7 @@ impl MemoGroup { target, consumer_count, candidates: Vec::new(), + rejected: Vec::new(), } } @@ -2519,6 +2953,93 @@ pub fn search_workload_with<'s, Id>( search_cse_workload_with(cse_workload(roots), strategies) } +/// [`search_workload_with`] plus a per-root end-to-end `AccuracyTarget` +/// (issue #172) — the workload's `QueryRequirements.accuracy`, threaded +/// alongside each root. After the search, every root that carries a target +/// has its group's bound [`Replacement::Summary`] candidates checked with +/// `accuracy_model`'s [`AccuracyModel::satisfies`]: a candidate whose +/// guarantee is absent (unknown) or misses the target is moved from +/// [`MemoGroup::candidates`] to [`MemoGroup::rejected`] *before* +/// [`PlanSpace::cost_sorted`]/[`PlanSpace::global_selection`] ever rank the +/// group, so a `CostModel` cannot pick it. A `KeepPreAsap` candidate is +/// exact and always survives — the raw/pre-ASAP alternative is what an +/// unsatisfiable root keeps. Logical [`Replacement::Rewrite`] candidates +/// are not bound values and are left alone; the targets *inside* a rewrite +/// are their own groups. +/// +/// Precedence against per-node `AggIntent.accuracy` is documented in +/// [`crate::accuracy`]'s module docs. +pub fn search_workload_with_targets<'s, Id>( + roots: Vec<(Id, Rc, Option)>, + strategies: &[Box], + accuracy_model: &dyn AccuracyModel, +) -> PlanSpace { + let mut targets = Vec::with_capacity(roots.len()); + let roots = roots + .into_iter() + .map(|(id, root, target)| { + targets.push(target); + (id, root) + }) + .collect(); + let mut space = search_workload_with(roots, strategies); + // `cse_workload` preserves root order, so targets zip by position. + let root_ptrs: Vec<(*const QueryExpr, AccuracyTarget)> = space + .roots + .iter() + .zip(targets) + .filter_map(|((_, root), target)| target.map(|t| (Rc::as_ptr(root), t))) + .collect(); + for (ptr, target) in root_ptrs { + let Some(group) = space.groups.get_mut(&ptr) else { + continue; + }; + let (legal, illegal): (Vec<_>, Vec<_>) = + group + .candidates + .drain(..) + .partition(|candidate| match &candidate.replacement { + Replacement::Summary(node) => node + .guarantee + .as_ref() + .is_some_and(|g| accuracy_model.satisfies(g, &target)), + Replacement::Rewrite(_) => true, + }); + group.candidates = legal; + group.rejected.extend(illegal.into_iter().map(|candidate| { + let (metric, bound, failure_probability) = match &candidate.replacement { + Replacement::Summary(node) => node + .guarantee + .as_ref() + .map(|g| { + ( + g.metric, + g.bound.evaluate(), + g.failure_probability.evaluate(), + ) + }) + .unwrap_or(( + asap_types::post_asap::ErrorMetric::AbsoluteValue, + None, + None, + )), + Replacement::Rewrite(_) => unreachable!("rewrites are never rejected here"), + }; + RejectedCandidate { + strategy: candidate.strategy, + description: format!("{} (root end-to-end target check)", candidate.rationale), + error: AccuracyError::TargetNotSatisfied { + metric, + bound, + failure_probability, + target: target.clone(), + }, + } + })); + } + space +} + fn cse_workload(roots: Vec<(Id, Rc)>) -> Vec<(Id, Rc)> { // `share_common_subtrees` wants owned `QueryExpr`s, not already-`Rc` // roots — the same `Rc::try_unwrap`-with-clone-fallback pattern @@ -2593,15 +3114,19 @@ fn search_cse_workload_with<'s, Id>( let target = TargetSubDAG::with_consumer_count(&root, consumer_count); let mut proposed = Vec::new(); + let mut rejected = Vec::new(); for strategy in strategies { if strategy.matches(&target) { let name = strategy.name(); - proposed.extend(strategy.replacements(&target).into_iter().map( - |mut candidate| { - candidate.strategy = name; - candidate - }, - )); + let proposals = strategy.propose(&target); + proposed.extend(proposals.candidates.into_iter().map(|mut candidate| { + candidate.strategy = name; + candidate + })); + rejected.extend(proposals.rejected.into_iter().map(|mut rejection| { + rejection.strategy = name; + rejection + })); } } if rollup_strategy.matches(&target) { @@ -2635,6 +3160,7 @@ fn search_cse_workload_with<'s, Id>( for candidate in proposed { group.add_candidate(candidate); } + group.rejected.extend(rejected); } // Any pointer `discover_new_descendant_targets` appended to `order` @@ -3562,10 +4088,20 @@ mod tests { fn enumerating_the_targets_candidates_does_not_leak_into_a_nested_aggregate() { // outer: quantile(0.99, ...) over inner: quantile(0.5, m) — both // Quantile, so both share the [Kll, DDSketch] candidate list. + // + // Rank-over-rank has no registered rule in `DefaultAccuracyModel` + // (issue #172 — see `approximate_over_approximate_is_rejected_by_default`), + // so this test injects `RankAdditiveModel` to admit the composition + // and keep exercising the per-node enumeration property it is about. let inner = agg(vec![2], default_quantile(0.5), metric_scan(&["job"])); let outer = Rc::new(agg(vec![], default_quantile(0.99), inner)); let target = TargetSubDAG::new(&outer); - let replacements = SketchAlgorithmStrategy::default_cost_model().replacements(&target); + let replacements = SketchAlgorithmStrategy::with_models( + &DefaultCostModel, + &RankAdditiveModel, + &EqualSplitAllocator, + ) + .replacements(&target); let ddsketch = replacements .iter() @@ -5424,4 +5960,374 @@ mod tests { }; assert_eq!(col, &ColumnRef::Named("bytes".into())); } + + // ── Accuracy guarantees and fail-closed composition (issue #172) ───── + + use asap_types::post_asap::{BoundExpr, ErrorMetric}; + + /// A test-only `AccuracyModel` that *registers* a rule the default + /// deliberately lacks — a sketch over rank-bounded inputs composes + /// additively, keeping the outer sketch's own metric — so the + /// composition/allocation machinery can be exercised end to end. + /// Everything else delegates to `DefaultAccuracyModel`. + struct RankAdditiveModel; + + impl AccuracyModel for RankAdditiveModel { + fn local_guarantee( + &self, + family: &SummaryFamilyType, + query: &PostAsapSketchQuery, + ) -> Option { + DefaultAccuracyModel.local_guarantee(family, query) + } + + fn propagate( + &self, + op: &CompositionOperator, + inputs: &[ResultGuarantee], + local: Option<&ResultGuarantee>, + stats: &PropagationStats, + ) -> Result { + let rank = |g: &ResultGuarantee| g.is_exact() || g.metric == ErrorMetric::Rank; + if let (CompositionOperator::ApproximateAggregate, true, Some(local)) = + (op, inputs.iter().all(rank), local) + { + let relabel = |g: &ResultGuarantee| ResultGuarantee { + metric: ErrorMetric::AbsoluteValue, + ..g.clone() + }; + let inputs: Vec<_> = inputs.iter().map(relabel).collect(); + let mut out = + DefaultAccuracyModel.propagate(op, &inputs, Some(&relabel(local)), stats)?; + out.metric = local.metric; + return Ok(out); + } + DefaultAccuracyModel.propagate(op, inputs, local, stats) + } + + fn satisfies(&self, guarantee: &ResultGuarantee, target: &AccuracyTarget) -> bool { + DefaultAccuracyModel.satisfies(guarantee, target) + } + } + + fn quantile_eps(q: f64, eps: f64) -> AggIntent { + AggIntent::Quantile { + col: None, + q, + accuracy: AccuracyTarget::Epsilon(eps), + } + } + + /// The `SketchParams::Kll { k }` of the top `SummaryAgg` under `node`. + fn kll_k_of(node: &SummaryNode) -> u32 { + match &node.expr { + SummaryExpr::SummaryEstimate { summary_input, .. } => kll_k_of(summary_input), + SummaryExpr::SummaryAgg { + family: SummaryFamilyType::Sketch(kind, _), + .. + } => match kind.params() { + SketchParams::Kll { k } => *k, + other => panic!("expected KLL params, got {other:?}"), + }, + other => panic!("expected a sketch SummaryAgg, got {other:?}"), + } + } + + fn summary_child(node: &SummaryNode) -> &Rc { + match &node.expr { + SummaryExpr::SummaryEstimate { summary_input, .. } => summary_child(summary_input), + SummaryExpr::SummaryAgg { child, .. } => child, + other => panic!("expected a SummaryAgg, got {other:?}"), + } + } + + #[test] + fn approximate_over_approximate_is_rejected_by_default_not_treated_as_exact() { + // quantile(0.99, quantile by (job) (0.5, m)): rank over rank — no + // registered rule, so every outer sketch candidate is refused with a + // typed reason and the raw/pre-ASAP alternative is what remains. + let inner = agg(vec![2], default_quantile(0.5), metric_scan(&["job"])); + let outer = Rc::new(agg(vec![], default_quantile(0.99), inner)); + let proposals = + SketchAlgorithmStrategy::default_cost_model().propose(&TargetSubDAG::new(&outer)); + assert!( + proposals.candidates.is_empty(), + "no outer sketch may be proposed over an approximate child without a rule: {:?}", + proposals.candidates + ); + // Every attempt — the as-declared composition and the equal-split + // re-sizing, for each of KLL/DDSketch — is refused for the same + // typed reason: no rule, whatever the budget. + assert_eq!(proposals.rejected.len(), 4, "{:?}", proposals.rejected); + for rejection in &proposals.rejected { + assert!( + matches!( + &rejection.error, + AccuracyError::UnsupportedComposition { input_metrics, .. } + if input_metrics == &vec![ErrorMetric::Rank] + ), + "{:?}", + rejection.error + ); + } + // Fallback keeps the whole subtree pre-ASAP — executed exactly. + let realized = realize_child(&outer, &DefaultCostModel).unwrap(); + assert!(matches!(realized.expr, SummaryExpr::KeepPreAsap(_))); + assert!(realized + .guarantee + .as_ref() + .is_some_and(ResultGuarantee::is_exact)); + + // Cross-metric: a quantile over a cardinality estimate. + let inner = agg(vec![2], default_cardinality(), metric_scan(&["job"])); + let outer = Rc::new(agg(vec![], default_quantile(0.99), inner)); + let proposals = + SketchAlgorithmStrategy::default_cost_model().propose(&TargetSubDAG::new(&outer)); + assert!(proposals.candidates.is_empty()); + assert!(proposals.rejected.iter().all(|r| matches!( + &r.error, + AccuracyError::UnsupportedComposition { input_metrics, .. } + if input_metrics == &vec![ErrorMetric::Cardinality] + ))); + } + + #[test] + fn exact_child_contributes_zero_error() { + // quantile(0.9, sum by (job) (m)): KLL over an exact Sum accumulator + // — the readout's guarantee is exactly KLL's own local guarantee. + let inner = agg(vec![2], AggIntent::Sum { col: None }, metric_scan(&["job"])); + let outer = agg(vec![], default_quantile(0.9), inner); + let root = realize(&outer).unwrap(); + let guarantee = root + .guarantee + .as_ref() + .expect("a readout carries a guarantee"); + assert_eq!(guarantee.metric, ErrorMetric::Rank); + assert_eq!(guarantee.bound.evaluate(), Some(2.0 / 200.0)); + assert_eq!(guarantee.approximate_layer_count(), 1); + assert!(guarantee.provenance.iter().any(|s| matches!( + s, + GuaranteeSource::ChildGuarantee { guarantee, .. } if guarantee.is_exact() + ))); + assert!(guarantee.provenance.iter().any(|s| matches!( + s, + GuaranteeSource::CompositionStep { rule, .. } if rule == "exact_input" + ))); + // The sketch *state* node carries no guarantee; the exact + // accumulator's state is its value and does. + let SummaryExpr::SummaryEstimate { summary_input, .. } = &root.expr else { + panic!() + }; + assert!(summary_input.guarantee.is_none()); + assert!(summary_child(&root) + .guarantee + .as_ref() + .is_some_and(ResultGuarantee::is_exact)); + } + + #[test] + fn exact_sum_over_approximate_child_keeps_the_row_count_unknown() { + // sum(count_distinct by (job) (m)): an exact sum over HLL estimates + // is representable (Σ B_i) but its bound depends on the group count + // and the true cardinalities — unknown at planning time, so the + // guarantee exists, says what it needs, and satisfies nothing. + let inner = agg(vec![2], default_cardinality(), metric_scan(&["job"])); + let outer = agg(vec![], AggIntent::Sum { col: None }, inner); + let root = realize(&outer).unwrap(); + let guarantee = root + .guarantee + .as_ref() + .expect("an exact sum carries a guarantee"); + assert_eq!(guarantee.metric, ErrorMetric::AbsoluteValue); + assert_eq!(guarantee.bound.evaluate(), None); + assert!(guarantee.provenance.iter().any(|s| matches!( + s, + GuaranteeSource::UnavailableStatistic { statistic } if statistic == "input_row_count" + ))); + assert!(!DefaultAccuracyModel.satisfies(guarantee, &AccuracyTarget::Epsilon(1e9))); + + // count(...) over the same child is exact: a row count does not + // depend on the rows' values. + let inner = agg(vec![2], default_cardinality(), metric_scan(&["job"])); + let outer = agg( + vec![], + AggIntent::Count { + accuracy: AccuracyTarget::Exact, + }, + inner, + ); + let root = realize(&outer).unwrap(); + assert!(root + .guarantee + .as_ref() + .is_some_and(ResultGuarantee::is_exact)); + } + + #[test] + fn equal_split_allocation_makes_a_legal_tighter_candidate_and_rejects_the_declared_one() { + // With a registered rank-additive rule: outer ε=0.1 over inner ε=0.1 + // composes to 0.2 > 0.1 as declared (cheap: k=20 each) — illegal. + // The equal split (0.05 + 0.05) re-sizes both layers to k=40 — + // pricier, and the only legal way to meet the outer target. + let inner = agg(vec![2], quantile_eps(0.5, 0.1), metric_scan(&["job"])); + let outer = Rc::new(agg(vec![], quantile_eps(0.99, 0.1), inner)); + let strategy = SketchAlgorithmStrategy::with_models( + &DefaultCostModel, + &RankAdditiveModel, + &EqualSplitAllocator, + ); + let proposals = strategy.propose(&TargetSubDAG::new(&outer)); + + let declared = proposals + .rejected + .iter() + .filter(|r| { + matches!( + &r.error, + AccuracyError::TargetNotSatisfied { + metric: ErrorMetric::Rank, + bound: Some(b), + target: AccuracyTarget::Epsilon(e), + .. + } if (b - 0.2).abs() < 1e-12 && *e == 0.1 + ) + }) + .count(); + assert_eq!( + declared, 1, + "the as-declared KLL composition is rejected: {:?}", + proposals.rejected + ); + + let kll: Vec<_> = proposals + .candidates + .iter() + .filter_map(|c| match &c.replacement { + Replacement::Summary(node) + if summary_family_algorithm(node) == SketchAlgorithm::Kll => + { + Some(node) + } + _ => None, + }) + .collect(); + assert_eq!( + kll.len(), + 1, + "exactly one legal KLL candidate (the allocated one)" + ); + let node = kll[0]; + assert_eq!(kll_k_of(node), 40, "outer re-sized to ε/2"); + assert_eq!(kll_k_of(summary_child(node)), 40, "inner re-sized to ε/2"); + let guarantee = node.guarantee.as_ref().unwrap(); + assert_eq!(guarantee.metric, ErrorMetric::Rank); + assert!((guarantee.bound.evaluate().unwrap() - 0.1).abs() < 1e-12); + assert_eq!(guarantee.approximate_layer_count(), 2); + assert!(guarantee.provenance.iter().any(|s| matches!( + s, + GuaranteeSource::BudgetAllocation { allocator, layer_count: 2, .. } + if allocator == "EqualSplitAllocator" + ))); + assert!(matches!(guarantee.bound, BoundExpr::Sum { .. })); + // No candidate with the cheaper illegal sizing exists anywhere. + assert!(proposals.candidates.iter().all(|c| match &c.replacement { + Replacement::Summary(node) + if summary_family_algorithm(node) == SketchAlgorithm::Kll => + kll_k_of(node) != 20, + _ => true, + })); + } + + #[test] + fn legality_precedes_cost_in_search_and_global_selection() { + // Same fixture through the workload search: the illegal cheaper + // candidate is absent from the group *before* any cost ranking, the + // rejection is recorded on the group, and global selection commits + // to the legal, more expensive one. + let inner = agg(vec![2], quantile_eps(0.5, 0.1), metric_scan(&["job"])); + let outer = Rc::new(agg(vec![], quantile_eps(0.99, 0.1), inner)); + let strategies: Vec> = + vec![Box::new(SketchAlgorithmStrategy::with_models( + &DefaultCostModel, + &RankAdditiveModel, + &EqualSplitAllocator, + ))]; + let space = search_workload_with(vec![("q", Rc::clone(&outer))], &strategies); + let root = &space.roots[0].1; + let group = space.group_for(root).unwrap(); + assert!(!group.rejected.is_empty()); + assert!(group.candidates.iter().all(|c| match &c.replacement { + Replacement::Summary(node) => node.guarantee.as_ref().is_some_and(|g| { + DefaultAccuracyModel.satisfies(g, &AccuracyTarget::Epsilon(0.1)) + }), + Replacement::Rewrite(_) => false, + })); + let ranked = space.cost_sorted(&DefaultCostModel); + let root_ranked = ranked.iter().find(|g| Rc::ptr_eq(g.target, root)).unwrap(); + assert_eq!(root_ranked.candidates.len(), group.candidates.len()); + + let selection = space.global_selection(&DefaultCostModel); + let chosen = selection + .for_target(root) + .unwrap() + .chosen + .expect("a legal candidate wins"); + let Replacement::Summary(node) = &chosen.replacement else { + panic!() + }; + assert_eq!(kll_k_of(node), 40); + } + + #[test] + fn root_target_check_removes_candidates_before_cost_ranking() { + let q = Rc::new(agg(vec![2], default_quantile(0.99), metric_scan(&["job"]))); + // A root target tighter than the node's own ε=0.01: every sketch + // candidate misses it and is moved to `rejected`; nothing is left + // for the cost model to rank. + let space = search_workload_with_targets( + vec![("q", Rc::clone(&q), Some(AccuracyTarget::Epsilon(0.001)))], + &default_strategies(), + &DefaultAccuracyModel, + ); + let root = &space.roots[0].1; + let group = space.group_for(root).unwrap(); + assert!(group + .candidates + .iter() + .all(|c| matches!(c.replacement, Replacement::Rewrite(_)))); + assert!(group.rejected.iter().all(|r| matches!( + r.error, + AccuracyError::TargetNotSatisfied { target: AccuracyTarget::Epsilon(e), .. } if e == 0.001 + ))); + assert!(group.rejected.len() >= 2); + let selection = space.global_selection(&DefaultCostModel); + assert!(selection.for_target(root).unwrap().chosen.is_none()); + + // A root target the node's own sizing meets keeps every candidate. + let space = search_workload_with_targets( + vec![("q", Rc::clone(&q), Some(AccuracyTarget::Epsilon(0.01)))], + &default_strategies(), + &DefaultAccuracyModel, + ); + let group = space.group_for(&space.roots[0].1).unwrap(); + assert!(group + .candidates + .iter() + .any(|c| matches!(c.replacement, Replacement::Summary(_)))); + + // An `Exact` root target admits only exact candidates. + let space = search_workload_with_targets( + vec![("q", Rc::clone(&q), Some(AccuracyTarget::Exact))], + &default_strategies(), + &DefaultAccuracyModel, + ); + let group = space.group_for(&space.roots[0].1).unwrap(); + assert!(group.candidates.iter().all(|c| match &c.replacement { + Replacement::Summary(node) => node + .guarantee + .as_ref() + .is_some_and(ResultGuarantee::is_exact), + Replacement::Rewrite(_) => true, + })); + } } diff --git a/crates/devtools/src/bin/dag_export.rs b/crates/devtools/src/bin/dag_export.rs index 3dd7dde..071c4a4 100644 --- a/crates/devtools/src/bin/dag_export.rs +++ b/crates/devtools/src/bin/dag_export.rs @@ -56,8 +56,8 @@ use std::time::Instant; use asap_aware_mapping::cost_model::DefaultCostModel; use asap_aware_mapping::replacement::{search_workload, Replacement, ReplacementSubDAG}; use asap_types::dag_export::{ - self, DagDecision, DagGraph, DagNote, NamedGraph, PostAsapSubstitution, TargetReplacement, - TargetReplacementAfter, WorkloadGraph, + self, DagDecision, DagGraph, DagNote, NamedGraph, PostAsapSubstitution, TargetRejection, + TargetReplacement, TargetReplacementAfter, WorkloadGraph, }; use asap_types::post_asap::SummaryExpr; use asap_types::pre_asap::cse::{structural_hash, HashCache}; @@ -358,6 +358,10 @@ struct PostAsapResults { /// [`dag_export::export_post_asap`] — every winning candidate spliced /// directly into that query's own pre-ASAP shape in place. post_graphs: Vec<(String, DagGraph)>, + /// One `(query_name, TargetRejection)` per accuracy-illegal candidate + /// the search refused (`MemoGroup::rejected`, issue #172) whose target + /// node is found in that query's own exported graph. + rejections: Vec<(String, TargetRejection)>, } /// Assign collision-free, explicit identities to structurally equal nodes @@ -545,7 +549,19 @@ fn run_post_asap_with_progress( // subtree and inside `post_graph` as a whole. let mut lookup_cache = HashCache::new(); let mut replacements = Vec::new(); + let mut rejections = Vec::new(); let mut matched = vec![false; winners.len()]; + // Groups with accuracy-refused candidates (issue #172): matched to a + // query's graph nodes the same hash-then-structural-equality way. + let rejected_groups: Vec<_> = space + .groups() + .filter(|group| !group.rejected.is_empty()) + .collect(); + let mut rejected_by_hash: HashMap> = HashMap::new(); + for (i, group) in rejected_groups.iter().enumerate() { + let hash = structural_hash(&group.target, &mut by_hash_cache); + rejected_by_hash.entry(hash).or_default().push(i); + } for (name, _, qe) in lowered_queries { let graph = dag_export::export(qe); for node in &graph.nodes { @@ -559,6 +575,24 @@ fn run_post_asap_with_progress( )); matched[i] = true; } + let hash = structural_hash(source_expr, &mut lookup_cache); + for &i in rejected_by_hash.get(&hash).into_iter().flatten() { + let group = rejected_groups[i]; + if *source_expr != *group.target { + continue; + } + rejections.extend(group.rejected.iter().map(|rejected| { + ( + name.clone(), + TargetRejection { + target_pre_id: node.id, + strategy: rejected.strategy.to_string(), + description: rejected.description.clone(), + error: rejected.error.clone(), + }, + ) + })); + } } } @@ -602,6 +636,7 @@ fn run_post_asap_with_progress( PostAsapResults { replacements, post_graphs, + rejections, } } @@ -675,6 +710,7 @@ async fn main() { graph, replacements: Vec::new(), post_graph: None, + rejections: Vec::new(), }); } for (explanation, matched) in explanations.iter().zip(matched) { @@ -704,6 +740,11 @@ async fn main() { named.post_graph = Some(post_graph); } } + for (query_name, rejection) in results.rejections { + if let Some(named) = queries.iter_mut().find(|q| q.name == query_name) { + named.rejections.push(rejection); + } + } } { diff --git a/crates/types/src/dag_export.rs b/crates/types/src/dag_export.rs index 252d319..5df2ed8 100644 --- a/crates/types/src/dag_export.rs +++ b/crates/types/src/dag_export.rs @@ -46,7 +46,7 @@ use std::rc::Rc; use serde::Serialize; -use crate::post_asap::{SummaryExpr, SummaryNode}; +use crate::post_asap::{AccuracyError, ResultGuarantee, SummaryExpr, SummaryNode}; use crate::pre_asap::cse::{structural_hash, HashCache}; use crate::pre_asap::query_expr::{QueryExpr, Source}; @@ -199,6 +199,12 @@ pub struct NamedGraph { /// `NamedGraph` is unaffected. #[serde(default, skip_serializing_if = "Option::is_none")] pub post_graph: Option, + /// Accuracy-illegal candidates a higher layer's search refused for + /// targets in this query (issue #172) — see [`TargetRejection`]. Always + /// empty coming out of this module; omitted from the JSON when empty, + /// same additive rule as `replacements`. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub rejections: Vec, } /// A batch of named queries — the shape the viewer's multi-query / compare @@ -277,6 +283,33 @@ pub struct SummaryDagNode { /// Child node ids, in the variant's field order (e.g. `SummaryJoin` is /// `[outer, inner]`). pub children: Vec, + /// The value's machine-readable accuracy guarantee (issue #172) — + /// [`SummaryNode::guarantee`] serialized structurally (metric, symbolic + /// bound, failure probability, provenance including any budget + /// allocation), not as prose. Omitted when the node carries none (raw + /// summary state, or a family with no error model), so every consumer + /// predating this field parses the same shape it always has. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub guarantee: Option, +} + +/// One accuracy-illegal candidate a higher layer's search refused for a +/// target (issue #172) — `asap_aware_mapping::replacement::RejectedCandidate` +/// re-shaped into this crate's own crate-agnostic vocabulary, the same +/// layering rule as [`TargetReplacement`]. Carried on +/// [`NamedGraph::rejections`] so a renderer can explain *why* a target kept +/// its raw/pre-ASAP form, not only what won elsewhere. +#[derive(Debug, Clone, Serialize)] +pub struct TargetRejection { + /// Id of the [`DagNode`] in this query's own `graph.nodes` the refused + /// candidate targeted. + pub target_pre_id: u32, + /// Which strategy considered the candidate. + pub strategy: String, + /// What the candidate would have been. + pub description: String, + /// The typed reason it was refused. + pub error: AccuracyError, } /// One post-ASAP `SummaryNode` tree, flattened the same way [`DagGraph`] @@ -315,6 +348,7 @@ fn push_summary_node( label: String, detail: serde_json::Value, children: Vec, + guarantee: Option, ) -> u32 { let id = nodes.len() as u32; nodes.push(SummaryDagNode { @@ -323,6 +357,7 @@ fn push_summary_node( label, detail, children, + guarantee, }); id } @@ -433,14 +468,21 @@ fn build_summary(node: &SummaryNode, nodes: &mut Vec) -> u32 { let inner_kind = pre_asap_subgraph.nodes[pre_asap_subgraph.root as usize].kind; let label = format!("KeepPreAsap({inner_kind})"); let detail = serde_json::json!({ "pre_asap_subgraph": pre_asap_subgraph }); - return push_summary_node(nodes, "KeepPreAsap", label, detail, vec![]); + return push_summary_node( + nodes, + "KeepPreAsap", + label, + detail, + vec![], + node.guarantee.clone(), + ); } let children: Vec = summary_children(&node.expr) .into_iter() .map(|child| build_summary(child, nodes)) .collect(); let (kind, label, detail) = summary_shape(&node.expr); - push_summary_node(nodes, kind, label, detail, children) + push_summary_node(nodes, kind, label, detail, children, node.guarantee.clone()) } /// One replacement site a higher layer (the `dag_export` binary) found by @@ -725,7 +767,17 @@ fn build_summary_hybrid( .into_iter() .map(|child| build_summary_hybrid(child, nodes, cache, find_winner)) .collect(); - let (kind, label, detail) = summary_shape(&node.expr); + let (kind, label, mut detail) = summary_shape(&node.expr); + // The merged graph's `DagNode` has no dedicated guarantee field (it is + // the pre-ASAP node shape); the guarantee rides in `detail` under the + // same key/shape `SummaryDagNode::guarantee` uses, additively. + if let Some(guarantee) = &node.guarantee { + if let (serde_json::Value::Object(map), Ok(value)) = + (&mut detail, serde_json::to_value(guarantee)) + { + map.insert("guarantee".into(), value); + } + } let id = push_summary_originated_node(nodes, kind, label, detail, children); nodes[id as usize].schema = Some(summary_schema_json(&node.schema)); id @@ -1415,4 +1467,131 @@ mod tests { on the Aggregate subtree it represents, not just the root" ); } + + /// Issue #172: a readout's guarantee is exported structurally — metric, + /// symbolic bound, failure probability, provenance (allocation + /// included) — and a rejection carries its typed reason. + #[test] + fn export_carries_guarantee_allocation_and_rejection_reason() { + use crate::post_asap::{ + BoundExpr, CompositionOperator, ErrorMetric, GroupingStrategy, GuaranteeSource, + ProbabilityExpr, SketchAlgorithm, SketchKind, SketchParams, SketchQuery, + SummaryFamilyType, SummarySchema, + }; + let leaf = Rc::new(scan("t", vec![Column::new("v", DataType::Float64, false)])); + let kept = Rc::new(SummaryNode { + expr: SummaryExpr::KeepPreAsap(Rc::clone(&leaf)), + schema: SummarySchema { + fields: vec![], + time_index: None, + }, + guarantee: Some(ResultGuarantee::exact("KeepPreAsap")), + }); + let agg = Rc::new(SummaryNode { + expr: SummaryExpr::SummaryAgg { + child: kept, + family: SummaryFamilyType::Sketch( + SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 40 }), + GroupingStrategy::default(), + ), + col: crate::pre_asap::expr_ir::ColumnRef::Named("v".into()), + reduction: Reduction::by(vec![]), + grouping: GroupingStrategy::default(), + }, + schema: SummarySchema { + fields: vec![], + time_index: None, + }, + guarantee: None, + }); + let guarantee = ResultGuarantee { + metric: ErrorMetric::Rank, + bound: BoundExpr::Sum { + terms: vec![ + BoundExpr::Constant { value: 0.05 }, + BoundExpr::Constant { value: 0.05 }, + ], + }, + failure_probability: ProbabilityExpr::UnionBound { + terms: vec![ProbabilityExpr::Constant { value: 0.01 }], + }, + provenance: vec![ + GuaranteeSource::CompositionStep { + operator: CompositionOperator::ApproximateAggregate, + rule: "additive_union_bound".into(), + }, + GuaranteeSource::BudgetAllocation { + allocator: "EqualSplitAllocator".into(), + layer: 0, + layer_count: 2, + local_target: AccuracyTarget::Epsilon(0.05), + end_to_end_target: AccuracyTarget::Epsilon(0.1), + }, + ], + }; + let root = SummaryNode { + expr: SummaryExpr::SummaryEstimate { + summary_input: agg, + query: SketchQuery::Quantile { q: 0.99 }, + }, + schema: SummarySchema { + fields: vec![], + time_index: None, + }, + guarantee: Some(guarantee), + }; + let graph = export_summary(&root); + let json = serde_json::to_value(&graph).unwrap(); + let root_json = &json["nodes"][graph.root as usize]; + assert_eq!(root_json["guarantee"]["metric"], "rank"); + assert_eq!(root_json["guarantee"]["bound"]["op"], "sum"); + assert_eq!( + root_json["guarantee"]["failure_probability"]["op"], + "union_bound" + ); + let provenance = root_json["guarantee"]["provenance"].as_array().unwrap(); + assert!(provenance + .iter() + .any(|s| s["kind"] == "budget_allocation" && s["layer_count"] == 2)); + assert!(provenance.iter().any(|s| s["kind"] == "composition_step")); + // Raw sketch state carries none; the exact leaf carries zero error. + let state = &json["nodes"][1]; + assert_eq!(state["kind"], "SummaryAgg"); + assert!(state.get("guarantee").is_none()); + assert_eq!(json["nodes"][0]["guarantee"]["bound"]["op"], "zero"); + + let named = NamedGraph { + name: "q".into(), + source: None, + graph: export(&leaf), + replacements: vec![], + post_graph: None, + rejections: vec![TargetRejection { + target_pre_id: 0, + strategy: "SketchAlgorithmStrategy".into(), + description: "quantile over quantile".into(), + error: AccuracyError::UnsupportedComposition { + operator: CompositionOperator::ApproximateAggregate, + input_metrics: vec![ErrorMetric::Rank], + local_metric: Some(ErrorMetric::Rank), + reason: "no registered rule".into(), + }, + }], + }; + let json = serde_json::to_value(&named).unwrap(); + assert_eq!( + json["rejections"][0]["error"]["kind"], + "unsupported_composition" + ); + assert_eq!(json["rejections"][0]["error"]["input_metrics"][0], "rank"); + // Additive: a graph with no rejections omits the key entirely. + let plain = NamedGraph { + rejections: vec![], + ..named + }; + assert!(serde_json::to_value(&plain) + .unwrap() + .get("rejections") + .is_none()); + } } diff --git a/crates/types/src/post_asap/expr.rs b/crates/types/src/post_asap/expr.rs index 2402b22..b93aa90 100644 --- a/crates/types/src/post_asap/expr.rs +++ b/crates/types/src/post_asap/expr.rs @@ -1,5 +1,6 @@ use std::rc::Rc; +use super::guarantee::ResultGuarantee; use super::schema::{SummaryFamilyType, SummarySchema}; use super::sketch::{GroupingStrategy, SketchQuery}; use crate::pre_asap::{ColumnRef, QueryExpr, Reduction}; @@ -16,6 +17,18 @@ pub struct SummaryNode { /// Output schema of `expr` — the schema of the data flowing on the edge /// leading *from* this node to its parent(s). pub schema: SummarySchema, + /// The machine-readable accuracy guarantee of the *value* this node + /// produces (issue #172) — `Some` on every finalized, caller-visible + /// value: a `SummaryEstimate` readout, an `ExactAggregate`-family + /// `SummaryAgg` (its state *is* the value), or a `KeepPreAsap` subtree + /// (executed exactly). `None` on raw summary state — a sketch-family + /// `SummaryAgg`, `SummaryMerge`, `SummarySubtract`, `SummaryDelete`, + /// `SummaryJoin` — whose guarantee only exists once something reads it + /// out; and `None` on a readout of a family the plugged-in + /// `AccuracyModel` has no local guarantee for (`Sample`/`Wavelet`/ + /// `StatModel`), which a fail-closed consumer must treat as "unknown", + /// never as exact. + pub guarantee: Option, } // ── Post-ASAP sketch-bound IR ──────────────────────────────────────────────── diff --git a/crates/types/src/post_asap/guarantee.rs b/crates/types/src/post_asap/guarantee.rs new file mode 100644 index 0000000..04471a7 --- /dev/null +++ b/crates/types/src/post_asap/guarantee.rs @@ -0,0 +1,450 @@ +//! Machine-readable accuracy guarantees for finalized post-ASAP values +//! (issue #172). +//! +//! A selected post-ASAP plan used to carry no statement about the error of +//! the value it produces: every approximate layer was sized from its own +//! [`AccuracyTarget`] as if its input were exact, so an approximate parent +//! could silently consume an approximate child. This module is the +//! *vocabulary* that fixes that — the typed metric, the symbolic bound and +//! failure-probability expressions, the provenance trail, and the typed +//! rejection reasons. The *algebra* that composes these (the `AccuracyModel` +//! trait, its default conservative rules, and budget allocation) lives one +//! layer up in `asap_aware_mapping::accuracy`, the same layering +//! [`crate::dag_export`] keeps for cost decisions: this crate defines the +//! shapes, the planning crate decides. +//! +//! ## What a guarantee says +//! +//! [`ResultGuarantee`] is attached to a finalized, caller-visible value — +//! [`super::SummaryNode::guarantee`] on a `SummaryEstimate` readout, an +//! exact accumulator, or a kept pre-ASAP subtree — never to raw summary +//! state (a `SummaryAgg` sketch node carries `None`; its readout carries the +//! guarantee). Its statement is: +//! +//! ```text +//! Pr[ err_metric(estimate, truth) > bound ] <= failure_probability +//! ``` +//! +//! where `err_metric` is fixed by [`ErrorMetric`] and each metric has its +//! own normalization (documented per variant). Metrics are **not** +//! interchangeable: a cardinality error and a frequency error are different +//! quantities, and composing them needs an explicit rule, never an implicit +//! "add the epsilons". +//! +//! ## Why expressions, not numbers +//! +//! [`BoundExpr`]/[`ProbabilityExpr`] are tiny serializable expression trees +//! rather than bare `f64`s so a planning-time guarantee can reference a +//! statistic it does not have (a group count, a stream's L1 norm) and stay +//! honestly *unknown* until something instantiates it — a deployment's own +//! cardinality estimate, or a runtime posterior observation (issue #239). +//! [`BoundExpr::evaluate`] returns `None`, never `0`, for such a bound; +//! "unknown" and "zero" are different answers and a fail-closed planner +//! treats them differently. +//! +//! ## What is deliberately *not* here +//! +//! No `CorrectnessPolicy`-style enum: [`AccuracyTarget`] remains the one +//! authoritative requirement type, and [`AccuracyError`] is the typed reason +//! a candidate failed against it. No independence assumptions: the only +//! probability combinator is the union bound. + +use serde::{Deserialize, Serialize}; + +use crate::types::AccuracyTarget; + +/// Which error quantity a [`ResultGuarantee`] bounds. `#[non_exhaustive]`: +/// a deployment's own `AccuracyModel` may need a metric this crate does not +/// enumerate yet, and downstream matches must not assume the list is closed. +#[non_exhaustive] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ErrorMetric { + /// `|estimate − truth| ≤ bound`, in the value's own units. + AbsoluteValue, + /// `|estimate − truth| ≤ bound · |truth|` — a multiplicative guarantee, + /// only meaningful for values of known sign (DDSketch's α). + RelativeValue, + /// The returned value's *rank* in the input multiset is within + /// `bound · n` of the requested rank (KLL's ε). Says nothing about how + /// far the returned *value* is from the true quantile value. + Rank, + /// `|estimate − truth| ≤ bound · truth` for a distinct count (HLL/Theta/ + /// KMV's relative standard error). + Cardinality, + /// `|estimate − truth| ≤ bound · ‖f‖₁` for a point-frequency query + /// (CMS's ε, normalized by the stream's L1 norm). + Frequency, + /// The returned key set equals the true top-k set. No shipped model + /// produces this yet: it needs a per-key interval margin certificate + /// (issue #172, PR 3). Present so the vocabulary can name the metric a + /// `TopK` readout would need without pretending a frequency bound is + /// one. + TopKMembership, +} + +/// A symbolic, serializable error-bound expression. Non-negative real +/// arithmetic only — there is no subtraction, so a bound can never be +/// tightened by construction, only by evaluating a known statistic. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(tag = "op", rename_all = "snake_case")] +pub enum BoundExpr { + /// Exactly zero error (deterministic exact computation). + Zero, + /// A resolved numeric bound in the metric's own normalization. + Constant { value: f64 }, + /// `Σ terms` — the additive composition rule. + Sum { terms: Vec }, + /// `Π factors` — e.g. the relative-error cross term, or a normalized + /// bound times the (possibly unknown) statistic it is normalized by. + Product { factors: Vec }, + /// `factor · inner` — an explicitly registered Lipschitz constant. + Scaled { factor: f64, inner: Box }, + /// `max(terms)` — exact max/min over bounded inputs. + Max { terms: Vec }, + /// A statistic this bound needs but nothing has supplied yet. Evaluates + /// to `None`, never `0`: an unknown quantity is not a small one. + Unknown { statistic: String }, +} + +impl BoundExpr { + /// Numeric value of this bound, or `None` if any [`BoundExpr::Unknown`] + /// leaf is reachable. + pub fn evaluate(&self) -> Option { + match self { + BoundExpr::Zero => Some(0.0), + BoundExpr::Constant { value } => Some(*value), + BoundExpr::Sum { terms } => terms.iter().map(BoundExpr::evaluate).sum(), + BoundExpr::Product { factors } => factors.iter().map(BoundExpr::evaluate).product(), + BoundExpr::Scaled { factor, inner } => inner.evaluate().map(|b| factor * b), + BoundExpr::Max { terms } => terms + .iter() + .map(BoundExpr::evaluate) + .try_fold(0.0_f64, |acc, t| t.map(|t| acc.max(t))), + BoundExpr::Unknown { .. } => None, + } + } + + /// `true` iff this bound is structurally zero (every leaf is + /// [`BoundExpr::Zero`], or a `Product`/`Scaled` contains a zero factor). + /// Distinct from `evaluate() == Some(0.0)` only in that it never + /// depends on floating-point evaluation. + pub fn is_zero(&self) -> bool { + match self { + BoundExpr::Zero => true, + BoundExpr::Constant { value } => *value == 0.0, + BoundExpr::Sum { terms } | BoundExpr::Max { terms } => { + terms.iter().all(BoundExpr::is_zero) + } + BoundExpr::Product { factors } => factors.iter().any(BoundExpr::is_zero), + BoundExpr::Scaled { factor, inner } => *factor == 0.0 || inner.is_zero(), + BoundExpr::Unknown { .. } => false, + } + } +} + +/// A symbolic, serializable failure-probability expression. The only +/// combinator over several events is the union bound — the default model +/// never assumes independence between sketch errors. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(tag = "op", rename_all = "snake_case")] +pub enum ProbabilityExpr { + /// The guarantee is deterministic. + Zero, + /// A resolved probability in `[0, 1]`. + Constant { value: f64 }, + /// `min(1, Σ terms)` — Boole's inequality over the listed events. + UnionBound { terms: Vec }, + /// `min(1, count · inner)` — the union bound over `count` events that + /// each fail with probability at most `inner` (e.g. one per input row of + /// an exact aggregation). `count` is a [`BoundExpr`] so it may be an + /// [`BoundExpr::Unknown`] statistic. + Scaled { + count: BoundExpr, + inner: Box, + }, + /// A probability nothing has supplied yet — same stance as + /// [`BoundExpr::Unknown`]. + Unknown { statistic: String }, +} + +impl ProbabilityExpr { + /// Numeric value clamped to `[0, 1]`, or `None` if any unknown leaf is + /// reachable. + pub fn evaluate(&self) -> Option { + let raw = match self { + ProbabilityExpr::Zero => 0.0, + ProbabilityExpr::Constant { value } => *value, + ProbabilityExpr::UnionBound { terms } => terms + .iter() + .map(ProbabilityExpr::evaluate) + .sum::>()?, + ProbabilityExpr::Scaled { count, inner } => count.evaluate()? * inner.evaluate()?, + ProbabilityExpr::Unknown { .. } => return None, + }; + Some(raw.clamp(0.0, 1.0)) + } + + /// `true` iff this probability is structurally zero. + pub fn is_zero(&self) -> bool { + match self { + ProbabilityExpr::Zero => true, + ProbabilityExpr::Constant { value } => *value == 0.0, + ProbabilityExpr::UnionBound { terms } => terms.iter().all(ProbabilityExpr::is_zero), + ProbabilityExpr::Scaled { count, inner } => count.is_zero() || inner.is_zero(), + ProbabilityExpr::Unknown { .. } => false, + } + } +} + +/// How a parent operator consumes its inputs' values — the shape an +/// `AccuracyModel::propagate` rule is registered against. `#[non_exhaustive]` +/// for the same reason [`ErrorMetric`] is. +#[non_exhaustive] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(tag = "op", rename_all = "snake_case")] +pub enum CompositionOperator { + /// An approximate summary built over its inputs' (approximate) values + /// — the sketch-over-sketch case. Its own `local` guarantee composes + /// with the inputs' under a same-metric rule. + ApproximateAggregate, + /// A deterministic transformation with an explicitly registered global + /// Lipschitz constant: `B_out ≤ constant · B_in + B_local`. The planner + /// never derives `constant` itself; only a caller that has proved it + /// may construct this operator. + Lipschitz { constant: f64 }, + /// An exact sum over approximate inputs: `B ≤ Σ B_i`, `δ ≤ Σ δ_i`. + ExactSum, + /// An exact max/min over approximate inputs — bounds the returned + /// *value* (`max` of the input bounds) but does not identify which key + /// is the true winner. + ExactExtremum, + /// A top-k selection over approximate inputs. Unsupported by the default + /// model until the margin certificate of issue #172 PR 3 exists. + TopKSelection, +} + +/// One entry in a [`ResultGuarantee`]'s provenance trail — enough for a +/// reader to reconstruct *why* the bound is what it is without re-running +/// the planner. `#[non_exhaustive]` so runtime evidence (issue #239) and +/// deployment-specific sources can be appended later. +#[non_exhaustive] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum GuaranteeSource { + /// Deterministic exact computation — zero error by construction. + Exact { + /// What made it exact (e.g. `"ExactAggregate(Sum)"`, + /// `"KeepPreAsap"`). + reason: String, + }, + /// The target this readout's sketch was sized against. + AccuracyTarget { target: AccuracyTarget }, + /// The concrete sketch a readout's local guarantee was derived from. + SketchReadout { + algorithm: String, + params: serde_json::Value, + query: String, + }, + /// A composed input's own guarantee, carried verbatim so the trail is + /// self-contained. `input_index` is the input's position in the + /// composition (0-based, in the parent's child order). + ChildGuarantee { + input_index: usize, + guarantee: Box, + }, + /// The propagation rule that produced this guarantee from its inputs. + CompositionStep { + operator: CompositionOperator, + /// Stable rule name (e.g. `"additive_union_bound"`). + rule: String, + }, + /// The budget split that produced this layer's local target — present + /// only when an `AccuracyBudgetAllocator` re-sized a layer. + BudgetAllocation { + allocator: String, + layer: usize, + layer_count: usize, + local_target: AccuracyTarget, + end_to_end_target: AccuracyTarget, + }, + /// A statistic the bound needs but nothing supplied — the reason a + /// [`BoundExpr::Unknown`]/[`ProbabilityExpr::Unknown`] leaf exists. + UnavailableStatistic { statistic: String }, + /// Query-time evidence (issue #239's posterior bounds). Never produced + /// at planning time; reserved so a runtime can append its observation + /// to the same trail instead of inventing a parallel one. + RuntimeObservation { + source: String, + detail: serde_json::Value, + }, +} + +/// The machine-readable accuracy statement attached to a finalized +/// post-ASAP value — see the module docs for its semantics. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct ResultGuarantee { + pub metric: ErrorMetric, + pub bound: BoundExpr, + pub failure_probability: ProbabilityExpr, + pub provenance: Vec, +} + +impl ResultGuarantee { + /// The zero-error, zero-failure guarantee of a deterministic exact + /// computation. `metric` is [`ErrorMetric::AbsoluteValue`]: an exact + /// value is exact under every metric, and absolute error is the one + /// every same-metric rule accepts as a zero input. + pub fn exact(reason: impl Into) -> Self { + Self { + metric: ErrorMetric::AbsoluteValue, + bound: BoundExpr::Zero, + failure_probability: ProbabilityExpr::Zero, + provenance: vec![GuaranteeSource::Exact { + reason: reason.into(), + }], + } + } + + /// `true` iff this guarantee promises zero error with certainty. + pub fn is_exact(&self) -> bool { + self.bound.is_zero() && self.failure_probability.is_zero() + } + + /// How many approximate sketch readouts contributed to this value — + /// `1` for a plain readout, `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. + pub fn approximate_layer_count(&self) -> usize { + self.provenance + .iter() + .map(|source| match source { + GuaranteeSource::SketchReadout { .. } => 1, + GuaranteeSource::ChildGuarantee { guarantee, .. } => { + guarantee.approximate_layer_count() + } + _ => 0, + }) + .sum() + } +} + +/// Why an accuracy check rejected a candidate. Typed, serializable, and +/// carried through to DAG export so a rejection is as inspectable as a +/// selection. Never a reason to "treat the child as exact". +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, thiserror::Error)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum AccuracyError { + /// No registered propagation rule covers this operator over these + /// input metrics (or this local metric). + #[error( + "unsupported accuracy composition: {operator:?} over inputs {input_metrics:?} \ + with local {local_metric:?} — {reason}" + )] + UnsupportedComposition { + operator: CompositionOperator, + input_metrics: Vec, + local_metric: Option, + reason: String, + }, + /// An approximate input carries no guarantee at all, so nothing can be + /// composed over it. + #[error("input {input_index} of {operator:?} carries no accuracy guarantee")] + MissingInputGuarantee { + operator: CompositionOperator, + input_index: usize, + }, + /// The composed guarantee does not satisfy the applicable + /// [`AccuracyTarget`]. `bound`/`failure_probability` are the evaluated + /// values when known. + #[error( + "composed guarantee ({metric:?}, bound {bound:?}, failure probability \ + {failure_probability:?}) does not satisfy {target:?}" + )] + TargetNotSatisfied { + metric: ErrorMetric, + bound: Option, + failure_probability: Option, + target: AccuracyTarget, + }, + /// No budget allocation could make the composition legal under the + /// end-to-end target. + #[error("no legal accuracy-budget allocation for {target:?} across {layer_count} layers")] + NoLegalAllocation { + target: AccuracyTarget, + layer_count: usize, + }, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn unknown_statistic_never_evaluates_to_zero() { + let b = BoundExpr::Product { + factors: vec![ + BoundExpr::Constant { value: 0.01 }, + BoundExpr::Unknown { + statistic: "input_row_count".into(), + }, + ], + }; + assert_eq!(b.evaluate(), None); + assert!(!b.is_zero()); + let p = ProbabilityExpr::Scaled { + count: BoundExpr::Unknown { + statistic: "input_row_count".into(), + }, + inner: Box::new(ProbabilityExpr::Constant { value: 0.01 }), + }; + assert_eq!(p.evaluate(), None); + } + + #[test] + fn union_bound_sums_and_clamps() { + let p = ProbabilityExpr::UnionBound { + terms: vec![ + ProbabilityExpr::Constant { value: 0.7 }, + ProbabilityExpr::Constant { value: 0.6 }, + ], + }; + assert_eq!(p.evaluate(), Some(1.0)); + } + + #[test] + fn exact_guarantee_is_zero_layers() { + let g = ResultGuarantee::exact("test"); + assert!(g.is_exact()); + assert_eq!(g.approximate_layer_count(), 0); + } + + #[test] + fn guarantee_round_trips_through_json() { + let g = ResultGuarantee { + metric: ErrorMetric::Frequency, + bound: BoundExpr::Sum { + terms: vec![ + BoundExpr::Constant { value: 0.01 }, + BoundExpr::Scaled { + factor: 2.0, + inner: Box::new(BoundExpr::Constant { value: 0.005 }), + }, + ], + }, + failure_probability: ProbabilityExpr::UnionBound { + terms: vec![ProbabilityExpr::Constant { value: 0.01 }], + }, + provenance: vec![GuaranteeSource::CompositionStep { + operator: CompositionOperator::Lipschitz { constant: 2.0 }, + rule: "lipschitz".into(), + }], + }; + let json = serde_json::to_value(&g).unwrap(); + assert_eq!(json["metric"], "frequency"); + assert_eq!(json["bound"]["op"], "sum"); + let back: ResultGuarantee = serde_json::from_value(json).unwrap(); + assert_eq!(back, g); + } +} diff --git a/crates/types/src/post_asap/mod.rs b/crates/types/src/post_asap/mod.rs index 5809eb7..5ce4538 100644 --- a/crates/types/src/post_asap/mod.rs +++ b/crates/types/src/post_asap/mod.rs @@ -28,11 +28,16 @@ //! — see `asap_aware_mapping::grouping`'s module docs for why. pub mod expr; +pub mod guarantee; pub mod query_time; pub mod schema; pub mod sketch; pub use expr::{SummaryExpr, SummaryNode}; +pub use guarantee::{ + AccuracyError, BoundExpr, CompositionOperator, ErrorMetric, GuaranteeSource, ProbabilityExpr, + ResultGuarantee, +}; pub use query_time::{ classic_cms_sizing, cms_posterior_error_bound, count_sketch_posterior_error_bound, cu_sketch_posterior_error_bound, traditional_a_priori_bound, diff --git a/crates/types/src/post_asap/sketch.rs b/crates/types/src/post_asap/sketch.rs index e3a4a8b..3e40aac 100644 --- a/crates/types/src/post_asap/sketch.rs +++ b/crates/types/src/post_asap/sketch.rs @@ -1,3 +1,5 @@ +use serde::{Deserialize, Serialize}; + use crate::pre_asap::ColumnRef; // ── Exact accumulators ────────────────────────────────────────────────────── @@ -38,7 +40,7 @@ pub enum ExactParams { /// [`SketchParams`]. Each algorithm belongs to exactly one [`SketchKind`] /// category (e.g. `Kll` and `DDSketch` both realize quantile sketches); /// [`SketchKind::new`] is where that classification is made. -#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)] +#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)] pub enum SketchAlgorithm { /// KLL quantile sketch (mergeable, ε-accurate rank queries). Kll, @@ -67,7 +69,7 @@ pub enum SketchAlgorithm { /// instance. The variant must correspond to the associated `SketchAlgorithm`; /// mismatches are caught at post-ASAP bind time, before any later, /// deployment-specific stage ever sees the plan. -#[derive(Debug, Clone, PartialEq)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub enum SketchParams { Kll { k: u32, diff --git a/docs/design_docs/asap_aware_mapping.md b/docs/design_docs/asap_aware_mapping.md index 821c5b5..dfc0fdc 100644 --- a/docs/design_docs/asap_aware_mapping.md +++ b/docs/design_docs/asap_aware_mapping.md @@ -413,6 +413,37 @@ Given an accuracy target, the mapping layer should expose configurations that sa Accuracy requirements therefore act as **constraints on the search space**, rather than as a separate decision made after a summary family has already been chosen. +## End-to-end guarantees for nested summaries (issue #172) + +A single summary is sized against its own `AccuracyTarget`, but a summary +over another summary's *readout* is only legal if the composed error still +meets the requirement on the outer value. `asap-aware-mapping` therefore +runs legality strictly before costing: + +```text +candidate generation + -> guarantee propagation (AccuracyModel::propagate) + -> AccuracyTarget satisfaction (AccuracyModel::satisfies) + -> legal candidates only (illegal ones -> MemoGroup::rejected) + -> cost ranking / global selection (CostModel) +``` + +Every finalized post-ASAP value carries a machine-readable +`ResultGuarantee` (`asap_types::post_asap::guarantee`): a typed +`ErrorMetric`, a symbolic `BoundExpr`, a `ProbabilityExpr`, and a +provenance trail. Exact values are zero-error; a sketch readout's guarantee +inverts the sizing formula that produced its parameters. The default +`AccuracyModel` (`asap_aware_mapping::accuracy`) is deliberately +conservative: same-metric additive/relative/Lipschitz rules and exact +sum/max/min over approximate inputs, union-bound probabilities (no +independence), unknown statistics kept unknown, and +`AccuracyError::UnsupportedComposition` for everything else — never +"treat the child as exact". An `AccuracyBudgetAllocator` (initially an +equal split) proposes re-sized layers so a legal nested plan can exist at +all; `CostModel` ranks only what survives. See that module's docs for the +precedence rules between a root `QueryRequirements.accuracy` and per-node +`AggIntent.accuracy`. + --- # Design Dimensions