[zzylol]

Closes #678.

Why

The stage pipeline gave HLL distinct counts a relative error (RSE) but no failure probability, so an HLL candidate could never meet a target with δ. The classic HLL confidence model (ClassicHllConfidence) existed but fed only the legacy candidate search.

What

• Deployment input: DataWorkload.max_distinct_per_evaluation: Evidence<BTreeMap<String, u64>>, per source (PromQL metric or SQL table), styled like stream_norms. It is an upper bound on the distinct items any one distinct-count summary over the source's own rows counts in one complete evaluation, including merged panes. A zero bound is rejected (WorkloadError::ZeroMaxDistinct).
• Stage 1 still sizes HLL precision from ε alone. For an HLL with one instance per group over a source's own rows (its scan, at most behind filters and windows):
  • the guarantee is relative error ≤ 1.04/√m;
  • the failure probability is left symbolic as classic_hll_failure_probability[p=P](max_distinct[source]).
    HLL over computed rows keeps the RSE-only guarantee, with δ unknown.
• Stage 3 binds that statistic through ClassicHllConfidence:
  • a bound outside the contract's range for the precision (above 4096, or outside the linear-counting regime) binds to δ = 1;
  • a missing bound leaves δ unknown, so the target is unmet.
• Deleted legacy-only code: EstimatorContract, AccuracyEvidenceProvider::estimator_contract and EstimatorAccuracy, with their test in replacement.rs. ClassicHllConfidence and its tests stay.

How

Stage 1 does not read the data workload. The bound arrives as a deployment input, and one new branch in Stage 3's workload_statistic resolves the symbolic statistic, the same pattern as stream norms in #687.

Key code interfaces added / modified

crates/types/src/workload.rs

/// The data workload: statistics and declarations about the data being queried.
/// The type stays; it gains one field.
pub struct DataWorkload {
    /// Whether the data is at rest or continuously ingesting.
    pub arrival: DataArrival,   // unchanged
    /// Cadence at which each PromQL source supplies a sample; PromQL needs it. Default unknown.
    #[serde(default)]
    pub data_ingestion_interval: Evidence<DurationMs>,   // unchanged
    /// Total ingested volume, in samples or rows. Default unknown.
    pub ingestion_volume: Evidence<u64>,   // unchanged
    /// Samples or rows per second. Default unknown.
    pub ingestion_rate: Evidence<Rate>,   // unchanged
    /// Number of distinct series or keys; the cost model's group count. Default unknown.
    pub input_cardinality: Evidence<u64>,   // unchanged
    /// Shape of the value distribution (e.g. Zipf). Default unknown.
    pub distribution: Evidence<DataDistribution>,   // unchanged
    /// Stream norms per source, which Count-Min and Count Sketch bounds are normalized by;
    /// a source without a norm leaves that bound unmet. Default empty.
    #[serde(default)]
    pub stream_norms: Evidence<BTreeMap<String, StreamNorms>>,   // unchanged
    /// Per-source bound, enforced by the deployment, on distinct items one distinct-count
    /// summary counts per complete evaluation; binds classic HLL's failure probability.
    /// Default empty (that probability stays unknown).
    #[serde(default)]
    pub max_distinct_per_evaluation: Evidence<BTreeMap<String, u64>>,  // new: per-source deployment bound on distinct items per evaluation
}

/// Why a workload failed validation; every variant is bad input. The type stays; it gains one variant.
pub enum WorkloadError {
    /// `data_ingestion_interval` evidence has expired at planning time (bad input).
    UnavailableDataIngestionInterval,   // unchanged
    /// A PromQL or MetricsQL workload has no data workload (bad input).
    MissingPromqlDataWorkload,   // unchanged
    /// A PromQL or MetricsQL workload has no `data_ingestion_interval` (bad input).
    MissingDataIngestionInterval,   // unchanged
    /// `data_ingestion_interval` is zero (bad input).
    ZeroDataIngestionInterval,   // unchanged
    /// A one-time query with zero invocations (bad input).
    ZeroInvocations,   // unchanged
    /// A fixed repetition interval of zero (bad input).
    ZeroRepetitionInterval,   // unchanged
    /// A repeated-query schedule with no times (bad input).
    EmptySchedule,   // unchanged
    /// A demand-estimate observation window whose start is not before its end (bad input).
    EmptyObservationWindow,   // unchanged
    /// A confidence outside [0, 1] or not finite (bad input).
    InvalidConfidence(f64),   // unchanged
    /// A rate that is negative or not finite (bad input).
    InvalidRate(f64),   // unchanged
    /// A response latency that is negative or not finite (bad input).
    InvalidLatency(f64),   // unchanged
    /// Data at rest declared with a positive ingestion rate (bad input).
    AtRestWithPositiveIngestionRate,   // unchanged
    /// A stream norm that is negative, NaN or infinite (bad input).
    InvalidStreamNorm(f64),   // unchanged
    /// A source's `max_distinct_per_evaluation` is zero (bad input).
    ZeroMaxDistinct,  // new: DataWorkload::validate rejects a zero bound
}

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

