feat(planner): certify distinct counts at scale: HLL sized from (ε, δ), provable KMV (#694)

[zzylol]

Closes #694. Closes #700.

Why

#693 gave classic HLL an (ε, δ) guarantee from a deployment bound on distinct items, but Pass 1 still sized precision from ε alone. Precision 12 at ε = 0.02 certified δ = 0.01 for at most 9 distinct items, and the contract stopped at 4096. KMV reported a fixed 0.01 failure probability whatever δ it was sized for (#700). Plan selection then rejected it for a stricter δ, and its k was ten times larger than a provable bound needs.

What

• HLL sized from (ε, δ) (B). Pass 1 takes the smallest precision with m ≥ 1/(2ε²δ) at which feat(planner): HLL confidence from a deployment distinct-count bound (#678) #693's tail formula certifies δ at the target ε for every bound up to ⌈1/ε⌉. It reads no data.
  • SketchParams::Hll records relative_error, the ε it was sized for (None: not sized; the error bound is then 1.04/√m). Equality ignores it, as with KLL's failure_probability.
  • The guarantee states the target ε as its error bound, not 1.04/√m. Its failure probability is evaluated at that ε: classic_hll_failure_probability[p=P,eps=E](max_distinct[source]).
  • Stage 3 still binds max_distinct[source] for the upper coverage limit.
  • The contract's cap rises from 4096 to 2^17, which is m/2 at precision 18. Evaluating a bound is O(bound).
• One sizing rule with feat(planner): error-budget allocation and whole-result δ scaling (#669, part b) #712. Stage 3's enlargement uses the same rule, the smallest p whose tail at the recorded ε meets δ, applied to the source's bound and δ/(n·G). It starts from Pass 1's precision and only raises it, so nothing is sized twice.
• Provable KMV (D). The bound is checked against asap_sketchlib's estimator (src/sketches/kmv.rs). The estimator is exact while it keeps fewer than k hashes. From then on it returns (k−1)/U, where U is the k-th smallest hash mapped into (0, 1]. A merge keeps the k smallest distinct hashes of the union. Let X(u) ~ Binomial(N, u) count the hashes below u:
  • Overestimate: (k−1)/U > (1+ε)N means X(u) ≥ k at u = (k−1)/((1+ε)N), where the mean μ = (k−1)/(1+ε) < k/(1+ε). Chernoff gives P ≤ exp(−ε²(k−1)/((1+ε)(2+ε))).
  • Underestimate: (k−1)/U < (1−ε)N means X(u′) ≤ k−1 = (1−ε)μ′. Chernoff gives P ≤ exp(−ε²(k−1)/(2(1−ε))), and 0 for ε ≥ 1.
  • The failure probability per output row is the sum of the two. It does not depend on N, so no distinct bound is needed.
  • SketchParams::Kmv records relative_error. k is the smallest k meeting δ, about 1 + (1+ε)(2+ε)·ln(2/δ)/ε², which is below the issue's 3·ln(2/δ)/ε² + 1.
  • Stage 3 enlarges k for a smaller per-value δ at the same ε. Theta keeps its Chebyshev calibration (theta_k_99).
• Review nits from feat(planner): HLL confidence from a deployment distinct-count bound (#678) #693. Negative tests cover the HLL scope condition: a join, computed rows, Hydra grouping, and a bound keyed to another source. max_distinct_per_evaluation's doc now says the bound must hold for the longest evaluation window any query over the source uses, and that it and δ are per output row.

How

• hll::hll_precision(ε, δ) and hll::precision_for(p, ε, bound, δ) share classic_failure_probability(p, ε, bound).
• cardinality::kmv_failure_probability(k, ε) and kmv_k(ε, δ). kmv_k binary-searches between the closed forms at ln(1/δ) and ln(2/δ).
• with_failure_probability gains the KMV case.

Key code interfaces added / modified

crates/types/src/post_asap/sketch.rs

/// Concrete parameters of one sketch instance; equality compares the built shape.
/// The type stays; this PR adds a field to `Hll` and `Kmv`.
pub enum SketchParams {
    /// A UnivMon universal sketch (frequency moments, entropy, distinct count).
    UnivMon {                                        // unchanged
        /// Top-k heap entries per layer.
        heap_size: u32,                              // unchanged
        /// Rows of each layer's Count Sketch.
        sketch_rows: u32,                            // unchanged
        /// Columns of each layer's Count Sketch.
        sketch_cols: u32,                            // unchanged
        /// Number of sampling layers.
        layers: u8,                                  // unchanged
        /// The δ the heap was sized for when it answers a distinct count; None = no distinct guarantee.
        distinct_failure_probability: Option<f64>,   // unchanged
    },
    /// A KLL quantile sketch.
    Kll {                                            // unchanged
        /// Compactor capacity.
        k: u32,                                      // unchanged
        /// The δ `k` was sized for when below 0.01; None = the empirical 99% calibration.
        failure_probability: Option<f64>,            // unchanged
    },
    /// A Count-Min sketch.
    Cms {                                            // unchanged
        /// Counters per row.
        width: u32,                                  // unchanged
        /// Rows.
        depth: u32,                                  // unchanged
    },
    /// A HyperLogLog distinct-count sketch.
    Hll {                                            // unchanged variant; gains relative_error
        /// `m = 2^precision` registers.
        precision: u8,                               // unchanged
        /// The relative error `precision` was sized for: its guarantee's error bound,
        /// at which its failure probability is evaluated.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        relative_error: Option<f64>,   // new: the ε precision was sized for; None = RSE 1.04/√m; ignored by equality
    },
    /// A DDSketch relative-error quantile sketch.
    DDSketch {                                       // unchanged
        /// Relative accuracy of each bucket.
        alpha: f64,                                  // unchanged
    },
    /// A Count-Min sketch with a top-k heap.
    CmsWithHeap {                                    // unchanged
        /// Counters per row.
        width: u32,                                  // unchanged
        /// Rows.
        depth: u32,                                  // unchanged
        /// Heap entries.
        heap_size: u32,                              // unchanged
    },
    /// A KMV (k minimum values) distinct-count sketch.
    Kmv {                                            // unchanged variant; gains relative_error
        /// The number of smallest hashes kept.
        k: u32,                                      // unchanged
        /// The relative error `k` was sized for: its guarantee's error bound, at which
        /// its failure probability is evaluated.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        relative_error: Option<f64>,   // new: the ε k was sized for; None = RSE 1/√(k-2); ignored by equality
    },
    /// A Theta distinct-count sketch.
    Theta {                                          // unchanged
        /// Nominal entries.
        k: u32,                                      // unchanged
    },
    /// A Count Sketch.
    CountSketch {                                    // unchanged
        /// Counters per row.
        width: u32,                                  // unchanged
        /// Rows.
        depth: u32,                                  // unchanged
    },
    /// A Count Sketch with a top-k heap.
    CountSketchWithHeap {                            // unchanged
        /// Counters per row.
        width: u32,                                  // unchanged
        /// Rows.
        depth: u32,                                  // unchanged
        /// Heap entries.
        heap_size: u32,                              // unchanged
    },
}

crates/asap-aware-mapping/src/accuracy/estimators/hll.rs

/// Pass 1's classic HLL precision for relative error `eps` at failure probability
/// `delta`, without data; plan selection later enlarges it for the source's bound.
pub(crate) fn hll_precision(eps: f64, delta: f64) -> u8;   // was hll_precision(eps)
/// The smallest precision from `precision` up whose failure probability at
/// `relative_error` over at most `max_distinct` items is at most `delta` (18 at most).
pub(crate) fn precision_for(
    precision: u8,
    relative_error: f64,   // new parameter
    max_distinct: u64,
    delta: f64,
) -> u8;
/// The relative error a classic HLL's guarantee states: the sized one, else its
/// relative standard error.
pub(crate) fn classic_relative_error(precision: u8, relative_error: Option<f64>) -> f64;   // new
/// Classic HLL's distinct-count guarantee over `source`'s own rows, with δ left
/// symbolic in the source's distinct bound; `None` for a precision outside the contract.
pub(crate) fn classic_guarantee(
    precision: u8,
    relative_error: Option<f64>,   // new parameter
    source: &str,
) -> Option<AccuracyGuarantee>;
/// The failure probability given the source's distinct bound; 1 (no certificate)
/// outside the contract's range.
pub(crate) fn classic_failure_probability(
    precision: u8,
    relative_error: f64,   // new parameter
    max_distinct: u64,
) -> f64;
// "classic_hll_failure_probability[p=P,eps=E](max_distinct[src])"; was "...[p=P](max_distinct[src])"
/// The statistic name classic HLL's failure probability is left as, for plan
/// selection to bind.
pub(crate) fn classic_failure_statistic(precision: u8, relative_error: f64, source: &str) -> String;
/// The precision, relative error and source such a statistic names; `None` otherwise.
pub(crate) fn parse_classic_failure_statistic(statistic: &str) -> Option<(u8, f64, &str)>;   // was Option<(u8, &str)>

/// The largest distinct bound the contract supports at any precision (`m/2` at precision 18).
const MAX_CONTRACT_DISTINCT: u32 = 1 << 17;   // ClassicHllConfidence::new's distinct-bound cap; was 4096

crates/asap-aware-mapping/src/accuracy/estimators/cardinality.rs, mod.rs

/// The smallest `k` the KMV estimate `(k-1)/U` is defined for.
const KMV_MIN_K: u32 = 2;         // new
/// The largest `k` sizing returns.
const KMV_MAX_K: u32 = 1 << 26;   // new
/// The probability KMV with `k` misses relative error `eps` on any number of
/// distinct items; 1 when it certifies nothing.
pub(crate) fn kmv_failure_probability(k: u32, eps: f64) -> f64;   // new
/// The smallest `k` whose failure probability at `eps` is at most `delta`, up to 2^26.
pub(crate) fn kmv_k(eps: f64, delta: f64) -> u32;                 // new
/// Theta's `k` for relative error `eps` at 99% confidence (Chebyshev).
pub(crate) fn theta_k_99(eps: f64) -> u32;                        // renamed from kmv_k_99; Theta only
// KMV contract: "kmv_exact_below_k_chernoff_v1" (was "kmv_unbiased_variance_chebyshev_99_v1")

// mod.rs
/// Sizes a sketch of `kind` for one aggregate intent's `(eps, delta)`.
pub(crate) fn size_params(kind: SketchAlgorithm, intent: &AggIntent, eps: f64, delta: f64) -> SketchParams;
    // Hll: hll_precision(eps, delta) with relative_error: Some(eps); Kmv: kmv_k(eps, delta) with relative_error: Some(eps)
/// `params` enlarged, never reduced, to fail with probability at most `delta`;
/// parameters without such a knob are returned as they are.
pub(crate) fn with_failure_probability(params: &SketchParams, delta: f64) -> SketchParams;
    // new case: Kmv { relative_error: Some(eps), .. } enlarges k to kmv_k(eps, delta)
/// The parameters of one build shared by several readers; `None` for different
/// algorithms or an unbuildable shared shape.
pub(crate) fn shared_size_params(params: &[&SketchParams]) -> Option<SketchParams>;
    // Hll/Kmv: keeps the smaller recorded relative_error

Usage, end to end: one SQL distinct count with ε = 0.05 and δ = 0.01, planned through run_planning_stages at this PR's merge commit. The numbers in the comments come from running this code there.

// 1. What the user writes: one dashboard panel, the number of distinct source IPs
//    in the last minute, refreshed every 10 s, within ε = 0.05 relative error
//    with failure probability δ = 0.01.
let panel = RepeatingEntry {
    query: Query("SELECT COUNT(DISTINCT src_ip) FROM flows WHERE ts >= now() - INTERVAL '1 minute'".into()),
    demand: RepeatedDemand::FixedInterval(RepetitionInterval(10_000)),
    requirements: QueryRequirements {
        // The target every candidate plan's guarantee is checked against in Stage 3.
        accuracy: AccuracyRequirement::Explicit(RequestedAccuracy {
            scope: AccuracyScope::WholeResult,
            ..RequestedAccuracy::epsilon_delta(0.05, 0.01)
        }),
        response_latency: LatencyRequirement::Unspecified,
    },
    predictability: Predictability::Predictable { known_at: None },
    time_selection: TimeSelection {
        scope: QueryTimeScope::RealTime,
        lookback: Some(DurationMs(60_000)),
        as_of: None,
    },
};
// 2. What the deployment declares about the data: at most 1,000 distinct items in
//    one evaluation of any distinct count over `flows` (the SQL table name is the
//    source key). Stage 1 never reads it; Stage 3 uses it to turn HLL's symbolic δ
//    into a number.
let data = DataWorkload {
    arrival: DataArrival::ContinuouslyIngesting,
    ingestion_rate: declared(Rate(1_000_000.0 / 15.0)),
    max_distinct_per_evaluation: declared(BTreeMap::from([("flows".into(), 1_000)])),
    ..Default::default()
};
// The planner input: the query workload plus the data facts.
let workload = PlanningWorkload {
    query_workload: QueryWorkload {
        language: QueryLanguage::SQL(SqlDialect::DataFusionSQL),
        query_batch: None,
        repeating_queries: Some(vec![panel]),
    },
    data_workload: Some(data),
};
// The `flows(ts, src_ip)` schema the SQL frontend resolves the query against.
let catalog = asap_frontend_sql::SqlCatalog::new().with_table(
    "flows",
    Schema::with_time_index(
        vec![
            Field::plain("ts", DataType::Timestamp, false),
            Field::plain("src_ip", DataType::Utf8, false),
        ],
        0,
        vec![],
    ),
);

// 3. Stage 0 (query-language frontends): `roots` holds one plan root per workload entry,
//    here a distinct count (`aggregate:cardinality`) over a scan of `flows`; Stage 1 reads it.
let roots = lower_sql_roots(&workload, &catalog)?;

// 4. Stages 1-3 in one call. `run` holds every stage's candidates and the selection,
//    which steps 5 and 6 read.
let run = run_planning_stages(&workload, roots, None).expect("Pass 1 enumerates");
//    Stage 1 / Pass 1 offers L1 exact, L2 HLL, L3 KMV and L4 UnivMon for the count.
//    For L2 it calls `size_params(Hll, &intent, 0.05, 0.01)`: the smallest m with
//    m ≥ 1/(2ε²δ) whose tail certifies δ at ε for every bound up to ⌈1/ε⌉ = 20,
//    i.e. `Hll { precision: 15, relative_error: Some(0.05) }`.
//    It then calls `hll::classic_guarantee(15, Some(0.05), "flows")`: error bound
//    0.05 (relative), and δ left as the symbol
//    `classic_hll_failure_probability[p=15,eps=0.05](max_distinct[flows])`,
//    because the distinct count it depends on is a data fact Stage 1 does not read.
//    L3 needs no bound: `Kmv { k: 4323 }` certifies δ = 0.009996 for any count.
//
// 5. Stage 3 (plan selection) binds the symbol: it parses (15, 0.05, "flows"), reads
//    the bound 1,000 from `max_distinct_per_evaluation`, and calls
//    `hll::classic_failure_probability(15, 0.05, 1_000)` = 0.0058 ≤ 0.01, so P2 (HLL)
//    meets the target without enlargement (`precision_for` keeps 15; precision 15
//    certifies bounds up to 1,176, precision 18 up to 9,636).
// `selection`: Stage 3's result (selected id, costs, rejections); Err only when no candidate is valid.
let selection = run.stage3_plan_selection?;

// 6. What the user gets back.
//    `selected`: the cheapest valid plan. The analytical cost model prices the
//    exact plan P1 at 1.7600001 cost/s and HLL (P2) and KMV (P3) at 1.7600003 (the
//    same scan and build, plus an estimate node), so exact is selected; HLL is valid
//    but costlier.
assert_eq!(selection.selected, "P1");
//    `costs`: one entry per valid candidate (P1-P4; UnivMon P4 costs 2.96).
assert_eq!(selection.costs.len(), 4);
//    `rejected`: P2's line is `valid: true, reason: "costlier: 1.760 vs 1.760
//    cost_per_second"`. With a bound of 10,000 instead, P2 is `valid: false`,
//    "failure probability Some(1.0), which misses target ..." (beyond 9,636 even at
//    precision 18); with no bound, "failure probability None".
// `hll`: the HLL candidate's line in the selection report, read to show why it was not chosen.
let hll = selection.rejected.iter().find(|r| r.id == "P2").unwrap();
assert!(hll.valid && hll.reason.starts_with("costlier"));

Before this PR

Example 2, Q1 (ε = 0.02, δ = 0.01)	Before
HLL size	precision 12, error bound 0.01625 (1.04/√m)
HLL valid for a bound of	at most 10 (9 at precision 12, 10 after Stage 3's enlargement); 100 and 10M rejected
KMV size and guarantee	k = 250,002, δ = 0.01 fixed (Chebyshev 99%)
KMV at δ = 10⁻⁴	rejected: 0.01 > 10⁻⁴

After this PR

Example 2, Q1 (ε = 0.02, δ = 0.01)	After
HLL size	precision 17, error bound 0.02
HLL valid for a bound of	up to 1,842 at precision 17, up to 3,791 after enlargement to 18; 10,000 and 10M rejected
KMV size and guarantee	k = 26,656, δ = 0.009998, no distinct bound needed
KMV at δ = 10⁻⁴	valid, with a larger k

So Q1's distinct count over 10M source IPs is certifiable through KMV at any δ. Classic HLL stays limited by its linear-counting range.

stage_pipeline for Example 2 (SQL, no distinct bound declared):

• Selection is unchanged: P1 (all exact) at 6.08 per second.
• The same 12 plans are valid, and their costs are unchanged; the analytical cost model's totals do not depend on k or precision here.
• The changes are KMV's k (250,002 → 26,656, contract kmv_exact_below_k_chernoff_v1) and HLL's precision (12 → 17, relative error 0.02).
• Examples 1, 3a and 3b are byte-identical.
• stage-viewer.md's numbers are unchanged, so it is not edited.

Verification

• Failing first: kmv_sized_for_a_failure_probability_meets_it (KMV sizing ignores the requested δ: guarantee reports 0.01 after sizing for δ = 0.001 #700: δ 0.01 > 0.001) and precision_is_sized_from_epsilon_and_delta (precision 9, not 15).
• New tests:
  • kmv_bound_dominates_the_exact_order_statistic_tails: the bound is at least the exact binomial tails for k ∈ {16, 64, 256}, ε ∈ {0.1, 0.3, 0.6}, and N from k to 20k.
  • estimates_k_minus_one_over_the_kth_smallest_hash: the executor's KMV estimate is the estimator the bound covers.
  • largest_certified_bounds_by_precision: reproduces Certify distinct counts at realistic scale: size HLL from (ε, δ) and give KMV a provable guarantee #694's table, including 9,636 at precision 18 past the old cap.
  • kmv_k_follows_the_failure_probability.
  • The HLL scope negative tests.
  • Example 2: stage3_plan_selection_accepts_kmv_for_any_failure_probability_without_a_bound, plus updated HLL bound tests.
• stage3_plan_selection_enlarges_tumbling_pane_hll: after the rebase onto feat(planner): tumbling-pane window composition as a Pass 2 rule (Example 3) #713, plan selection enlarges a tumbling-pane PaneChain's HLL. A bound of 3,000 is valid only at precision 18, one above Pass 1's 17.
• Rebased onto 3466f32f. cargo nextest run on asap-aware-mapping, asap-physical-operators, asap-integration-tests, asap-types and asap-devtools: 1339 passed, 31 skipped.
• cargo fmt --all -- --check and cargo clippy --workspace --all-targets --all-features --locked -- -D warnings are clean.
• tools/verify_metricsql_external_consumer.sh builds.

Limitations and follow-ups

• KMV's model assumes independent, uniform and distinct 64-bit hashes. Hash collisions and the 2^-64 granularity are not modelled; HLL's contract makes the same kind of hash assumption.
• asap_sketchlib's KMV insert is O(k) and its merge O(k²) (KMV insert and merge are O(k) and O(k²) in asap_sketchlib; unusable at the k the planner sizes #709). The smaller k here helps, but k = 26,656 at ε = 0.02, or about 1.5·10⁵ at ε = 0.01 and δ = 10⁻³, is still slow at runtime. The fix belongs upstream.
• HLL over computed rows still has no δ.
• A shared build takes the smallest recorded ε, like the other shared sizes, which follow sizes rather than targets. A consumer with a larger ε and a smaller δ can therefore be rejected, or enlarged more than it needs.
• When no precision up to 18 certifies ⌈1/ε⌉ (for example ε = 0.01, or ε ≥ 1), hll_precision returns 18 without a certificate. ε-only SQL requests (default δ = 0.01) at ε = 0.01 therefore get precision 18 instead of 14, which is 16× the registers.
• The offline sketch-evidence schema has no relative_error yet (Update the docs after the G1–G11 consolidation groups land #677).
• Docs that describe the old models are left for Update the docs after the G1–G11 consolidation groups land #677: docs/design_docs/concepts/accuracy-models.md (KMV "Chebyshev at 99%"; HLL "maxima from 1 to 4096") and docs/develop_docs/end-to-end-accuracy-guarantees.md.

[zzylol]

Usage:

let params = size_params(SketchAlgorithm::Hll, &intent, 0.05, 0.01); // Hll { precision, relative_error: Some(0.05) }
let guarantee = hll::classic_guarantee(precision, Some(0.05), "a"); // δ left as classic_failure_statistic(precision, 0.05, "a")
let delta = hll::classic_failure_probability(precision, 0.05, max_distinct_of_a); // bound by plan selection

This is confusing, basically, you should show the usage with a e2e workflow. what does even these three lines mean and used for?

[zzylol]

Rewrote the Usage as an end-to-end run, checked by running it at the merge commit. It starts from a SQL panel, SELECT COUNT(DISTINCT src_ip) FROM flows ... with ε = 0.05 and δ = 0.01, plus a declared max_distinct_per_evaluation of 1,000 for flows. It then goes through lower_sql_roots and run_planning_stages. The comments say where each of the three calls happens and what it produces. In Stage 1 / Pass 1, size_params sizes HLL at precision 15 for (0.05, 0.01), and classic_guarantee leaves δ as the symbol classic_hll_failure_probability[p=15,eps=0.05](max_distinct[flows]). In Stage 3, classic_failure_probability(15, 0.05, 1000) binds that symbol to δ = 0.0058 ≤ 0.01, so the HLL plan P2 is valid. The example ends with the result: P1 (exact, 1.7600001 cost/s) is selected, and P2 is valid: true, "costlier" at 1.7600003. With a bound of 10,000 P2 is rejected, and with no bound it is rejected with failure probability None. Every variable is commented. The SketchParams block now lists every variant instead of // ....