fix(asap-physical-operators): seed each CountSketch row independently

[zzylol]

Why

The planner certifies a CountSketch's point-query failure probability with the exact binomial tail P(Bin(depth, p) >= (depth+1)/2) (#702), and Count-Min's with the matching per-row bound. Both assume the rows hash independently. factory::create_planner_accumulator did not build independent rows:

• Plain CountSketch wrapped asap_sketchlib's portable CountSketch, which slices one shared Packed64 hash across rows. At width 30000 only about four rows get independent hash bits, and deep shapes panic in debug builds.
• CmsWithHeap and CountSketchWithHeap wrapped the portable CMSHeap/CSHeap, which hash row r with SEEDLIST[r % 20]. Row r + 20 copies row r, so the planner's CountSketch depths (23 to 149) are not independent. These updaters also rounded weights and dropped nonpositive ones, and their states could not be loaded by the DAG executor, whose validate_state expects the weighted frequency kernel.

What

• The factory builds plain CountSketch, CmsWithHeap and CountSketchWithHeap through one WeightedFrequencyUpdater. It wraps asap_sketchlib's per-row-seeded weighted frequency kernel (controller-OMP #160 upstream), the kernel the DAG's keyed summary build already uses. Plain CountSketch uses capacity 0 (no candidate heap). Rows are pairwise independent at any depth, weights are exact Float64 values, and the states are the DAG's state type.
• A negative Count-Min input now fails validate_single_input instead of being dropped. CountSketch accepts signed weights.
• An even CountSketch depth is rejected by plan validation, by kernel construction, and by the empirical cost evidence check (asap-aware-mapping empirical_cost::valid_params), because the median needs an odd row count. The planner's count_sketch_depth already picks odd depths.
• The portable plain-CountSketch adapter (summary_kernels/count_sketch.rs, CountSketchAccumulator) is removed. The portable heap updaters (CmsHeapAccumulatorUpdater, CountSketchWithHeapAccumulatorUpdater) and accumulators stay public for callers that construct them directly, but the Planner factory no longer uses them.
• KMV approx_memory_bytes now counts k * 8 bytes, since asap_sketchlib reserves all k hash slots at construction.
• asap_sketchlib pin for asap-physical-operators: e6b0314 -> 9415c5b (still with features = ["experimental"] for KMV). The new pin is e6b0314 merged with upstream commit 04ffc39, which adds a public WeightedFrequency::estimate and allows capacity 0. The adapter exposes the estimate as WeightedFrequency::point_estimate.

How

Each label value of the key is one Utf8 identity, as in the DAG's keyed build, and each raw sample's value is the weight. WeightedFrequency::configuration maps a CmsWithHeap/CountSketchWithHeap family to its (algorithm, width, depth, capacity), so the factory and validate_state agree on the shape. Panes merge cell by cell, because every pane seeds row r the same way.

Plain CountSketch only reaches this factory path: the DAG executor has no native plain-CountSketch state, so such a family cannot bind there.

Key code interfaces added / modified

crates/asap-physical-operators/src/summary_kernels/factory.rs

/// Keyed frequency updater over sketchlib's per-row-seeded weighted frequency
/// kernel; the updater the factory builds for these families.
pub struct WeightedFrequencyUpdater {   // new: plain CountSketch (capacity 0), CmsWithHeap, CountSketchWithHeap
    /// The sketch state being built.
    acc: WeightedFrequency,
    /// Count-Min or CountSketch; picks the readout and which weights are valid.
    algorithm: FrequencyAlgorithm,
    /// Columns per row.
    width: usize,
    /// Rows, each hashed with its own seed.
    depth: usize,
    /// Top-k heap capacity; 0 means no heap.
    capacity: usize,
}

impl WeightedFrequencyUpdater {
    /// Fails for a zero dimension, or an even CountSketch `depth`.
    /// Builds an empty updater for the given shape.
    pub fn new(
        algorithm: FrequencyAlgorithm,
        width: usize,
        depth: usize,
        capacity: usize,
    ) -> Result<Self, crate::Error>;
}

/// The factory's updater interface for these families (new impl).
impl AccumulatorUpdater for WeightedFrequencyUpdater {
    /// Rejects a non-finite input, or a negative Count-Min input (bad input).
    fn validate_single_input(&self, value: f64) -> Result<(), String>;   // new: a negative Count-Min input is an error (was dropped)
    /// Not used: the updater is keyed; a debug build asserts if it is called.
    fn update_single(&mut self, value: f64, timestamp_ms: i64);          // new
    /// Adds `value` as the weight of the key's identity tuple.
    fn update_keyed(&mut self, key: &KeyByLabelValues, value: f64, _timestamp_ms: i64);  // new: each label is one Utf8 identity
    /// Returns the built `WeightedFrequency` state and leaves an empty one of the same shape.
    fn take_accumulator(&mut self) -> Box<dyn AggregateCore>;            // new
    /// Returns a copy of the built state; the updater keeps building.
    fn snapshot_accumulator(&self) -> Box<dyn AggregateCore>;            // new
    /// Returns the built state, consuming the updater.
    fn into_accumulator(self: Box<Self>) -> Box<dyn AggregateCore>;      // new
    /// Empties the state, keeping (algorithm, width, depth, capacity).
    fn reset(&mut self);                                                 // new
    /// `true`: items come from the key's labels.
    fn is_keyed(&self) -> bool;                                          // new
    /// The state's approximate footprint, reserved by the executor.
    fn memory_usage_bytes(&self) -> usize;                               // new
}