/// HLL's relative standard error for `m = 2^precision` buckets.
fn relative_standard_error(precision: u8) -> f64;  // new: 1.04 / sqrt(2^precision)

/// The statistic name classic HLL's failure probability at `precision` over `source`
/// is left as; plan selection binds it.
// new: "classic_hll_failure_probability[p=P](max_distinct[source])"
pub(crate) fn classic_failure_statistic(precision: u8, source: &str) -> String;
/// The precision and source a `classic_failure_statistic` name encodes; `None` if it is not one.
pub(crate) fn parse_classic_failure_statistic(statistic: &str) -> Option<(u8, &str)>;  // new

/// Classic HLL's guarantee for a distinct count over `source`'s own rows; `None` for a
/// precision outside the contract's.
// new: relative error 1.04/sqrt(m), failure probability left symbolic in the source's bound
pub(crate) fn classic_guarantee(precision: u8, source: &str) -> Option<AccuracyGuarantee>;
/// The failure probability a `classic_failure_statistic` names, given the source's bound.
// new: the value Stage 3 binds through ClassicHllConfidence; 1.0 outside the contract's range
pub(crate) fn classic_failure_probability(precision: u8, max_distinct: u64) -> f64;

/// An AccuracyModel wrapper that sized and checked HLL from a per-expression estimator
/// contract, for the legacy candidate search only. The type is removed, with its
/// `impl AccuracyModel for EstimatorAccuracy<'_>`.
pub(crate) struct EstimatorAccuracy<'a> {
    /// The accuracy model it delegated every other check to.
    base: &'a dyn AccuracyModel,   // removed
    /// The expression's trusted contract, if any.
    contract: Option<EstimatorContract>,   // removed
    /// The target's ε.
    epsilon: f64,   // removed
    /// The target's δ.
    delta: f64,   // removed
}

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

/// A trusted per-expression source assertion. The type is removed, replaced by
/// `DataWorkload::max_distinct_per_evaluation`.
pub enum EstimatorContract {
    /// Classic HLL with independent uniform bucket hashing and a distinct bound per evaluation.
    ClassicHll { max_distinct_per_evaluation: u32 },   // removed
}

/// Supplies typed planning-time evidence that propagation rules need. The trait stays; it loses one method.
pub trait AccuracyEvidenceProvider {
    /// The trusted estimator contract for one complete aggregate expression.
    fn estimator_contract(&self, _expression: &OperatorNode) -> Option<EstimatorContract>;   // removed
    /// Enforced bound on distinct (partition, item) identities across one TopK evaluation;
    /// union-bounds score errors of adaptively selected candidates. Default `None`.
    fn topk_max_distinct_items(&self, _expression: &OperatorNode) -> Option<u64>;   // unchanged
    /// Enforced domain of an operand's true values (quantile ratios, products and quotients
    /// of estimates). Default `None` (unknown).
    fn operand_domain(&self, _operand: &OperatorNode) -> Option<OperandDomain>;   // unchanged
    /// The deployment's empirical accuracy for a candidate's whole output; Stage 3 accepts a
    /// candidate without a theoretical guarantee only through it. Default `None`.
    fn output_guarantee(&self, _output: &QueryRoot) -> Option<AccuracyGuarantee>;   // unchanged
    /// Statistics a propagation rule needs for one composition step. Default empty.
    fn propagation_stats(&self, _op: &CompositionOperator, _family: &FieldDataType,
        _query: Option<&SketchStatistic>) -> PropagationStats;   // unchanged
}

/// Public re-exports of the accuracy evidence types.
pub use evidence::{
    AccuracyEvidenceProvider, NoAccuracyEvidence, OperandDomain, PropagationStats,
    WorkloadAccuracyEvidence,
};  // EstimatorContract no longer re-exported

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

/// Classic HLL guarantee for one-instance-per-group distinct counts over a source's own
/// rows; `None` for other summaries or an HLL over computed rows.
// new: Stage 1 guarantee for a per-group HLL Cardinality over a source's own rows
fn classic_hll_guarantee(
    family: &FieldDataType,
    query: &SketchStatistic,
    input: &Rc<OperatorNode>,
) -> Option<AccuracyGuarantee>;
/// The scan's schema when `node` is a source scan, at most behind schema-preserving
/// filters and windows; `None` otherwise.
// new, split out of raw_value_column: the scan's schema behind filters and windows
fn source_scan(node: &Rc<OperatorNode>) -> Option<&Schema>;

/// Value of a statistic Stage 1 left symbolic (row count, stream norm, or classic HLL
/// failure probability), from evidence valid at any planning time; `None` if unknown.
// Stage 3: now also binds classic_hll_failure_probability from max_distinct_per_evaluation
fn workload_statistic(statistic: &str, data: &DataWorkload) -> Option<f64>;

