refactor(stage3): constraint (K), accuracy at each query root (#1207 PR 8)

crates/accuracy/src/estimators/hll.rs

@@ -81,7 +81,6 @@ pub fn parse_classic_failure_statistic(statistic: &str) -> Option<(u8, f64, &str
 /// distinct bound ([`classic_failure_statistic`]). The failure probability
 /// is per output row: each group's estimate fails with at most it. `None`
 /// for a precision outside the contract's.
-#[allow(dead_code)] // Read again in #1207 PR 8: δ of a distinct count over a source's own rows, with θ.
 pub(crate) fn classic_guarantee(
     precision: u8,
     relative_error: Option<f64>,

crates/accuracy/src/estimators/mod.rs

@@ -86,6 +86,49 @@ pub fn local_guarantee(
         | FieldDataType::StatModel(..) => None,
     }
 }
+/// The guarantee of a distinct count read from a summary with parameters
+/// `parameters` built over the own rows of one source `source` (behind
+/// filters and time windows at most), whose distinct items per evaluation
+/// are at most `max_distinct`
+/// (`SourceFacts::max_distinct_per_evaluation`): classic HLL's
+/// ([`hll::classic_guarantee`]) or a distinct-count UnivMon's
+/// ([`univmon::distinct_guarantee`]), with its failure probability bound
+/// at `max_distinct` (1, no certificate, where the contract does not cover
+/// it). `None` for other parameters and for a shape the contracts do not
+/// cover; the generic guarantee of [`local_guarantee`] then applies, whose
+/// failure probability is unknown.
+pub fn distinct_count_guarantee_over_source(
+    parameters: &SketchParams,
+    source: &str,
+    max_distinct: u64,
+) -> Option<AccuracyGuarantee> {
+    let guarantee = match parameters {
+        SketchParams::Hll {
+            precision,
+            relative_error,
+        } => hll::classic_guarantee(*precision, *relative_error, source)?,
+        SketchParams::UnivMon { .. } => univmon::distinct_guarantee(parameters, source)?,
+        _ => return None,
+    };
+    Some(guarantee.bind(&|statistic| {
+        if let Some((precision, relative_error, _)) =
+            hll::parse_classic_failure_statistic(statistic)
+        {
+            return Some(hll::classic_failure_probability(
+                precision,
+                relative_error,
+                max_distinct,
+            ));
+        }
+        let (heap_size, layers, _) = univmon::parse_incomplete_top_layer_statistic(statistic)?;
+        Some(univmon::incomplete_top_layer_probability(
+            heap_size,
+            layers,
+            max_distinct,
+        ))
+    }))
+}
+
 /// `params` with the parameters its per-value failure probability depends
 /// on enlarged, never reduced, so that each value it answers fails with
 /// probability at most `delta` and its error bound is unchanged or smaller
@@ -587,4 +630,36 @@ mod tests {
             assert!(local_guarantee(&exact, None, &total).is_some(), "{exact:?}");
         }
     }
+
+    /// A distinct count over a source's own rows gets classic HLL's or a
+    /// distinct-count UnivMon's failure probability at the source's
+    /// distinct bound: certified (below 1) for a small bound, 1 (no
+    /// certificate) past the contract's range; other parameters get none.
+    #[test]
+    fn a_distinct_count_over_a_source_binds_its_failure_probability() {
+        let hll = asap_types::stages::parameter_list(SketchAlgorithm::Hll)[0].clone();
+        let univmon = asap_types::stages::parameter_list(SketchAlgorithm::UnivMon)
+            .iter()
+            .find(|p| {
+                matches!(
+                    p,
+                    SketchParams::UnivMon {
+                        distinct_failure_probability: Some(_),
+                        ..
+                    }
+                )
+            })
+            .unwrap()
+            .clone();
+        for parameters in [&hll, &univmon] {
+            let small = distinct_count_guarantee_over_source(parameters, "flows", 100).unwrap();
+            assert_eq!(small.spec.accuracy_metric, AccuracyMetric::Relative);
+            let delta = small.spec.failure_probability.evaluate().unwrap();
+            assert!(delta < 1.0, "{parameters:?}: {delta}");
+            let huge = distinct_count_guarantee_over_source(parameters, "flows", u64::MAX).unwrap();
+            assert_eq!(huge.spec.failure_probability.evaluate(), Some(1.0));
+        }
+        let kll = asap_types::stages::parameter_list(SketchAlgorithm::Kll)[0].clone();
+        assert!(distinct_count_guarantee_over_source(&kll, "flows", 1_000).is_none());
+    }
 }