// removed: pub struct CountSketchAccumulatorUpdater { acc: CountSketchAccumulator, row_num: usize, col_num: usize }
// removed: pub fn CountSketchAccumulatorUpdater::new(row_num: usize, col_num: usize) -> Self
// removed: fn cms_heap_dims(params: &SketchParams) -> (usize, usize, usize)
// kept public, no longer built here: CmsHeapAccumulatorUpdater, CountSketchWithHeapAccumulatorUpdater

/// 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>;   // CountSketch, CmsWithHeap, CountSketchWithHeap => WeightedFrequencyUpdater

crates/asap-physical-operators/src/summary_kernels/weighted_frequency.rs, mod.rs

impl WeightedFrequency {
    /// Point estimate for one identity tuple: the minimum over rows for
    /// Count-Min, the sign-corrected median over rows for CountSketch.
    /// Used by readouts of one key's frequency.
    pub fn point_estimate(&self, values: &[Value]) -> Result<f64, Error>;   // new
}

// removed: pub mod count_sketch; pub use count_sketch::*;
// removed: pub struct CountSketchAccumulator { pub inner: CountSketch }   (one hash shared by all rows)
// removed: pub fn CountSketchAccumulator::new(row_num: usize, col_num: usize) -> Self
// removed: pub fn CountSketchAccumulator::query_key(&self, key: &KeyByLabelValues) -> f64

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

/// Checks, before a plan is accepted, that the executor can build this summary.
pub fn validate_summary_kernel(
    family: &SummaryFamilyType,
    input: &SummaryUpdate,
    grouping: &GroupingStrategy,
) -> Result<(), String>;   // now rejects an even CountSketch depth

impl AggregateCore for KmvAccumulator {
    /// Approximate state size in bytes, used for memory accounting.
    fn approx_memory_bytes(&self) -> usize;   // was k_vals.len() * 8; now counts the k reserved hashes (k * 8)
}

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

/// Whether a sketch's parameters are valid for offline evidence.
fn valid_params(algorithm: &SketchAlgorithm, params: &SketchParams) -> bool;   // CountSketch / CountSketchWithHeap evidence needs an odd depth

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 = "9415c5b161237ad238e101ed9166d4dbaace3031", features = ["experimental"] }  # was rev e6b0314; adds public WeightedFrequency::estimate and capacity 0