Usage, end to end (numbers from running the stages at this PR's merge commit ac64c279):

// 1. What the user writes: one dashboard panel, the number of distinct source IPs in the
//    last minute, every 10 s, within relative error ε = 0.02 with failure probability δ = 0.01.
//    `q1` is the workload entry; the frontend lowers its query and Stage 3 checks its target.
let q1 = RepeatingEntry {
    query: Query("distinct_over_time(flows_src_ip[1m])".into()),
    demand: RepeatedDemand::FixedInterval(RepetitionInterval(10_000)),
    requirements: QueryRequirements {
        accuracy: AccuracyRequirement::Explicit(AccuracyTarget::EpsilonDelta { epsilon: 0.02, delta: 0.01 }),
        response_latency: LatencyRequirement::Unspecified,
    },
    predictability: Predictability::Predictable { known_at: None },   // a dashboard panel, known ahead
    time_selection: TimeSelection {                                    // real time, over the last minute
        scope: QueryTimeScope::RealTime,
        lookback: Some(DurationMs(60_000)),
        as_of: None,
    },
};

// 2. What the deployment declares (new): at most 9 distinct values of `flows_src_ip` in any one
//    evaluation. `data` is the data workload; only Stage 3 reads this bound, Stage 1 never does.
let data = DataWorkload {
    max_distinct_per_evaluation: declared(BTreeMap::from([("flows_src_ip".into(), 9)])),
    arrival: DataArrival::ContinuouslyIngesting,                     // a live flow stream
    data_ingestion_interval: declared(DurationMs(15_000)),           // PromQL source cadence
    ingestion_volume: Evidence::default(),                           // unknown
    ingestion_rate: declared(Rate(1_000_000.0 / 15.0)),              // rows per second, for the cost model
    input_cardinality: declared(10_000_000),                         // 10M source IPs, for the cost model
    distribution: declared(DataDistribution::Zipf),                  // value skew, for the cost model
    stream_norms: Evidence::default(),                               // none; no Count-Min/Count Sketch here
};

// 3. `workload` is the planner input; `run` holds every stage's output for it
//    (the same pipeline stage_pipeline runs).
let workload = PlanningWorkload {
    query_workload: QueryWorkload { language: QueryLanguage::PromQL, query_batch: None, repeating_queries: Some(vec![q1]) },
    data_workload: Some(data),
};
let run = run_planning_stages(&workload, lower_promql_roots(&workload)?, None)?;

// 4. Stage 1 (stage1_logical_asap_optimization) offers L1 exact, L2 HLL, L3 KMV, L4 UnivMon.
//    L2's HLL is sized from ε alone: precision 12. Its guarantee comes from
//    hll::classic_guarantee(12, "flows_src_ip"): relative error 1.04/√4096 = 0.01625, and a
//    failure probability left symbolic, since Stage 1 does not read the data workload:
//      failure_probability: Unknown { statistic: "classic_hll_failure_probability[p=12](max_distinct[flows_src_ip])" }

// 5. Stage 3 (stage3_plan_selection) binds that statistic in workload_statistic:
//    parse_classic_failure_statistic gives (12, "flows_src_ip"), the data workload gives 9, and
//    classic_failure_probability(12, 9) = 0.00879 ≤ δ = 0.01, so P2 (from L2) is valid.

// 6. What the user gets back in `run.stage3_plan_selection`:
//    selected: "P1" (exact, cost 44.80; the analytical cost model prices exact lowest here)
//    P2 HLL precision 12: valid, cost 1,671.20
//    P3 KMV k = 250,002: valid, cost 800,039.20
//    P4 UnivMon: rejected, "q1: the output has no guarantee, ..."
//    With a bound of 100 instead, P2 is rejected: "failure probability Some(0.7302...), which misses
//    target EpsilonDelta { epsilon: 0.02, delta: 0.01 }"; with no bound: "failure probability None".

Before this PR

In Example 2, Q1's HLL candidate (ε = 0.02, δ = 0.01) was invalid: its failure probability was unknown.

After this PR

Q1 HLL (precision 12), Example 2	Result
no max_distinct_per_evaluation	invalid; failure probability None
bound 9	valid (HLL for Q1, exact for Q2/Q3)
bound 10, 100 or 10,000,000	invalid; δ exceeds the target

Verification

• Local, before merge:
  • cargo nextest run --workspace: 1613 passed, 50 skipped.
  • cargo fmt --all --check and cargo clippy --workspace --all-targets -- -D warnings clean.
• GitHub CI also passed on head 448e7491 (test, format-and-lint).
• New tests, each run and seen failing before the implementation:
  • planner_layering_example2: stage3_plan_selection_accepts_hll_given_a_distinct_bound, stage3_plan_selection_rejects_hll_whose_distinct_bound_is_too_large, and stage1_logical_asap_optimization_leaves_the_hll_failure_probability_symbolic (Stage 1 output is identical with and without a bound).
  • hll.rs: the statistic name round-trips, and binding matches ClassicHllConfidence.
  • workload.rs: a zero bound is rejected.

Limitations and follow-ups

• With ε-only sizing, precision 12 certifies δ ≤ 0.01 for at most 9 distinct items per evaluation, and the contract supports bounds of at most 4096. A realistic bound for Example 2's 10M source IPs therefore still leaves HLL invalid. Sizing precision jointly from ε and δ is not done here.
• HLL over computed (non-source) rows still has no δ.