refactor(physical-operators): one asap_sketchlib pin, library-owned sketch checks, dead portable heap updaters removed

[zzylol]

Why

#706 found that asap_sketchlib's portable frequency sketches broke the planner's guarantees (one shared hash across CountSketch rows, SEEDLIST[r % 20] row seeds in the integer heaps, rounded and dropped weights) and fixed it by building those families on the library's per-row-seeded weighted frequency kernel. The planner still kept sketch-library behavior of its own: copies of the kernel's shape and weight rules in four places, a copy of UnivMon's hash-budget rule in two (one of them missing the 20-row bound), a guessed KMV footprint, clones to work around &mut self reads, unsafe Send/Sync impls, and the dead portable heap updaters "for callers that construct them directly" (there are none in the workspace). The standing rule is that accumulators wrap asap_sketchlib and gaps are fixed upstream.

What

Moved upstream (each with failing-first tests; none merged yet):

Planner code before this PR	Upstream PR
odd CountSketch depth and positive-dimension checks (configuration, capability, empirical_cost)	ProjectASAP/asap_sketchlib#226 WeightedFrequency::check_shape
"finite, nonnegative for Count-Min" sample check (WeightedFrequencyUpdater::validate_single_input)	#226 WeightedFrequency::check_weight
(algorithm, width, depth, capacity) kept beside each state to rebuild it on reset	#226 WeightedFrequency::clear
per-update clone of every retained heap candidate (hot path)	ProjectASAP/asap_sketchlib#227 (2x faster at capacity 100)
KMV clone to read an estimate, clone to merge, guessed k * 8 footprint; O(k) per insert in the library	ProjectASAP/asap_sketchlib#228
UnivMon check_dimensions (executor) and hash_independent (accuracy model)	ProjectASAP/asap_sketchlib#229 UnivMon::check_shape
portable CountSketch sharing one 64-bit hash across rows (#706's item a; no planner caller left)	ProjectASAP/asap_sketchlib#230

Already upstream before this PR: per-row seeding, exact Float64 weights and the odd-depth error in the weighted frequency kernel (ProjectASAP/asap_sketchlib#160, #225).

In this repo:

• One asap_sketchlib pin for the workspace: asap-physical-operators inherits the workspace dependency (with experimental for KMV) instead of its own rev. The rev moves from 831a504 (workspace) and 9415c5b (physical operators) to 3f44cf1 on branch planner-pin/sketch-library-fixes: 9415c5b with refactor: rename PromQL-specific QueryExpr nodes (and Merge/Distinct/WindowFunc/EvalTime) for clarity #226 to SQL: generalize the ClickHouse-builtin stub-registration mechanism from AggregateUDF to ScalarUDF #230 merged in. Once those PRs and controller-OMP #160/SQL: extract dialect builtin-function catalog into independent, generatable data (inspired by polyglot-sql-function-catalogs) #225 merge, the pin must move to upstream main.
• The planner asks the library: WeightedFrequency::check_shape and check_weight, UnivMon::check_shape, KMV::approx_memory_bytes, and KMV::estimate/merge by reference.
• WeightedFrequencyUpdater keeps only its state.
• Deleted: CmsHeapAccumulatorUpdater, CountSketchWithHeapAccumulatorUpdater, TopkWeight, CountMinSketchWithHeapAccumulator (count_min_sketch_with_heap.rs), CountSketchWithHeapAccumulator (count_sketch_with_heap.rs), UnivMon check_dimensions, the accuracy model's column_bits, and DatasketchesKLLAccumulator's unsafe impl Send/Sync (KllSketch is already both).

Behavior changes: the accuracy model now refuses a UnivMon with more than 20 rows, which the executor already refused. A rejected sample's message is now the kernel's ("weighted frequency requires finite weights; CMS additionally requires nonnegative weights"). KMV memory is the library's allocated footprint, which grows from 1024 slots toward k instead of a flat k * 8.

How

Each planner check becomes one library call. asap-physical-operators and asap-aware-mapping both already depend on asap_sketchlib, so no new dependency edges. update_keyed builds the kernel identity directly from the label strings (update_labels) instead of going through Value::Utf8(Arc<str>).

Key code interfaces added / modified

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

/// Keyed frequency updater over sketchlib's weighted frequency kernel; the
/// updater the factory builds for plain CountSketch (capacity 0), CmsWithHeap
/// and CountSketchWithHeap. The type stays.
pub struct WeightedFrequencyUpdater {
    /// The summary being built; it also knows its algorithm and shape.
    acc: WeightedFrequency,               // unchanged
    // removed: algorithm: FrequencyAlgorithm   (read from the state)
    // removed: width: usize, depth: usize, capacity: usize   (kept only to rebuild on reset)
}

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

impl AccumulatorUpdater for WeightedFrequencyUpdater {
    /// Rejects a sample the kernel would refuse: non-finite, or negative for Count-Min (bad input).
    fn validate_single_input(&self, value: f64) -> Result<(), String>;   // was: its own copy of the rule
    /// Adds `value` as the weight of the key's label tuple, one Utf8 identity per label.
    fn update_keyed(&mut self, key: &KeyByLabelValues, value: f64, _timestamp_ms: i64);   // was: via Value::Utf8(Arc<str>)
    /// Empties the state in place for the next window.
    fn reset(&mut self);   // was: rebuilt a new state from the stored shape
}

// removed: pub struct CmsHeapAccumulatorUpdater { acc, row_num, col_num, heap_size, weight }
// removed: pub fn CmsHeapAccumulatorUpdater::new(row_num: usize, col_num: usize, heap_size: usize, weight: TopkWeight) -> Self
// removed: pub struct CountSketchWithHeapAccumulatorUpdater { acc, row_num, col_num, heap_size, weight }
// removed: pub fn CountSketchWithHeapAccumulatorUpdater::new(row_num: usize, col_num: usize, heap_size: usize, weight: TopkWeight) -> Self
// removed: pub enum TopkWeight { Value, Count }

summary_kernels/count_min_sketch_with_heap.rs, summary_kernels/count_sketch_with_heap.rs, summary_kernels/mod.rs (files removed; links at the base)

// removed: pub mod count_min_sketch_with_heap; pub use count_min_sketch_with_heap::*;
// removed: pub struct CountMinSketchWithHeapAccumulator { pub inner: asap_sketchlib::CountMinSketchWithHeap }
//          with new, query_key, insert_value, topk_by_value, get_topk_keys and its AggregateCore impl
// removed: pub mod count_sketch_with_heap; pub use count_sketch_with_heap::*;
// removed: pub struct CountSketchWithHeapAccumulator { pub inner: asap_sketchlib::CountSketchWithHeap }
//          with new, query_key, insert_value, topk_by_value, get_topk_keys and its AggregateCore impl

summary_kernels/weighted_frequency.rs

/// The planner's summary state for CmsWithHeap, CountSketchWithHeap and plain
/// CountSketch: a newtype over asap_sketchlib's weighted frequency kernel that
/// implements AggregateCore. The type itself stays; only these methods change.
pub struct WeightedFrequency {
    /// The library kernel holding the counters and the candidate heap.
    inner: asap_sketchlib::WeightedFrequency,   // unchanged
}

impl WeightedFrequency {
    /// Maps a CmsWithHeap / CountSketchWithHeap family to (algorithm, width,
    /// depth, capacity); fails for another family, a zero heap, or a shape the
    /// kernel's check_shape refuses (bad input). Shared by the factory,
    /// plan validation and the DAG's state check.
    pub(crate) fn configuration(kind: &SketchKind) -> Result<(FrequencyAlgorithm, usize, usize, usize), Error>;   // was: its own odd-depth and zero checks
    /// Adds `weight` to the identity tuple of `labels` (one Utf8 per label). Used by the factory updater.
    pub(crate) fn update_labels(&mut self, labels: &[String], weight: f64) -> Result<(), Error>;   // new
    /// The kernel's weight rule; Err is bad input. Used by the factory updater's validation.
    pub(crate) fn check_weight(&self, weight: f64) -> Result<(), WeightedFrequencyError>;   // new
    /// Empties the state in place, keeping algorithm and shape. Used on reset.
    pub(crate) fn clear(&mut self);   // new
}

summary_kernels/kmv.rs, summary_kernels/univmon.rs, summary_kernels/datasketches_kll.rs, capability.rs

impl AggregateCore for KmvAccumulator {
    /// The state's footprint for memory reservations: the library's KMV::approx_memory_bytes.
    fn approx_memory_bytes(&self) -> usize;   // was size_of::<Self>() + k * 8
    /// Distinct count, read through &KMV.
    fn estimate(&self, query: &SketchStatistic) -> Result<f64, KernelError>;   // was: cloned the state to call KMV::estimate(&mut self)
    /// Merge of two same-k states, merging the other by reference.
    fn merge_with(&self, other: &dyn AggregateCore) -> Result<Box<dyn AggregateCore>, KernelError>;   // was: cloned `other` to call KMV::merge(&mut KMV)
}

// removed: pub(crate) fn univmon::check_dimensions(heap_size: usize, rows: usize, cols: usize, layers: usize) -> Result<(), Error>
//          (UnivMonAccumulator::new / from_sketch and validate_summary_kernel call asap_sketchlib::UnivMon::check_shape)
// removed: unsafe impl Send for DatasketchesKLLAccumulator {}; unsafe impl Sync for DatasketchesKLLAccumulator {} (and their SAFETY comment)

/// 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>;
// unchanged signature; plain CountSketch and UnivMon shapes are now checked by the library's check_shape

crates/asap-aware-mapping/src/accuracy/estimators/univmon.rs, empirical_cost.rs

/// Layers asap_sketchlib builds; Stage 2 sizes distinct-count pyramids with it.
pub(crate) const MAX_LAYERS: u8 = asap_sketchlib::sketch_framework::univmon::MAX_LAYER_SIZE as u8;   // was: the literal 19
/// Whether asap_sketchlib builds this UnivMon shape with independent hashes; the
/// accuracy model certifies no guarantee for a shape it refuses.
pub(crate) fn hash_independent(rows: u32, cols: u32, layers: u8) -> bool;   // was: its own copy of the rule, without the 20-row bound
// removed: fn column_bits(cols: u32) -> u32
/// Whether the CountSketch kernel builds `width x depth`, so cost evidence for it describes an executable shape.
fn count_sketch_builds(width: u32, depth: u32) -> bool;   // new (private), used by valid_params

Cargo.toml, crates/asap-physical-operators/Cargo.toml

# workspace: was rev = "831a5045847010a3a65f5a8e614e0208581d4a05"
asap_sketchlib = { git = "https://github.com/ProjectASAP/asap_sketchlib", rev = "3f44cf1c2d823b844c17796f24408e03863149ee" }
# asap-physical-operators: was its own git dependency at rev = "9415c5b161237ad238e101ed9166d4dbaace3031"
asap_sketchlib = { workspace = true, features = ["experimental"] }

Usage, end to end (a top-k over raw samples; checked by a scratch test on precompute_raw_samples.rs's harness at this PR's head):

// 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; plan selection confirms the weights are counter
//    samples (`m` is declared a counter). `candidates` holds them; two answer with a heap per
//    service, and the executor's family mapping (`configuration`) asks the library's check_shape
//    about each:
//      CmsWithHeap         { width: 136,  depth: 5,  heap_size: 50 }
//      CountSketchWithHeap { width: 7500, depth: 85, heap_size: 50 }   // 85 is odd: check_shape is Ok
let candidates = candidates(query, accuracy);

// Steps 3 to 7 run once per candidate; both give the same answer.
for candidate in &candidates {
    // 3. `dag` is this candidate's physical DAG at ingestion time and `root` its raw-sample
    //    SummaryAgg node; `family`, `input`, `grouping` are that node's summary declaration,
    //    which the factory reads in step 4.
    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 */;

    // 4. The factory maps the family to the kernel. `updater` is a WeightedFrequencyUpdater
    //    holding only its state; it receives the samples in step 5. The executor checks each
    //    sample first with the kernel's own rule:
    //      CmsWithHeap:          validate_single_input(-1.0) and (NaN) are both
    //                            Err("weighted frequency requires finite weights; CMS additionally requires nonnegative weights")
    //      CountSketchWithHeap:  validate_single_input(-1.0) is Ok (signed weights); NaN is the same Err
    //    The same family at depth 86 fails here: Err("invalid DAG: invalid weighted frequency dimensions").
    let mut updater = create_planner_accumulator(&family, &input, &grouping)?;

    // 5. service="a"'s 9 samples. `item(instance)` builds the label set without `service`, the
    //    key update_keyed turns into one Utf8 identity per label (straight from the label String);
    //    the weight is the sample value.
    let item = |instance| KeyByLabelValues::new_with_labels(vec![format!(r#"{{"__name__":"m","instance":"{instance}"}}"#)]);
    for (instance, value) in [("2", 3.0), ("2", 4.0), ("1", 2.0), ("2", 7.0), ("1", 3.0), ("2", 8.0), ("1", 6.0), ("2", 11.0), ("1", 7.0)] {
        updater.update_keyed(&item(instance), value, 1_000);
    }
    // `bytes` is the executor's memory charge for this window's state, used for its memory
    // reservation: 5,844 bytes (CmsWithHeap) and 5,100,404 bytes (CountSketchWithHeap,
    // 7500 x 85 Float64 cells).
    let bytes = updater.memory_usage_bytes();

    // 6. The window closes. `state` is service="a"'s finished summary as a boxed AggregateCore;
    //    `take_accumulator` then resets the updater with `clear`, reusing the 637,500-cell
    //    allocation instead of building a new one.
    let state = updater.take_accumulator();
    // `heap` is the same summary seen as the planner's WeightedFrequency: the counters plus
    // the 50 retained heaviest instances, which step 7 reads the top 2 from.
    let heap = state.as_any().downcast_ref::<WeightedFrequency>().unwrap();

    // 7. What the user gets back, from either candidate: instance "2" at 33.0, instance "1" at
    //    18.0 (the exact sums at this scale). The reset updater's next snapshot has no rows.
    heap.rows(2);   // [[Utf8({"__name__":"m","instance":"2"}), 33.0], [Utf8({"__name__":"m","instance":"1"}), 18.0]]
}

Before this PR

• Two asap_sketchlib revs in one workspace (831a504 and 9415c5b), so two copies of the library compiled.
• The kernel's odd-depth and weight rules, and UnivMon's hash budget, were restated in the planner; the accuracy model's copy accepted a 21-row UnivMon the executor refused.
• KMV reads and merges cloned up to k hashes each; its footprint was guessed.
• About 560 lines of portable heap updaters and accumulators that nothing built.

After this PR

• One rev; the planner keeps only the glue: mapping SketchParams to constructors, turning labels and Values into sketchlib identities, the AggregateCore/AccumulatorUpdater adapters, and the accuracy model's sizing.
• Every shape and weight rule has one owner, the library.
• 97 lines added, 677 removed.

Interface simplifications (author's choice)

• WeightedFrequency (planner) is a newtype over asap_sketchlib::WeightedFrequency only to implement AggregateCore and convert Values. AggregateCore is a local trait, so it could be implemented on the library type directly and the newtype removed; rows/point_estimate/update(&[Value]) would become an extension trait. Trade-off: every downcast_ref::<WeightedFrequency>() (values.rs, operators/summary, tests) changes type; smaller code, one fewer name.
• WeightedFrequency::configuration(kind) -> (FrequencyAlgorithm, usize, usize, usize) could return (FrequencyAlgorithm, (usize, usize, usize)), matching the kernel's shape() tuple that validate_state compares against. Trade-off: purely cosmetic.
• WeightedFrequencyUpdater::new(algorithm, width, depth, capacity) could be From<WeightedFrequency>; its only caller builds the state anyway. Trade-off: one less constructor, slightly less obvious call site.
• hash_independent(rows, cols, layers) passes a dummy heap size of 1 to UnivMon::check_shape; its two callers could call check_shape with their real heap size and drop the helper. Trade-off: two call sites name the library directly.

Efficiency

• update_keyed (per sample): was 1 Vec<Value> + 1 Arc<str> per label in the planner, then 1 Vec + 1 String per label in the adapter; now 1 Vec + 1 String per label. Inside the kernel, normalize still clones the identity again and serializes a key Vec, so about 2 Strings per label and 3 Vecs per sample remain. Cheaper option (follow-up, upstream): an update that takes the identity by value or hashes borrowed &strs, removing the clone.
• Heap rescoring: each update re-hashes every retained key (capacity x depth) and sorts the candidates (capacity log capacity). feat(sql): extract function-name catalog, generalize uniqExact/countIf rewrite #227 removes the per-update clone of every candidate: 66 to 33 us (Count-Min) and 77 to 43 us (CountSketch) per update at capacity 100. Caching each candidate's cells would remove the hashing at capacity x depth x 16 bytes; follow-up.
• reset: zero-fills in place. Before, it allocated a fresh width x depth state per window: 637,500 cells (5.1 MB) for the 7500 x 85 CountSketch above.
• KMV at k = 4096: inserts 2.43 us to 37 ns (Concat drops unique_keys unconditionally — no way for a producer to assert cross-branch disjointness #228, the early return for hashes past the k-th); estimate and merge_with no longer copy up to 4096 hashes.
• The new plan-time checks are O(1).

Verification

• cargo nextest run -p asap-physical-operators -p asap-aware-mapping -p asap-integration-tests: 1075 passed, 30 skipped; at the new head, cargo nextest run -p asap-physical-operators -p asap-aware-mapping: 803 passed, 4 skipped.
• cargo test --doc -p asap-physical-operators -p asap-aware-mapping; cargo fmt --all -- --check; cargo clippy --workspace --all-targets --all-features --locked -- -D warnings; tools/verify_metricsql_external_consumer.sh: all pass on the head.
• New regression test univmon_past_twenty_rows_is_not_hash_independent (accuracy/estimators/univmon.rs): hash_independent(20, 8, 1) is true, hash_independent(21, 8, 1) is false, and a 21-row, 8-column UnivMon certifies no L2 guarantee. Against main's rule (rows · (⌈log2 cols⌉ + 1) ≤ 128 with no row bound, restored temporarily in place of the check_shape call) it fails with assertion failed: !hash_independent(21, 8, 1), since 21 · 4 = 84 hash bits fit the budget; with this PR it passes.
• Changed test: adoption_rejects_terminal_or_invalid_sketches can no longer build a 21-row UnivMon (the library refuses it), so it sets sketch_row = 21 on a built state to check from_sketch still refuses it. memory_counts_the_k_reserved_hashes became memory_covers_the_retained_hashes.
• The upstream PRs carry their own failing-first tests; the pin branch passes cargo test --all-features --locked in asap_sketchlib.

Limitations and follow-ups

• The pin must move to upstream main once feat: own Float64 weighted CMS and CountSketch kernels in sketchlib asap_sketchlib#160, SQL: extract dialect builtin-function catalog into independent, generatable data (inspired by polyglot-sql-function-catalogs) #225, refactor: rename PromQL-specific QueryExpr nodes (and Merge/Distinct/WindowFunc/EvalTime) for clarity #226, feat(sql): extract function-name catalog, generalize uniqExact/countIf rewrite #227, Concat drops unique_keys unconditionally — no way for a producer to assert cross-branch disjointness #228, docs(asap-aware-mapping): correct module doc on where CSE detection runs #229 and SQL: generalize the ClickHouse-builtin stub-registration mechanism from AggregateUDF to ScalarUDF #230 merge. docs(asap-aware-mapping): correct module doc on where CSE detection runs #229 merged onto main after QueryExpr has duplicate relational/scalar-language shapes for the same semantics (Scalar/Literal, BinaryOp/Compare/Arith) #220; the pin carries its identical pre-rebase commit.
• Not moved, and why:
  • The plain Cms family (CmsAccumulatorUpdater, CountMinSketchAccumulator) still wraps the portable CountMinSketch, which drops nonpositive weights silently, wraps row seeds past 20 rows in per-row mode, reports a constant 16 KiB, and reads its bare count by copying the whole matrix. Moving it onto the weighted frequency kernel at capacity 0 would change the DAG's stored plain-Cms state type, so it needs a design decision.
  • Per-row-seeded native CountMin/Count/CMSHeap/CSHeap still repeat row r as row r + 20 past 20 rows. The wire refuses those states; whether constructors should refuse them or derive more seeds is an upstream design question (raised in SQL: generalize the ClickHouse-builtin stub-registration mechanism from AggregateUDF to ScalarUDF #230). The planner no longer builds them.
  • The integer heaps' rounding and dropping of weights is by design for integer counters; the Float64 answer is the weighted frequency kernel, which the planner uses.
  • HydraCms tracks its i32 counter total and DatasketchesKLL/DDSketch updaters charge a fixed 4 KiB; both could become library reports in a follow-up.
  • The accuracy model's sizing constants (MAX_L2_ROWS, HASH_BITS in l2_shape) stay: they choose shapes, which is planner logic.