Usage, end to end (a top-k over raw samples; the numbers come from a scratch test on precompute_raw_samples.rs's harness, run at cebffbdd):

// 1. What the user writes: the 2 largest series per service by 5-minute sum, within ε = 0.02.
//    `query` and `accuracy` are what the planner plans and checks candidates against.
let query = "topk by (service) (2, sum_over_time(m[5m]))";
let accuracy = AccuracyTarget::Epsilon(0.02);

// 2. Stage 1 composes the logical candidates; Stage 3's weight check (`with_confirmed_weights`,
//    with `m` declared a counter in the data workload) proves the weights non-negative, which a
//    heap needs. `candidates` holds the confirmed candidates. Two answer with a whole-expression
//    heap per service:
//      CmsWithHeap         { width: 136,  depth: 5,  heap_size: 50 }
//      CountSketchWithHeap { width: 7500, depth: 85, heap_size: 50 }   // before this PR the factory
//                                                                     // repeated rows 0..19 as rows 20..84
let candidates = candidates(query, accuracy);
// `candidate` is the CountSketchWithHeap one; step 3 runs it in the DAG, step 4 through the factory.
let candidate = /* the CountSketchWithHeap candidate in `candidates` */;

// 3. In the DAG. `dag` is the candidate's physical DAG at ingestion time; (`source`, `root`) are
//    the raw-sample input and its SummaryAgg node. `family`, `input`, `grouping` are that node's
//    summary declaration, which step 4 hands to the factory. `rows` holds 25 samples of `m`
//    (5 series x 5 steps; {service="a", instance="2"} at 3, 4, 7, 8, 11). `states` holds one
//    (service, WeightedFrequency) state per population, built by the keyed DAG summary.
let dag = compile_at_ingestion_physical_asap_dag(&candidate)?;
let (source, root) = raw_summaries(&dag)[0];
let SummaryAgg { family, input, grouping, .. } = /* node `root`'s payload */;
let states = execute(&dag, source, root, &rows);   // precompute::compile + one ingestion window

// 4. Through the factory, the path callers outside the DAG use. `updater` is this PR's
//    WeightedFrequencyUpdater: create_planner_accumulator maps (CountSketchWithHeap, 7500 x 85,
//    heap 50) to sketchlib's per-row-seeded kernel, the one the DAG build in step 3 uses
//    (an even depth is rejected here). Feed it service="a"'s 10 samples: the item is the label
//    set without `service`, the weight is the sample value.
let mut updater = create_planner_accumulator(&family, &input, &grouping)?;
updater.update_keyed(&KeyByLabelValues::new_with_labels(vec![r#"{"__name__":"m","instance":"2"}"#.into()]), 3.0, 1_000);
/* ... the other 9 samples of service="a" ... */
// `state` holds the finished summary; `heap` is it as the DAG's state type, the type
// validate_state accepts, so the DAG could load it as a summary value.
let state = updater.take_accumulator();
let heap = state.as_any().downcast_ref::<WeightedFrequency>().unwrap();

// 5. What the user gets back: service="a"'s top 2 and one point estimate (the median over 85
//    independently seeded rows). Step 3's DAG state for service="a" returns the same.
heap.rows(2);   // [[Utf8({"__name__":"m","instance":"2"}), 33.0], [Utf8({"__name__":"m","instance":"1"}), 18.0]]
heap.point_estimate(&[Value::Utf8(r#"{"__name__":"m","instance":"2"}"#.into())])?;   // 33.0
// From `states`: service="b" -> instance "" 78.0, instance "1" 48.0; no service -> instance "3" 63.0.
// All equal the exact sums at this scale; the CmsWithHeap candidate returns the same rows.

Before this PR

• A 15 x 30000 plain CountSketch shared hash bits across rows, and a 25-row factory heap sketch repeated rows 0 to 4 as rows 20 to 24. Neither delivered the failure probability the planner reports.
• Factory heap states rounded fractional weights and could not be passed as typed DAG summary values.

After this PR

Every factory-built frequency sketch has independent rows at any odd depth, keeps weights exactly, and loads as a DAG summary value. The planner's δ certification describes what the executor runs.

Verification

New tests:

• deployment::deep_plain_count_sketch_rows_hash_independently: a 15 x 30000 plain CountSketch builds. For 2000 keys each key touches exactly one cell per row, and every pair of rows puts at most 5 keys in the same column (about 0.07 expected under independence). Keys are packed 33 per state with weights 3^p, so each cell's balanced-ternary digits name the keys that hit it.
• deployment::deep_heap_sketch_rows_hash_independently: the same check for 25 x 30000 CmsWithHeap and CountSketchWithHeap over 1000 keys (past the 20-row seed list).
• deployment::heap_sketch_states_load_as_dag_summary_values: factory heap states load into a Batch as typed summary values, and weights 0.25 + 0.5 estimate exactly 0.75.
• deployment::count_min_heap_rejects_negative_inputs
• deployment::deep_plain_count_sketch_estimates_meet_the_certified_bound: on a Zipf-like stream over 3000 keys, every point estimate is within sqrt(3/w) * ||f||_2.
• deployment::plain_count_sketch_merge_equals_union_build: merged cells and estimates equal the union build.
• plan_properties::count_sketch_kernel_requires_an_odd_depth and empirical_cost::tests::count_sketch_evidence_requires_an_odd_depth
• kmv::tests::memory_counts_the_k_reserved_hashes

The heap tests, the odd-depth evidence test and the KMV memory test were run and failed before their fixes.

Commands run on the PR head:

• cargo nextest run -p asap-physical-operators: 205 passed, 3 skipped (already #[ignore]d).
• cargo nextest run -p asap-aware-mapping: 530 passed.
• cargo nextest run -p asap-integration-tests: 245 passed, 33 skipped (already #[ignore]d).
• cargo fmt --all -- --check and cargo clippy --workspace --all-targets --all-features --locked -- -D warnings: clean.

Screenshots and measurements: not applicable.

Limitations and follow-ups

• The backend imports summary_kernels::CountSketchAccumulator (otel.rs), which this PR removes. It will need an update when it bumps its ASAPPlanner pin.
• The new pin 9415c5b sits on the planner pin branch in asap_sketchlib. Upstream commit 04ffc39 (branch feat/weighted-frequency-point-estimate) is in feat(weighted_frequency): public point estimate and heap-less capacity 0 asap_sketchlib#225, still open.
• The portable heap updaters and accumulators remain in the crate. The backend uses them directly (collector wire decoding and a test fixture), so removing them is left for a follow-up.