crates/accuracy/src/estimators/univmon.rs

@@ -219,7 +219,6 @@ fn distinct_heap(eps: f64, delta: f64) -> u32 {
 
 /// The statistic a UnivMon distinct count over `source` leaves its
 /// top-layer failure probability as.
-#[allow(dead_code)] // Read again in #1207 PR 8: δ of a distinct count over a source's own rows, with θ.
 pub(crate) fn incomplete_top_layer_statistic(heap_size: u32, layers: u8, source: &str) -> String {
     format!("{INCOMPLETE_TOP_LAYER}[k={heap_size},layers={layers}]({MAX_DISTINCT}[{source}])")
 }
@@ -264,7 +263,6 @@ pub fn incomplete_top_layer_probability(heap_size: u32, layers: u8, max_distinct
 /// `δ_c` plus the chance of an incomplete top layer, left symbolic in the
 /// source's distinct bound. `None` for a shape that was not sized for a
 /// distinct count, a heap under 8, or a shape the library would not build.
-#[allow(dead_code)] // Read again in #1207 PR 8: δ of a distinct count over a source's own rows, with θ.
 pub(crate) fn distinct_guarantee(params: &SketchParams, source: &str) -> Option<AccuracyGuarantee> {
     let SketchParams::UnivMon {
         heap_size,

crates/devtools/src/planner_fuzz/mod.rs

@@ -458,8 +458,8 @@ impl Planner {
     /// summary at the θ its consumers' parameter floors allow. The cost is
     /// that plan's, and `derived_root` the nodes of the realized plan only
     /// the last root reads. `last_root_is_key_filter` is accepted for the
-    /// relations' interface; accuracy is not composed again here (constraint
-    /// (K), #1207 PR 8).
+    /// relations' interface; Stage 3 checks accuracy at the roots
+    /// (constraint (K)).
     pub fn price(
         &self,
         workload: &FuzzWorkload,

crates/devtools/src/planning_stages.rs

@@ -348,7 +348,7 @@ pub fn run_stages_until(
         &physical,
         entries,
         data,
-        &models.accuracy_check_inputs(),
+        &models.accuracy_check_inputs(&roots),
         models.cost_inputs,
         &realizer,
     ) {

crates/planner/src/pass.rs

@@ -16,7 +16,7 @@
 use std::collections::BTreeMap;
 use std::rc::Rc;
 
-use asap_types::ir::OperatorNode;
+use asap_types::ir::{OperatorNode, QueryRoot};
 use asap_types::workload::parsed_workload::ParsedWorkload;
 use asap_types::workload::WorkloadError;
 
@@ -105,16 +105,21 @@ impl<'a> PlanningModels<'a> {
         }
     }
 
-    /// Stage 3's accuracy check inputs: `evidence`, and `accuracy`, or
+    /// Stage 3's accuracy check inputs over the front ends' `query_roots`
+    /// (by query): `evidence`, and `accuracy`, or
     /// [`IgnoreAccuracyGuarantees`] with `ignore_accuracy_guarantees`.
-    pub fn accuracy_check_inputs(&self) -> AccuracyCheckInputs<'a> {
+    pub fn accuracy_check_inputs<'b>(&self, query_roots: &'b [QueryRoot]) -> AccuracyCheckInputs<'b>
+    where
+        'a: 'b,
+    {
         AccuracyCheckInputs {
             model: if self.ignore_accuracy_guarantees {
                 &IGNORE_ACCURACY_GUARANTEES
             } else {
                 self.accuracy
             },
             evidence: self.evidence,
+            query_roots,
         }
     }
 

crates/planner/src/run.rs

@@ -179,7 +179,7 @@ pub fn run_planning(
             physical,
             entries,
             data,
-            &models.accuracy_check_inputs(),
+            &models.accuracy_check_inputs(&roots),
             models.cost_inputs,
             &realizer,
         );

crates/planner/tests/accuracy_at_query_roots.rs

@@ -0,0 +1,160 @@
+//! #1207 PR 8: constraint (K), each query's composed accuracy at its root,
+//! end to end through the facade on #1207's problem example.
+
+use asap_frontend_promql::promql_metric_names;
+use asap_planner::{e2e_plan, FrontendInput, PlanOutput, PlanningModels, UserInput};
+use asap_types::ir::properties::{AccuracyMetric, AccuracyScope, AccuracyTarget};
+use asap_types::ir::schema::SketchParams;
+use asap_types::workload::{
+    DataArrival, DataWorkload, DurationMs, PlanningWorkload, Predictability, Query, QueryLanguage,
+    QueryRequirements, QueryWorkload, RepeatedDemand, RepeatingEntry, RepetitionInterval,
+    TimestampMs,
+};
+
+/// PromQL queries with their accuracy targets, every 10 s at phase 0, over
+/// stored samples of `latency` (1,000 series sampled every 15 s, values
+/// declared non-negative).
+fn workload(queries: &[(&str, AccuracyTarget)]) -> PlanningWorkload {
+    let mut data = DataWorkload {
+        arrival: DataArrival::AtRest,
+        ..DataWorkload::sampled_every(
+            queries
+                .iter()
+                .flat_map(|(query, _)| promql_metric_names(query).unwrap_or_default()),
+            DurationMs(15_000),
+        )
+        .with_series_count(1_000)
+    };
+    for facts in data.sources.values_mut() {
+        facts.values_non_negative = true;
+    }
+    PlanningWorkload {
+        query_workload: QueryWorkload {
+            language: QueryLanguage::PromQL,
+            query_batch: None,
+            repeating_queries: Some(
+                queries
+                    .iter()
+                    .map(|(query, accuracy)| RepeatingEntry {
+                        query: Query((*query).into()),
+                        demand: RepeatedDemand::FixedIntervalAt {
+                            interval: RepetitionInterval(10_000),
+                            evaluation_phase: TimestampMs(0),
+                        },
+                        requirements: QueryRequirements {
+                            accuracy: accuracy.clone(),
+                            max_latency_ms: None,
+                        },
+                        predictability: Predictability::Unknown,
+                        range_evaluation: None,
+                    })
+                    .collect(),
+            ),
+        },
+        data_workload: Some(data),
+    }
+}
+
+async fn plan(queries: &[(&str, AccuracyTarget)], ignore_accuracy_guarantees: bool) -> PlanOutput {
+    let workload = workload(queries);
+    let models =
+        PlanningModels::builtin().with_ignore_accuracy_guarantees(ignore_accuracy_guarantees);
+    e2e_plan(UserInput::new(&workload, FrontendInput::Promql, models))
+        .await
+        .expect("workload plans")
+}
+
+/// #1207's problem example: A, the p95 at rank error 0.01, and B, the
+/// ratio of the p99 and the p50 at relative error 0.03.
+fn problem_example() -> [(&'static str, AccuracyTarget); 2] {
+    [
+        (
+            "quantile_over_time(0.95, latency[1m])",
+            AccuracyTarget::max_error(0.01).accuracy_metric(AccuracyMetric::Rank),
+        ),
+        (
+            "quantile_over_time(0.99, latency[1m]) / quantile_over_time(0.5, latency[1m])",
+            AccuracyTarget::max_error(0.03).accuracy_metric(AccuracyMetric::Relative),
+        ),
+    ]
+}
+
+/// The θ of the selected plan's sketch builds.
+fn sketch_parameters(output: &PlanOutput) -> Vec<SketchParams> {
+    output
+        .realized_plan
+        .as_ref()
+        .expect("a realized plan")
+        .plan
+        .dag
+        .nodes
+        .iter()
+        .filter_map(|node| node.parameters.clone())
+        .collect()
+}
+
+/// A meets rank error 0.01 with a KLL (a DDSketch readout's relative error
+/// does not meet a rank target); B reads one DDSketch through both
+/// operands, with `alpha` small enough that the ratio's composed bound
+/// $(\alpha + \alpha) / (1 - 0.1)$ is within 0.03, which `alpha` 0.02 (each
+/// operand alone within 0.03) is not.
+#[tokio::test]
+async fn each_query_meets_its_target_at_its_root() {
+    let output = plan(&problem_example(), false).await;
+    let selection = output.selection.as_ref().expect("a plan is selected");
+    let accepted = &selection.accepted_accuracy_per_query;
+    assert_eq!(accepted[0].accepted_metric, Some(AccuracyMetric::Rank));
+    assert_eq!(accepted[0].sketches, ["Kll"]);
+    assert_eq!(accepted[1].accepted_metric, Some(AccuracyMetric::Relative));
+    assert_eq!(accepted[1].sketches, ["DDSketch"]);
+    let parameters = sketch_parameters(&output);
+    let alphas: Vec<f64> = parameters
+        .iter()
+        .filter_map(|p| match p {
+            SketchParams::DDSketch { alpha } => Some(*alpha),
+            _ => None,
+        })
+        .collect();
+    assert_eq!(
+        alphas.len(),
+        1,
+        "one DDSketch for both operands: {parameters:?}"
+    );
+    assert!((alphas[0] + alphas[0]) / (1.0 - 0.1) <= 0.03, "{alphas:?}");
+}
+
+/// Ignoring accuracy guarantees, (K) applies to no query: every output is
+/// accepted regardless of its guarantee.
+#[tokio::test]
+async fn ignoring_accuracy_guarantees_drops_k() {
+    let ignored = plan(&problem_example(), true).await;
+    let selection = ignored.selection.expect("a plan");
+    assert!(selection
+        .accepted_accuracy_per_query
+        .iter()
+        .all(|accepted| accepted.accepted_regardless_of_guarantee));
+}
+
+/// The listed sizes fail with probability 0.01 per value, so a
+/// whole-result target of 0.01 over 1,000 series allows 1e-5 per value:
+/// no sketch meets it, and the plan reads the raw data exactly.
+#[tokio::test]
+async fn a_whole_result_failure_probability_is_divided_over_the_values() {
+    let output = plan(
+        &[(
+            "quantile_over_time(0.95, latency[1m])",
+            AccuracyTarget::max_error(0.01)
+                .accuracy_metric(AccuracyMetric::Rank)
+                .failure_probability(0.01)
+                .accuracy_scope(AccuracyScope::WholeResult),
+        )],
+        false,
+    )
+    .await;
+    let selection = output.selection.as_ref().expect("a plan is selected");
+    assert_eq!(
+        selection.accepted_accuracy_per_query[0].accepted_metric,
+        None
+    );
+    assert!(sketch_parameters(&output).is_empty());
+}

crates/planner/tests/main.rs

@@ -4,6 +4,7 @@
 
 mod time_series;
 
+mod accuracy_at_query_roots;
 mod columnar_batches;
 mod e2e_plan;
 mod execution_contexts;

crates/stage3-plan-selection/src/lib.rs

@@ -12,8 +12,12 @@
 //! $y$ exists only on admissible edges) and (R). It decodes $\{x = 1\}$ and
 //! $\{y = 1\}$ into a `DagOfSelectedCandidates` (one node per selected
 //! execution), which the injected realizer builds. Accuracy at each query
-//! root, constraint (K), with per-candidate error bounds $E_c$ and $D_c$
-//! (section 4), is #1207 PR 8. Costs come from `asap-cost`.
+//! root is constraint (K) (section 4): error bounds $E_{\kappa,c,m}$ per
+//! execution and metric and failure probabilities $D_{\kappa,c}$, big-M on
+//! the selected edges, root checks against each query's target
+//! ([`crate::milp::error_bounds`]); the selected plan's composed guarantee
+//! is checked again with the deployment's accuracy model. Costs come from
+//! `asap-cost`.
 
 mod execution_cost;
 mod milp;