feat(planner): error-budget allocation and whole-result δ scaling (#669, part b)

[zzylol]

Why

#669 composes guarantees through arithmetic and comparisons, but two parts were missing (part b):

• When a query combines several approximate inputs (count(a) - count(b), a summary over an approximate input), each input was sized for the query's whole budget. The composed error and failure probability could therefore be up to n times the target.
• Every guarantee is per value (one series or row), while a target covers the whole returned result by default. Stage 3 compared the two directly, so a target "every value within ε with probability 1 − δ" was checked against a δ that holds for one value only.

What

• Equal budget split at Stage 1 composition. When a query's value combines n ≥ 2 approximate inputs, each summary is sized for an equal share of the query's target: ε/n (for a relative error, (1 + ε)^{1/n} − 1) and δ/n. Only inputs that are approximate in that candidate count. A summary several queries read takes the largest n. Stage 1 still reads no data workload.

• Whole-result scope. A per-value guarantee no longer meets a whole-result target directly. Stage 3 states it over the G values one evaluation returns, with δ_whole = G · δ_per (union bound). A whole-result guarantee meets a per-value target directly. Scope does not matter for a target without δ, or for a guarantee that never fails (δ = 0).

• G, the number of values one evaluation returns:

  • 1 for a scalar root, and for a root that returns at most one row (a global aggregate, PromQL or SQL, with only row-keeping operators or joins of such inputs above it);
  • the declared input cardinality for any other PromQL instant vector;
  • unbound for any other SQL result (until a max_groups[source] input exists) and for a range vector. An unbound G leaves a whole-result δ target unmet, with its own rejection reason.

• Stage 3 enlarges summaries before pricing, so that cost reflects the larger sketch. Each summary is sized to fail with probability at most δ / (n · G) for every query that reads it. The parameters enlarged are:

  • Count-Min depth;
  • Count Sketch depth (the binomial tail of its median);
  • KLL k at the smaller δ (its δ scale);
  • a distinct-count UnivMon's heap;
  • classic HLL's precision, from its source's distinct bound, up to 18.

  The selected plan (Selection.plan) and the selected logical roots carry the enlarged sketches.

• accuracy/allocation.rs is deleted. That removes AccuracyBudgetAllocator, EqualSplitAllocator, CompositionShape and AccuracyAllocation. accuracy/budget.rs (equal_share, equal_shares) replaces them on AccuracySpec. The legacy replacement search calls it directly, and ASAPStrategies::new_with_planning_inputs* and HydraGroupingStrategy::new_with_planning_inputs_and_evidence no longer take an allocator.

How

• compose_logical_candidate first composes each candidate with every summary sized for its whole budget. For each query whose guarantee is not exact, it then counts the sketch summary states below the root (approximate_inputs). If any query composes two or more, it composes the candidate again with each summary sized for its share.
• sized_for_failure_probability runs in select_plan before the accuracy check and pricing:
  1. computes the required per-value δ of every summary state;
  2. rebuilds those states with enlarged parameters (estimators::with_failure_probability, hll::precision_for);
  3. composes the guarantees above them again (with_resized_summaries; a node without a guarantee keeps none);
  4. re-exports the physical DAG, keeping Stage 2's placement.
• whole_result_guarantee wraps the per-value δ in ProbabilityExpr::Scaled { count: G }, or an unknown output_values_per_evaluation when G is unbound. satisfies gains a scope check.
• The legacy replacement search checks each node per value, which is how it sizes it, so its behavior is unchanged.

Key code interfaces added / modified

crates/asap-aware-mapping/src/accuracy/budget.rs (new), lib.rs

// new module, re-exported as accuracy::{equal_share, equal_shares}
/// The target of one approximate input (`shares = 1`), or of a nested sub-DAG
/// holding `shares` of the `of` inputs; also `None` for a non-positive or non-finite ε.
/// ε·shares/of (relative: (1 + ε)^{shares/of} − 1) and δ·shares/of; None for an exact target.
pub fn equal_shares(
    target: &AccuracyTarget,
    metric: ErrorMetric,
    of: usize,
    shares: usize,
) -> Option<AccuracyTarget>;
/// One of `approximate_inputs` equal shares of `target`.
pub fn equal_share(
    target: &AccuracyTarget,
    metric: ErrorMetric,
    approximate_inputs: usize,
) -> Option<AccuracyTarget>;   // equal_shares(target, metric, approximate_inputs, 1)

// removed: accuracy/allocation.rs, and its crate-root re-exports
// removed: pub struct CompositionShape { pub metric: ErrorMetric, pub approximate_layer_count: usize }
// removed: pub struct AccuracyAllocation { pub allocator: &'static str, pub layers: Vec<AccuracyTarget> }
// removed: pub fn AccuracyAllocation::inner_target(&self, shape: &CompositionShape) -> Option<AccuracyTarget>
// removed: pub trait AccuracyBudgetAllocator { fn allocations(&self, target: &AccuracyTarget, composition: &CompositionShape) -> Vec<AccuracyAllocation>; }
// removed: pub struct EqualSplitAllocator;

crates/asap-aware-mapping/src/accuracy/mod.rs

/// The deployment-extensible accuracy algebra; `DefaultAccuracyModel` is the shipped one.
/// The trait stays; only `satisfies` changes.
pub trait AccuracyModel {
    /// The registered rule for applying an exact operation to an approximate input;
    /// `None` (default) means the operation is exact only over exact inputs.
    fn exact_operation_rule(&self, operation: &ExactOperation) -> Option<CompositionOperator>;   // unchanged
    /// The guarantee of reading `query` from a summary of `family` built over an exact input;
    /// `None` when the model has no error model for the family.
    fn local_guarantee(&self, family: &FieldDataType, query: &SketchStatistic)
        -> Option<AccuracyGuarantee>;   // unchanged
    /// Composes the inputs' guarantees with the parent's own under `op`;
    /// `Err` is the fail-closed answer (no rule, or a missing input guarantee).
    fn propagate(&self, op: &CompositionOperator, inputs: &[AccuracyGuarantee],
        local: Option<&AccuracyGuarantee>, stats: &PropagationStats)
        -> Result<AccuracyGuarantee, AccuracyError>;   // unchanged
    /// Whether `guarantee` meets `target`.
    /// Now also checks the scope: a per-value δ does not meet a whole-result target.
    fn satisfies(&self, guarantee: &AccuracyGuarantee, target: &AccuracyTarget) -> bool;   // changed
}

/// Whether the guarantee's failure probability covers the values the target's does;
/// irrelevant without a δ requirement or for a guarantee that never fails.
fn scope_meets(guarantee: &AccuracyGuarantee, target: &AccuracyTarget) -> bool;   // new, private

/// The statistic an unbound output-value count `G` is left as; nothing binds it,
/// so such a failure probability stays unknown.
pub const OUTPUT_VALUES: &str = "output_values_per_evaluation";   // new; the statistic an unbound G stays as

/// `guarantee` stated over one evaluation's whole result, by the union bound;
/// a whole-result guarantee is returned as it is.
/// A per-value δ becomes G · δ (ProbabilityExpr::Scaled { count: G }); the error is unchanged.
pub fn whole_result_guarantee(
    guarantee: &AccuracyGuarantee,
    output_values: Option<f64>,   // G; None leaves δ symbolic in OUTPUT_VALUES
) -> AccuracyGuarantee;   // new

crates/asap-aware-mapping/src/stage3_plan_selection.rs

/// Why plan selection could not finish. The type stays; `Sizing` is new.
pub enum SelectionError {
    /// A candidate's roots, targets and recurrences differ in number: an internal bug.
    TargetCount { candidate: String, roots: usize, targets: usize, recurrences: usize },   // unchanged
    /// Pricing a node failed: an internal bug or a cost model without a price for it.
    Cost { candidate: String, node: PhysicalASAPNodeId, error: AnalyticalCostError },   // unchanged
    /// Every candidate was rejected; the rejections say why (an unsatisfiable request).
    NoValidCandidate(Vec<Rejection>),   // unchanged
    /// A query's requested accuracy cannot be resolved for its root, e.g. a quantile
    /// target with no metric: bad input.
    AccuracyRequest { query: usize, error: AccuracyRequestError },   // unchanged
    /// A candidate's summaries could not be rebuilt with the parameters its
    /// failure probabilities need; an internal bug, resizing should not break a schema.
    Sizing { candidate: String, error: SizingError },   // new
}

/// Why a candidate's summaries could not be rebuilt for a failure probability.
pub enum SizingError {   // new
    /// The resized DAG's schema derivation failed.
    Structure(#[from] SchemaDerivationError),
    /// The resized DAG's execution data states failed to validate.
    Export(#[from] ExecutionDataStateError),
}

/// The plan plan selection prices and runs, for the selected candidate's logical roots too.
/// The roots with each summary enlarged to fail with probability at most δ / (n · G).
pub fn sized_roots(
    roots: &[QueryRoot],
    targets: &[Option<AccuracyTarget>],
    data: &DataWorkload,
) -> Result<Vec<QueryRoot>, SchemaDerivationError>;   // new

// removed: fn root_guarantee(root: &QueryRoot) -> Option<AccuracyGuarantee>  (moved to per_computation_candidates)

crates/asap-aware-mapping/src/replacement.rs, grouping.rs

/// The Pass 1 summary-replacement strategies.
impl<'a> ASAPStrategies<'a> {
    /// A strategy with explicit cost and accuracy models; accuracy settles legality
    /// before cost ranks what is left.
    pub fn new_with_planning_inputs(
        cost_model: &'a dyn CostModel,
        accuracy_model: &'a dyn AccuracyModel,
    ) -> Self;   // was (cost_model, accuracy_model, allocator: &'a dyn AccuracyBudgetAllocator)
    /// Like `new_with_planning_inputs`, with typed planning-time evidence.
    pub fn new_with_planning_inputs_and_evidence(
        cost_model: &'a dyn CostModel,
        accuracy_model: &'a dyn AccuracyModel,
        evidence: &'a dyn AccuracyEvidenceProvider,
    ) -> Self;   // allocator parameter removed
}

/// The Hydra grouping strategy.
impl<'a> HydraGroupingStrategy<'a> {
    /// The strategy with explicit cost and accuracy models and planning-time evidence.
    pub fn new_with_planning_inputs_and_evidence(
        cost_model: &'a dyn CostModel,
        accuracy_model: &'a dyn AccuracyModel,
        evidence: &'a dyn AccuracyEvidenceProvider,
    ) -> Self;   // allocator parameter removed
}

// removed field: CandidatePlanningInputs::allocator: &'a dyn AccuracyBudgetAllocator

crates/asap-aware-mapping/src/accuracy/estimators/mod.rs, kll.rs, univmon.rs, hll.rs

// estimators/mod.rs: new; same family, enlarged to fail with probability at most delta
/// Enlarges, never reduces, so the error bound is unchanged or smaller; families
/// with no failure-probability parameter are returned as they are.
pub(crate) fn with_failure_probability(params: &SketchParams, delta: f64) -> SketchParams;

// kll.rs: new; k for the current rank error at delta, never smaller
/// A KLL resized to fail with probability at most `delta` at no larger rank error;
/// unchanged when it already fails no more often.
pub(crate) fn with_failure_probability(params: &SketchParams, delta: f64) -> SketchParams;

// univmon.rs: new; a distinct-count heap for the current error at delta, never smaller
/// A distinct-count UnivMon resized to fail with probability at most `delta`;
/// unchanged for a shape not sized for a distinct count.
pub(crate) fn with_failure_probability(params: &SketchParams, delta: f64) -> SketchParams;

// hll.rs: new; smallest precision from `precision` up whose classic failure probability is at most delta
/// The classic HLL precision for `delta` over a source of at most `max_distinct`
/// items; `precision` itself when none up to 18 meets it.
pub(crate) fn precision_for(precision: u8, max_distinct: u64, delta: f64) -> u8;

crates/asap-aware-mapping/src/per_computation_candidates.rs

/// Empty for an exact or unguaranteed value; plan selection sizes each input for its δ share.
/// Every sketch summary state a non-exact root value composes, each once.
pub(crate) fn approximate_inputs(root: &QueryRoot) -> Vec<Rc<OperatorNode>>;   // new
/// The composed theoretical guarantee of a query's output.
pub(crate) fn root_guarantee(root: &QueryRoot) -> Option<AccuracyGuarantee>;   // moved from stage3_plan_selection
/// `None` from `resize` keeps a summary's parameters; timing, materialization and sharing are kept.
/// Rebuilds summaries with the parameters `resize` gives and recomposes guarantees above them.
pub(crate) fn with_resized_summaries(
    roots: &[QueryRoot],
    resize: &dyn Fn(&OperatorNode, &SketchParams) -> Option<SketchParams>,
) -> Result<Vec<QueryRoot>, SchemaDerivationError>;   // new
/// The one source whose distinct bound gives a summary's failure probability;
/// `None` for a state over computed rows or several sources.
pub(crate) fn summary_source(state: &OperatorNode) -> Option<String>;   // new

Usage, end to end (checked by running the planning stages at a18fa4a6 with a scratch test): a p99 panel over 10,000 series with a whole-result rank target.

// What the user writes: one workload entry. `entry` holds the query, how often it runs
// (every 10 s over the last 5 min) and its target: rank error ε = 0.01 with δ = 0.01,
// over the whole result (the default scope: every returned series at once). Stage 1
// sizes summaries from this target; Stage 3 checks every candidate against it.
let entry = RepeatingEntry {
    query: Query("quantile_over_time(0.99, latency_seconds[5m])".into()),
    demand: RepeatedDemand::FixedInterval(RepetitionInterval(10_000)),
    requirements: QueryRequirements {
        accuracy: AccuracyRequirement::Explicit(
            RequestedAccuracy::epsilon_delta(0.01, 0.01).with_metric(ErrorMetric::Rank),
        ),
        response_latency: LatencyRequirement::Unspecified,
    },
    predictability: Predictability::Predictable { known_at: None },
    time_selection: TimeSelection {
        scope: QueryTimeScope::RealTime,
        lookback: Some(DurationMs(300_000)),
        as_of: None,
    },
};
// `data` holds the deployment's facts. `input_cardinality` = 10,000 series is G, the
// number of values one evaluation of this PromQL vector returns: Stage 3 uses it to
// state a per-series δ over the whole result. Without it, G is unbound and the
// whole-result target is unmet.
let data = DataWorkload {
    arrival: DataArrival::ContinuouslyIngesting,
    data_ingestion_interval: declared(DurationMs(15_000)),
    input_cardinality: declared(10_000),
    ..Default::default()
};
// `workload` is the planner's input: the entry plus the data facts.
let workload = PlanningWorkload {
    query_workload: QueryWorkload {
        language: QueryLanguage::PromQL,
        query_batch: None,
        repeating_queries: Some(vec![entry]),
    },
    data_workload: Some(data),
};
// `roots` holds the PromQL frontend's lowered query, one root per entry; the stages plan over it.
let roots = lower_promql_roots(&workload)?;
// `run` holds every stage's output:
// - Stage 1, Pass 1 offers L1 exact, L2 KLL and L3 DDSketch. The query reads one
//   approximate input (n = 1), so its budget is not split: the KLL is sized per series
//   for the whole (0.01, 0.01), k = 269. A query that combines n ≥ 2 approximate inputs
//   (e.g. `count(a) - count(b)`) is composed again with each input sized for
//   equal_share(its target, its metric, n): ε/n and δ/n.
// - Stage 3 (select_plan) calls whole_result_guarantee(the KLL's per-series guarantee,
//   Some(10_000.0)), which states its δ over all 10,000 series (10,000 · δ). To meet
//   δ = 0.01, sized_roots(P2's roots, the targets, data) first enlarges each series' KLL
//   to δ = 0.01 / 10,000 = 1e-6 (k = 521); DefaultAccuracyModel.satisfies then checks
//   the result, and the enlarged sketch is what gets priced.
let run = run_planning_stages(&workload, roots, None)?;
// `selection` is Stage 3's answer: the selected candidate, the costs of the valid ones
// and why each other candidate lost.
let selection = run.stage3_plan_selection?;
assert_eq!(selection.selected, "P2");   // the KLL plan, 0.188 cost_per_second
// P1 (exact): valid but costlier, 0.216. P3 (DDSketch): invalid, its relative bound misses a rank target.
// selection.plan, the plan to run, carries Kll { k: 521, failure_probability: Some(1e-6) }.

Before this PR

• count by (job) (a) - count by (job) (b) with ε = 0.01, δ = 0.001: both Count-Min sketches were 272 × 7, sized for the full budget.
• quantile_over_time(0.99, x[5m]) over 10 000 series with rank ε = 0.01, δ = 0.01: KLL was accepted with a per-value δ of 0.01 against a whole-result target.

After this PR

• The two Count-Min sketches are 544 × 8 each (ε/2 = 0.005, δ/2 = 0.0005).

• The KLL is enlarged to fail with probability 1e-6 per series (δ/G), and that larger sketch is what gets priced and selected. A per-value target keeps the Stage 1 size.

• docs: propose workload-wide planning, summary sharing, and materialization #509 examples (stage_pipeline):

  Example	Selection and costs	What changes
  1	Unchanged: all-exact selected at 7.940	Q2's CMS+heap and CountSketch+heap are still rejected, because top-k membership has no guarantee. Had they been certified, CMS depth would grow from 7 to 21 (δ = 1e-9 over 1M series).
  3a	Unchanged: P485, KLL ×5, 21724.83	The selected KLLs grow from k = 547 to 1246 at δ = 1e-8 per series. The analytical cost prices a query-time KLL build as rows × depth 1, independent of k.
  3b	Unchanged: P2, KLL, 2.100	The KLL grows from k = 269 to 613 at δ = 1e-8.
  2 (SQL)	Same as main: 82 plans, 70 invalid, P1 selected	Q1–Q3 are global aggregates, so G = 1.

  No row of docs/user_guide_docs/stage-viewer.md changes.

Verification

• New tests:
  • equal split (accuracy::budget);
  • a combination shares the budget, and an exact input takes no share (per_computation_candidates);
  • per-value guarantees against whole-result targets over G values, and scope with δ = 0 or no δ (accuracy; this replaces the test that pinned the direct comparison);
  • enlargement meets δ at no larger error for Count-Min, Count Sketch, KLL and UnivMon distinct, and HLL precision for its distinct bound (estimators);
  • Stage 3: a whole-result target enlarges KLL for G, and per-value does not; an unbound G leaves the target unmet; a root of at most one row returns one value (written failing first), and a grouped SQL root stays unbound.
• cargo nextest run for asap-aware-mapping, asap-planner, asap-integration-tests, asap-devtools and asap-types: 1107 passed, 34 skipped. That includes feat(logical-asap-optimization): Example 2 — summary-capability rule and UnivMon frequency statistics #707's stage3_plan_selection_accepts_sql_univmon_l2_beside_its_exact_population_count.
• cargo fmt --all -- --check and cargo clippy --workspace --all-targets --all-features --locked -- -D warnings: clean.
• tools/verify_metricsql_external_consumer.sh: passes.

Test changes

• Example 2's PromQL stand-in states its targets per value. It reads one series, but its declared input cardinality (10M) counts source IPs, which plan selection would take as G.
• HLL over a distinct bound of 10 is now accepted at a larger precision. The test is renamed stage3_plan_selection_enlarges_hll_for_a_larger_distinct_bound.
• Estimator sizing tests state per-value targets.

Limitations (all conservative: unmet, never wrongly accepted)

• A sum or other fold over approximate per-series values multiplies δ by its rows. The δ / (n · G) sizing does not cover that factor, so such a whole-result target can stay unmet.
• KMV and Theta (a fixed 99% calibration) and an L2 or entropy UnivMon are not enlarged.
• Count Sketch with a non-power-of-two width reaches its 255-row cap before about δ = 1e-4.
• Classic HLL certifies only small distinct bounds: a bound of 100 is met at no precision.
• A shared summary instance is counted once per input it serves (δ/n each); counting it once is Count a shared summary instance once in the failure-probability split (#669 follow-up) #711.
• The viewer's Stage 2 lane shows the Stage 1 sizes. The Stage 3 plan and Selection.plan carry the enlarged ones.
• docs/design_docs/concepts/accuracy-models.md still describes AccuracyBudgetAllocator; doc updates are tracked in Update the docs after the G1–G11 consolidation groups land #677.

Follow-ups

• Count a shared summary instance once in the failure-probability split (#669 follow-up) #711: count a shared summary instance once in the failure-probability split.
• A max_groups[source] input to bind G for grouped SQL results.

[zzylol]

Planned simplifications of this PR's interfaces. They land in the follow-up PR after #717, together with the AccuracyTarget / AccuracyMetric renames:

• equal_shares(target, metric, of, shares) and equal_share(target, metric, n) become one function, per_input_target(target, metric, approximate_inputs). Only the legacy replacement core uses shares > 1, and it is being deleted.
• whole_result_guarantee(guarantee, output_values: Option<f64>) becomes guarantee.over_whole_result(values_per_result: Option<u64>), and OUTPUT_VALUES becomes VALUES_PER_RESULT.
• SizingError and SelectionError::Sizing are removed. If resizing a candidate for its δ fails, that candidate is rejected (a Rejection with the reason) instead of the whole select_plan failing.
• Names that say what the function does: sized_roots becomes resize_for_failure_probability, summary_source becomes distinct_bound_source, and hll::precision_for(precision, ε, max_distinct, δ) becomes hll::min_precision(ε, δ, max_distinct, at_least).
• The two ASAPStrategies constructors belong to the legacy core and go away when it is deleted.