feat(asap-physical-operators): KMV distinct-count kernel (#697)

[zzylol]

Closes #697.

Why

Pass 1 offers KMV as a distinct-count candidate and Stage 3 prices it, but asap-physical-operators had no KMV kernel, so a Stage 3-selected plan that uses KMV could not execute. #694 plans to make KMV the certified distinct-count summary at realistic scale, so the executor needs it.

What

• New KmvAccumulator summary kernel (summary_kernels/kmv.rs) that wraps asap_sketchlib::KMV (no local copy): build over sample values, merge, and the Cardinality estimate.
• KMV is registered wherever HLL is: summary-kernel validation (k >= 2), native-family validation, sketch-evaluation validation, stored-state validation, and the accumulator factory.
• asap_sketchlib is now built with its experimental feature (KMV is gated behind it) at the existing pin e6b0314.

How

Each sample value is hashed by its bits (-0.0 folded into 0.0, NaN skipped), the same convention as the HLL kernel, through the library's shared CANONICAL_HASH_SEED. KMV::merge keeps the k smallest distinct hashes of both states, so a merged state equals the union build. A merge across different k is rejected. The kernel answers Cardinality and a bare PointCount (both read the distinct count); other statistics return an error.

Key code interfaces added / modified

crates/asap-physical-operators/src/summary_kernels/kmv.rs (new), mod.rs

/// The KMV (k minimum values) distinct-count kernel module.
pub mod kmv;          // new
/// Re-exports `KmvAccumulator` and `KMV_MIN_K` from `summary_kernels`.
pub use kmv::*;       // new

/// The smallest `k` the estimate `(k - 1) / U(k)` is defined for; smaller `k` is rejected at validation.
pub const KMV_MIN_K: u32 = 2;   // smallest k the executor accepts

/// The executor's KMV summary state; merging two equals one build over the union.
pub struct KmvAccumulator {
    /// The library sketch holding the `k` smallest distinct hashes.
    pub inner: KMV,   // asap_sketchlib::KMV
}

impl KmvAccumulator {
    /// An empty KMV keeping the `k` smallest hashes.
    pub fn new(k: u32) -> Self;
    /// Observes one sample value, hashed by its bits.
    pub fn insert_sample(&mut self, value: f64);   // NaN skipped, -0.0 hashed as 0.0
}

/// KMV as a summary state the executor merges and reads (new impl).
impl AggregateCore for KmvAccumulator {
    /// A boxed copy of this state, used when a state is shared or snapshotted.
    fn clone_boxed_core(&self) -> Box<dyn AggregateCore>;   // new
    /// This state as `Any`, for downcasting to `KmvAccumulator` (merge, `validate_state`).
    fn as_any(&self) -> &dyn std::any::Any;                 // new
    /// Mutable `Any` view, for in-place downcasts.
    fn as_any_mut(&mut self) -> &mut dyn std::any::Any;     // new
    /// Merges with another KMV state; errors (bad input) on a non-KMV state or a different `k`.
    fn merge_with(&self, other: &dyn AggregateCore) -> Result<Box<dyn AggregateCore>, KernelError>;  // new: same k only
    /// The distinct count (a bare `PointCount` reads it too); other readouts are errors (bad plan).
    fn estimate(&self, query: &SketchStatistic) -> Result<f64, KernelError>;  // new: Cardinality, or PointCount { value: None, .. }
    /// The struct size plus 8 bytes per kept hash; used for memory reservations.
    fn approx_memory_bytes(&self) -> usize;                 // new
    // is_empty, shared_state: trait defaults (false, None)
}
/// A KMV state is its own updater: it observes unkeyed sample values (new impl).
impl AccumulatorUpdater for KmvAccumulator {
    /// `false`: KMV counts sample values, not keyed items.
    fn is_keyed(&self) -> bool;                                                   // new
    /// Same as `approx_memory_bytes`; the executor reserves it per state.
    fn memory_usage_bytes(&self) -> usize;                                        // new
    /// Observes one sample value through `insert_sample`; the timestamp is ignored.
    fn update_single(&mut self, value: f64, timestamp_ms: i64);                   // new
    /// Same as `update_single`; the key is ignored.
    fn update_keyed(&mut self, key: &KeyByLabelValues, value: f64, timestamp_ms: i64);   // new
    /// Returns the built state and resets this one to an empty KMV with the same `k`.
    fn take_accumulator(&mut self) -> Box<dyn AggregateCore>;                     // new
    /// Returns a copy of the built state; this one keeps building.
    fn snapshot_accumulator(&self) -> Box<dyn AggregateCore>;                     // new
    /// Returns the updater itself as the state.
    fn into_accumulator(self: Box<Self>) -> Box<dyn AggregateCore>;               // new
    /// Empties the state, keeping `k`.
    fn reset(&mut self);                                                          // new
    // validate_single_input: trait default (finite values only)
}

crates/asap-physical-operators/src/capability.rs, summary_kernels/factory.rs, values.rs

/// Checks, before a plan is accepted, that the executor can build this
/// summary; the same contract `create_planner_accumulator` uses.
pub fn validate_summary_kernel(
    family: &SummaryFamilyType,
    input: &SummaryUpdate,
    grouping: &GroupingStrategy,
) -> Result<(), String>;   // accepts (Kmv, SketchParams::Kmv { k }) with k >= KMV_MIN_K
/// Errors unless the executor has a native kernel for `family`.
pub fn validate_native_family(family: &SummaryFamilyType) -> Result<(), Error>;   // Kmv is now native
/// Errors unless the executor can read `query` from `family` directly.
pub fn validate_sketch_evaluation(
    family: &SummaryFamilyType,
    query: &SketchStatistic,
) -> Result<(), Error>;   // (Kmv, Cardinality) answers; other Kmv readouts only as a bare count

/// Builds the kernel a Planner `SummaryAgg` declares; unsupported payloads are errors.
pub fn create_planner_accumulator(
    family: &SummaryFamilyType,
    input: &planner_types::post_asap::SummaryUpdate,
    grouping: &planner_types::post_asap::GroupingStrategy,
) -> Result<Box<dyn AccumulatorUpdater>, String>;   // (Kmv, SketchParams::Kmv { k }) => KmvAccumulator::new(k)

/// Checks a stored state matches its declared family and parameters before it is used.
fn validate_state(family: &SummaryFamilyType, state: &dyn AggregateCore) -> Result<(), Error>;   // checks a KMV state's k

crates/asap-physical-operators/Cargo.toml

# The pinned sketch library for the executor's kernels.
asap_sketchlib = { git = "https://github.com/ProjectASAP/asap_sketchlib", rev = "e6b03142628a3109b6482ebe8512bff7e75f5b35", features = ["experimental"] }  # was: no features; KMV needs "experimental"

Usage, end to end (Example 2's Q1 through its PromQL stand-in; the numbers come from stage3_plan_selection_selected_kmv_plan_executes_end_to_end, run at 2357ecec):

// 1. What the user writes. `workload` is Example 2: Q1 counts distinct source IPs over the
//    last minute every 10 s, within ε = 0.02 at δ = 0.01 (Q2 and Q3, entropy and L2, ride
//    along). Every stage below reads it.
let workload = standin_workload();   // Q1: "distinct_over_time(flows_src_ip[1m])",
                                     // requirements.accuracy: Explicit(epsilon_delta(0.02, 0.01))
// `entries` holds the three normalized query entries; targets and recurrence are read from them.
let entries: Vec<_> = workload.query_workload.entries().collect();
// `data` holds the source facts (10M distinct source IPs, 1M samples per 15 s); Stage 3
// prices row counts from it.
let data = workload.data_workload.as_ref().unwrap();

// 2. Stages 0–2. The frontend lowers Q1 to a distinct count. Pass 1 offers exact, HLL, UnivMon
//    and KMV for it and sizes KMV from ε (k = 250,002 here). Stage 2 makes one physical
//    candidate per logical one. `run` holds all of them; step 3 picks from it.
let run = run_stages(&workload, lower_promql(&workload));

// 3. Stage 3. `kmv` holds the 3 physical candidates that answer Q1 with KMV and Q2, Q3 exactly.
//    Over all candidates Stage 3 selects the all-exact P1 (8.64 cost_per_second), so this
//    restriction is what puts a KMV plan in front of the executor.
let kmv: Vec<PhysicalCandidate> = /* candidates whose Q1 option is "Kmv" and Q2, Q3 "exact" */;
// `targets` holds one AccuracyTarget per query, from each entry's requirements.accuracy:
// [ε 0.02 δ 0.01, ε 0.05 δ 0.01, ε 0.01 δ 0.01]. Stage 3 checks each guarantee against it.
let targets: Vec<_> = entries.iter().map(|e| Some(e.requirements.accuracy.target())).collect();
// `recurrence` holds each query's 10 s repetition; Stage 3 turns cost per run into cost per second.
let recurrence: Vec<_> = entries.iter().map(|e| e.recurrence.clone()).collect();
// `selection` holds the chosen candidate, each valid candidate's cost and each rejection.
let selection = select_plan(&kmv, &targets, &recurrence, data, PlanningModels::builtin())?;
// -> selected "P25" at 9.04 cost_per_second; P26 and P27 tie at 9.04 and lose on enumeration order.
// `p` is P25's physical DAG. Its node 2 is Q1's SummaryAgg { Kmv, SketchParams::Kmv { k: 250002 } }.
let p = run.physical(&selection.selected);

// 4. The executor compiles Q1's sub-DAG. `q1` is Q1's root node id; `inputs` maps the
//    time_range node to a bounded input contract for flows_src_ip. Compiling node 2 calls this
//    PR's validate_summary_kernel (Kmv with k >= KMV_MIN_K = 2 is accepted) and
//    create_planner_accumulator, which returns KmvAccumulator::new(250_002).
//    `plan` is the compiled operator graph that step 5 runs.
let q1 = p.query_roots[0] as u64;
let plan = compile(p.dag(), inputs, &[q1])?;

// 5. Run it. `batches` holds 1,000 flow records over 50 s with 300 distinct source IPs. Each
//    sample goes through KmvAccumulator::insert_sample (hashed by its bits). Q1's estimate node
//    reads estimate(&SketchStatistic::Cardinality). `result` holds Q1's output: what the user gets.
let result = execute(&plan, batches, Scope::Query { evaluation_time_ms: 55_000, revision: 1 });
// -> [300.0]: fewer than k distinct values, so the KMV estimate is exact.

Before this PR

A plan with a KMV summary failed when it was bound to a kernel.

After this PR

A Stage 3-selected KMV plan for Example 2 Q1 executes end to end and returns the exact distinct count when the input has fewer than k distinct values. Raw-sample precompute checks KMV against its kernel.

Verification

• cargo nextest run -p asap-physical-operators: 200 passed, 3 skipped. New kernel tests: fewer-than-k distinct items are counted exactly; many distinct items are estimated within the KMV error; merge equals the union build; different-k merges and quantile queries are rejected; -0.0 equals 0.0 and NaN is not an item.
• cargo nextest run -p asap-integration-tests: 245 passed, 33 skipped (already #[ignore]d on main). New: stage3_plan_selection_selected_kmv_plan_executes_end_to_end.
• cargo clippy --workspace --all-targets --all-features --locked -- -D warnings and cargo fmt --all -- --check: clean.
• Screenshots and measurements: not applicable.

Limitations and follow-ups

• fix(asap-physical-operators): seed each CountSketch row independently #706 (merged after this PR) bumps the asap_sketchlib pin in crates/asap-physical-operators/Cargo.toml to 9415c5b and keeps features = ["experimental"].

• The end-to-end test covers the exact regime (fewer than k distinct values). The certified-bound regime at realistic scale is Certify distinct counts at realistic scale: size HLL from (ε, δ) and give KMV a provable guarantee #694.