From c6a9f7829f6215b0187673bf76c237a872618855 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Thu, 15 Oct 2026 04:45:10 +0000 Subject: [PATCH 1/5] build(deps): pin asap_sketchlib to the weighted frequency point estimate Move the executor's pin from e6b0314 to 9415c5b, which merges asap_sketchlib's public WeightedFrequency::estimate and heap-less capacity 0 (04ffc39) into the previous pin. The adapter exposes the estimate as point_estimate. --- Cargo.lock | 4 ++-- crates/asap-physical-operators/Cargo.toml | 2 +- .../src/summary_kernels/weighted_frequency.rs | 10 ++++++++++ 3 files changed, 13 insertions(+), 3 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index bce4c50f2..adfa58ef1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -407,7 +407,7 @@ dependencies = [ "asap-aware-mapping", "asap-frontend-promql", "asap-types", - "asap_sketchlib 0.3.0 (git+https://github.com/ProjectASAP/asap_sketchlib?rev=e6b03142628a3109b6482ebe8512bff7e75f5b35)", + "asap_sketchlib 0.3.0 (git+https://github.com/ProjectASAP/asap_sketchlib?rev=9415c5b161237ad238e101ed9166d4dbaace3031)", "chrono", "futures", "regex", @@ -464,7 +464,7 @@ dependencies = [ [[package]] name = "asap_sketchlib" version = "0.3.0" -source = "git+https://github.com/ProjectASAP/asap_sketchlib?rev=e6b03142628a3109b6482ebe8512bff7e75f5b35#e6b03142628a3109b6482ebe8512bff7e75f5b35" +source = "git+https://github.com/ProjectASAP/asap_sketchlib?rev=9415c5b161237ad238e101ed9166d4dbaace3031#9415c5b161237ad238e101ed9166d4dbaace3031" dependencies = [ "bincode", "bytes", diff --git a/crates/asap-physical-operators/Cargo.toml b/crates/asap-physical-operators/Cargo.toml index bdc938484..77d01f9c1 100644 --- a/crates/asap-physical-operators/Cargo.toml +++ b/crates/asap-physical-operators/Cargo.toml @@ -7,7 +7,7 @@ edition = "2021" chrono = { version = "0.4.39", default-features = false, features = ["std"] } futures = "0.3" planner-types = { package = "asap-types", path = "../types" } -asap_sketchlib = { git = "https://github.com/ProjectASAP/asap_sketchlib", rev = "e6b03142628a3109b6482ebe8512bff7e75f5b35", features = ["experimental"] } +asap_sketchlib = { git = "https://github.com/ProjectASAP/asap_sketchlib", rev = "9415c5b161237ad238e101ed9166d4dbaace3031", features = ["experimental"] } serde = { version = "1", features = ["derive", "rc"] } serde_json = "1" tracing = "0.1" diff --git a/crates/asap-physical-operators/src/summary_kernels/weighted_frequency.rs b/crates/asap-physical-operators/src/summary_kernels/weighted_frequency.rs index 7c1122305..39b7be953 100644 --- a/crates/asap-physical-operators/src/summary_kernels/weighted_frequency.rs +++ b/crates/asap-physical-operators/src/summary_kernels/weighted_frequency.rs @@ -1,4 +1,8 @@ //! ASAP type and trait adapter for sketchlib's Float64 weighted frequency kernel. +//! +//! The kernel backs Count-Min and CountSketch, with a top-k heap or, at +//! capacity 0, without one. Each row hashes with its own seed, so a shape of +//! any depth keeps the row independence the planner's guarantees assume. use crate::AggregateCore; use crate::{values::Value, Error}; pub use asap_sketchlib::FrequencyAlgorithm; @@ -97,6 +101,12 @@ impl WeightedFrequency { let values = values.iter().map(identity).collect::, _>>()?; self.inner.update(&values, weight).map_err(adapt_error) } + /// Point estimate for one identity tuple: the minimum over rows for + /// Count-Min, the sign-corrected median over rows for CountSketch. + pub fn point_estimate(&self, values: &[Value]) -> Result { + let values = values.iter().map(identity).collect::, _>>()?; + self.inner.estimate(&values).map_err(adapt_error) + } pub fn rows(&self, n: usize) -> Vec> { self.inner .topk(n) From c272be3bf6aeb2375d5c2acfc437de5f5eaccc26 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Thu, 15 Oct 2026 04:45:10 +0000 Subject: [PATCH 2/5] fix(asap-physical-operators): seed each CountSketch row independently The plain CountSketch kernel wrapped asap_sketchlib's portable CountSketch, which slices one shared hash across rows: at width 30000 only about four rows are independent, and deeper shapes panic in debug builds. The planner's binomial-tail failure probability assumes independent rows at any depth. The plain CountSketch now builds sketchlib's per-row-seeded weighted frequency kernel in CountSketch mode with no candidate heap, the same kernel the keyed CountSketch-with-heap build uses. Each label value of the key is one Utf8 identity. Plan validation and kernel construction reject an even depth, since the median needs an odd row count. The portable-CountSketch adapter is removed. Tests: a 15 x 30000 shape builds; its rows put keys in pairwise independent columns; every point estimate on a skewed stream is within the certified sqrt(3/w) * L2 bound; merging two panes equals the union build. --- .../asap-physical-operators/src/capability.rs | 7 +- .../src/summary_kernels/count_sketch.rs | 99 ----------- .../src/summary_kernels/factory.rs | 58 +++--- .../src/summary_kernels/mod.rs | 2 - .../tests/deployment.rs | 168 ++++++++++++++++++ .../tests/plan_properties.rs | 37 ++++ 6 files changed, 247 insertions(+), 124 deletions(-) delete mode 100644 crates/asap-physical-operators/src/summary_kernels/count_sketch.rs diff --git a/crates/asap-physical-operators/src/capability.rs b/crates/asap-physical-operators/src/capability.rs index 8865396a0..c8441313e 100644 --- a/crates/asap-physical-operators/src/capability.rs +++ b/crates/asap-physical-operators/src/capability.rs @@ -61,9 +61,10 @@ pub fn validate_summary_kernel( } (A::Hll, P::Hll { precision }) if (4..=18).contains(precision) => false, (A::Kmv, P::Kmv { k }) if *k >= crate::summary_kernels::kmv::KMV_MIN_K => false, - (A::Cms, P::Cms { width, depth }) - | (A::CountSketch, P::CountSketch { width, depth }) - if valid_matrix(*width, *depth) => + (A::Cms, P::Cms { width, depth }) if valid_matrix(*width, *depth) => true, + // The median needs an odd row count. + (A::CountSketch, P::CountSketch { width, depth }) + if valid_matrix(*width, *depth) && depth % 2 == 1 => { true } diff --git a/crates/asap-physical-operators/src/summary_kernels/count_sketch.rs b/crates/asap-physical-operators/src/summary_kernels/count_sketch.rs deleted file mode 100644 index c3aaa40c9..000000000 --- a/crates/asap-physical-operators/src/summary_kernels/count_sketch.rs +++ /dev/null @@ -1,99 +0,0 @@ -//! CountSketch accumulator backed by `asap_sketchlib::CountSketch`. -//! -//! Per-key queries delegate to sketchlib's median-of-signed-rows estimator. -//! Top-k requires the separate heap-bearing accumulator. - -use crate::{AggregateCore, KeyByLabelValues}; -use asap_sketchlib::CountSketch; - -/// Count Sketch accumulator — inner matrix of signed counts. -#[derive(Debug, Clone)] -pub struct CountSketchAccumulator { - pub inner: CountSketch, -} - -impl CountSketchAccumulator { - pub fn new(row_num: usize, col_num: usize) -> Self { - Self { - inner: CountSketch::new(row_num, col_num), - } - } - - /// Median-of-signed-rows point estimate for `key`, via - /// `asap_sketchlib::CountSketch::estimate`. - pub fn query_key(&self, key: &KeyByLabelValues) -> f64 { - self.inner.estimate(&key.to_semicolon_str()) - } -} - -impl AggregateCore for CountSketchAccumulator { - fn clone_boxed_core(&self) -> Box { - Box::new(self.clone()) - } - - fn as_any(&self) -> &dyn std::any::Any { - self - } - fn as_any_mut(&mut self) -> &mut dyn std::any::Any { - self - } - - fn merge_with( - &self, - other: &dyn AggregateCore, - ) -> Result, Box> { - let other_cs = other - .as_any() - .downcast_ref::() - .ok_or("Failed to downcast to CountSketchAccumulator")?; - - let merged_inner = CountSketch::merge_refs(&[&self.inner, &other_cs.inner])?; - Ok(Box::new(Self { - inner: merged_inner, - })) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_query_key_uses_real_sketchlib_estimator() { - // `query_key` must match sketchlib's estimator and hash specification. - let mut cs = CountSketchAccumulator::new(4, 1000); - let key = KeyByLabelValues::new_with_labels(vec!["web".to_string()]); - cs.inner.update(&key.to_semicolon_str(), 10.0); - assert_eq!( - cs.query_key(&key), - cs.inner.estimate(&key.to_semicolon_str()) - ); - } - - #[test] - fn test_aggregate_core_merge_matches_matrix_add() { - let a = CountSketchAccumulator { - inner: CountSketch::from_legacy_matrix(vec![vec![1.0, -2.0], vec![3.0, -4.0]], 2, 2), - }; - let b = CountSketchAccumulator { - inner: CountSketch::from_legacy_matrix(vec![vec![-1.0, 2.0], vec![-3.0, 4.0]], 2, 2), - }; - let merged_box = a.merge_with(&b).expect("merge ok"); - let merged = merged_box - .as_any() - .downcast_ref::() - .expect("downcast ok"); - let m = merged.inner.sketch(); - assert_eq!(m[0], vec![0.0, 0.0]); - assert_eq!(m[1], vec![0.0, 0.0]); - } - - #[test] - fn test_aggregate_core_merge_wrong_type_rejects() { - use crate::summary_kernels::count_min_sketch::CountMinSketchAccumulator; - let cs = CountSketchAccumulator::new(2, 3); - let cms = CountMinSketchAccumulator::new(2, 3); - let result = cs.merge_with(&cms); - assert!(result.is_err()); - } -} diff --git a/crates/asap-physical-operators/src/summary_kernels/factory.rs b/crates/asap-physical-operators/src/summary_kernels/factory.rs index 4e216bdd4..3b92fbe73 100644 --- a/crates/asap-physical-operators/src/summary_kernels/factory.rs +++ b/crates/asap-physical-operators/src/summary_kernels/factory.rs @@ -1,9 +1,9 @@ use crate::summary_kernels::hll_sketch::HllSketchAccumulator; use crate::summary_kernels::univmon::UnivMonAccumulator; +use crate::summary_kernels::weighted_frequency::{FrequencyAlgorithm, WeightedFrequency}; use crate::summary_kernels::{ - CountMinSketchAccumulator, CountMinSketchWithHeapAccumulator, CountSketchAccumulator, - CountSketchWithHeapAccumulator, DDSketchAccumulator, DatasketchesKLLAccumulator, HydraCms, - HydraKllSketchAccumulator, + CountMinSketchAccumulator, CountMinSketchWithHeapAccumulator, CountSketchWithHeapAccumulator, + DDSketchAccumulator, DatasketchesKLLAccumulator, HydraCms, HydraKllSketchAccumulator, }; use crate::{AggregateCore, KeyByLabelValues}; use planner_types::post_asap::{FieldDataType as SummaryFamilyType, SketchAlgorithm, SketchParams}; @@ -324,29 +324,32 @@ impl AccumulatorUpdater for CmsHeapAccumulatorUpdater { } // --------------------------------------------------------------------------- -// CountSketchAccumulatorUpdater (real median-of-signed-rows CountSketch) +// CountSketchAccumulatorUpdater (median-of-signed-rows CountSketch, no heap) // --------------------------------------------------------------------------- -/// Keyed point-frequency updater backed by a real `asap_sketchlib::CountSketch` +/// Keyed point-frequency updater backed by sketchlib's per-row-seeded +/// weighted frequency kernel in CountSketch mode with no candidate heap /// (signed rows, median-of-rows estimator) — distinct math from -/// `CmsAccumulatorUpdater`'s CMS (min-of-rows). +/// `CmsAccumulatorUpdater`'s CMS (min-of-rows). Each label value of the key +/// is one `Utf8` identity, as in the keyed heap build. /// /// As with bare CMS, each raw Prometheus sample contributes its `value`. /// Unit event counting must be selected explicitly by a future typed plan /// contract rather than being implied by `SketchAlgorithm::CountSketch`. pub struct CountSketchAccumulatorUpdater { - acc: CountSketchAccumulator, + acc: WeightedFrequency, row_num: usize, col_num: usize, } impl CountSketchAccumulatorUpdater { - pub fn new(row_num: usize, col_num: usize) -> Self { - Self { - acc: CountSketchAccumulator::new(row_num, col_num), + /// Fails for a zero or even `row_num`: the median needs an odd row count. + pub fn new(row_num: usize, col_num: usize) -> Result { + Ok(Self { + acc: WeightedFrequency::new(FrequencyAlgorithm::CountSketch, col_num, row_num, 0)?, row_num, col_num, - } + }) } } @@ -359,13 +362,28 @@ impl AccumulatorUpdater for CountSketchAccumulatorUpdater { } fn update_keyed(&mut self, key: &KeyByLabelValues, value: f64, _timestamp_ms: i64) { - self.acc.inner.update(&key.to_semicolon_str(), value); + let identity = key + .labels + .iter() + .map(|label| crate::values::Value::Utf8(label.as_str().into())) + .collect::>(); + // Inputs pass `validate_single_input` (finite) first; only a sum past + // `f64::MAX` can fail here. + self.acc + .update(&identity, value) + .expect("CountSketch counter overflow"); } impl_clone_accumulator_methods!(acc); fn reset(&mut self) { - self.acc = CountSketchAccumulator::new(self.row_num, self.col_num); + self.acc = WeightedFrequency::new( + FrequencyAlgorithm::CountSketch, + self.col_num, + self.row_num, + 0, + ) + .expect("dimensions validated at construction"); } fn is_keyed(&self) -> bool { @@ -373,8 +391,7 @@ impl AccumulatorUpdater for CountSketchAccumulatorUpdater { } fn memory_usage_bytes(&self) -> usize { - std::mem::size_of::() - + self.row_num * self.col_num * std::mem::size_of::() + self.acc.approx_memory_bytes() } } @@ -582,7 +599,7 @@ pub fn create_planner_accumulator( } (SketchAlgorithm::CountSketch, params @ SketchParams::CountSketch { .. }) => { let (r, c) = cms_dims(params); - Box::new(CountSketchAccumulatorUpdater::new(r, c)) + Box::new(CountSketchAccumulatorUpdater::new(r, c).map_err(|e| e.to_string())?) } (SketchAlgorithm::CmsWithHeap, params @ SketchParams::CmsWithHeap { .. }) => { let (r, c, h) = cms_heap_dims(params); @@ -790,11 +807,12 @@ mod planner_parameter_regression { (s.inner.rows(), s.inner.cols()) } SketchAlgorithm::CountSketch => { - let s = state + let (width, depth, _) = state .as_any() - .downcast_ref::() - .unwrap(); - (s.inner.rows, s.inner.cols) + .downcast_ref::() + .unwrap() + .shape(); + (depth, width) } SketchAlgorithm::CmsWithHeap => { let s = state diff --git a/crates/asap-physical-operators/src/summary_kernels/mod.rs b/crates/asap-physical-operators/src/summary_kernels/mod.rs index b30e87fcc..5efef8913 100644 --- a/crates/asap-physical-operators/src/summary_kernels/mod.rs +++ b/crates/asap-physical-operators/src/summary_kernels/mod.rs @@ -1,7 +1,6 @@ //! In-memory summary state: thin adapters over `asap_sketchlib` and exact Planner state. pub mod count_min_sketch; pub mod count_min_sketch_with_heap; -pub mod count_sketch; pub mod count_sketch_with_heap; pub mod datasketches_kll; pub mod dd_sketch; @@ -15,7 +14,6 @@ pub mod univmon; pub use count_min_sketch::*; pub use count_min_sketch_with_heap::*; -pub use count_sketch::*; pub use count_sketch_with_heap::*; pub use datasketches_kll::*; pub use dd_sketch::*; diff --git a/crates/asap-physical-operators/tests/deployment.rs b/crates/asap-physical-operators/tests/deployment.rs index 42c85ca5b..e87b55749 100644 --- a/crates/asap-physical-operators/tests/deployment.rs +++ b/crates/asap-physical-operators/tests/deployment.rs @@ -95,3 +95,171 @@ fn native_count_sketch_dimensions_are_not_packed_wire_dimensions() { .unwrap(); assert_eq!(state.query_key(&key), 7.0); } + +fn plain_count_sketch(width: u32, depth: u32) -> FieldDataType { + FieldDataType::Sketch( + SketchKind::new( + SketchAlgorithm::CountSketch, + SketchParams::CountSketch { width, depth }, + ), + Default::default(), + ) +} + +fn host(i: usize) -> asap_physical_operators::KeyByLabelValues { + asap_physical_operators::KeyByLabelValues::new_with_labels(vec![format!("host-{i}")]) +} + +/// Build a plain CountSketch over unit events, one per key index. +fn build_plain_count_sketch( + family: &FieldDataType, + events: impl IntoIterator, +) -> Box { + build_weighted_plain_count_sketch(family, events.into_iter().map(|i| (i, 1.0))) +} + +/// Build a plain CountSketch over `(key index, weight)` updates. +fn build_weighted_plain_count_sketch( + family: &FieldDataType, + updates: impl IntoIterator, +) -> Box { + use asap_physical_operators::planner::post_asap::SummaryInputExpr; + let mut update = SummaryUpdate::column(ColumnRef::SampleValue); + update.item = Some(SummaryInputExpr::Column(ColumnRef::Named("host".into()))); + let mut operator = create_planner_accumulator(family, &update, &Default::default()).unwrap(); + for (at, (i, weight)) in updates.into_iter().enumerate() { + operator.validate_single_input(weight).unwrap(); + operator.update_keyed(&host(i), weight, at as i64); + } + operator.into_accumulator() +} + +fn plain_estimate(state: &dyn AggregateCore, i: usize) -> f64 { + use asap_physical_operators::{ + summary_kernels::weighted_frequency::WeightedFrequency, values::Value, + }; + state + .as_any() + .downcast_ref::() + .expect("plain CountSketch state is the per-row-seeded weighted frequency kernel") + .point_estimate(&[Value::Utf8(format!("host-{i}").into())]) + .unwrap() +} + +/// Zipf-like unit event stream: key `i` occurs `1000 / (i + 1) + 1` times. +fn zipf_events(keys: usize) -> Vec { + (0..keys) + .flat_map(|i| std::iter::repeat_n(i, 1000 / (i + 1) + 1)) + .collect() +} + +// A deep plain CountSketch, sized from a small δ, builds and keeps every row +// independent: each row hashes with its own seed, so no row geometry limit +// applies and two rows rarely put a key in the same column. +#[test] +fn deep_plain_count_sketch_rows_hash_independently() { + let (width, depth) = (30_000usize, 15usize); + let family = plain_count_sketch(width as u32, depth as u32); + let keys = 2_000; + // Key `base + p` of a batch carries weight 3^p, so a cell holds a signed + // sum whose balanced-ternary digits name the keys that hit it. + const BATCH: usize = 33; + let mut columns = vec![vec![None; depth]; keys]; + for base in (0..keys).step_by(BATCH) { + let batch = base..keys.min(base + BATCH); + let state = build_weighted_plain_count_sketch( + &family, + batch.clone().map(|i| (i, 3f64.powi((i - base) as i32))), + ); + let state = state + .as_any() + .downcast_ref::() + .unwrap(); + let cells = serde_json::to_value(state).unwrap()["cells"].clone(); + for (index, cell) in cells.as_array().unwrap().iter().enumerate() { + let (row, column) = (index / width, index % width); + let mut rest = cell.as_f64().unwrap() as i64; + let mut position = 0; + while rest != 0 { + let digit = rest.rem_euclid(3); + if digit != 0 { + let slot = &mut columns[base + position][row]; + assert!(slot.is_none(), "one cell per row"); + *slot = Some(column); + rest -= if digit == 1 { 1 } else { -1 }; + } + rest /= 3; + position += 1; + } + } + } + let columns = columns + .into_iter() + .map(|c| { + c.into_iter() + .collect::>>() + .expect("one cell per row") + }) + .collect::>(); + for a in 0..depth { + for b in a + 1..depth { + let shared = columns.iter().filter(|c| c[a] == c[b]).count(); + // Independent rows share a column for 2000 / 30000 < 1 key per pair. + assert!(shared <= 5, "rows {a} and {b} share {shared} columns"); + } + } +} + +// The planner certifies `|estimate − truth| ≤ √(3/w) ‖f‖₂` for every point +// query of a plain CountSketch; a deep shape meets it on a skewed stream. +#[test] +fn deep_plain_count_sketch_estimates_meet_the_certified_bound() { + let width = 30_000u32; + let keys = 3_000; + let events = zipf_events(keys); + let state = build_plain_count_sketch(&plain_count_sketch(width, 15), events.iter().copied()); + let mut truth = vec![0.0f64; keys]; + for &i in &events { + truth[i] += 1.0; + } + let l2 = truth.iter().map(|f| f * f).sum::().sqrt(); + let bound = (3.0 / f64::from(width)).sqrt() * l2; + for (i, truth) in truth.iter().enumerate() { + let error = (plain_estimate(&*state, i) - truth).abs(); + assert!(error <= bound, "key {i}: error {error} > bound {bound}"); + } +} + +// Panes merge cell by cell: every pane seeds row `r` the same way, so merging +// two panes equals building the union in one pane. +#[test] +fn plain_count_sketch_merge_equals_union_build() { + let family = plain_count_sketch(1_001, 7); + let events = zipf_events(500); + let (left, right) = events.split_at(events.len() / 3); + let merged = build_plain_count_sketch(&family, left.iter().copied()) + .merge_with(&*build_plain_count_sketch(&family, right.iter().copied())) + .unwrap(); + let union = build_plain_count_sketch(&family, events.iter().copied()); + let cells = |state: &dyn AggregateCore| { + serde_json::to_value( + state + .as_any() + .downcast_ref::() + .unwrap(), + ) + .unwrap()["cells"] + .clone() + }; + assert_eq!(cells(&*merged), cells(&*union)); + for i in 0..500 { + assert_eq!(plain_estimate(&*merged, i), plain_estimate(&*union, i)); + } + // A different shape is a different family and does not merge. + assert!(merged + .merge_with(&*build_plain_count_sketch( + &plain_count_sketch(1_001, 5), + [0] + )) + .is_err()); +} diff --git a/crates/asap-physical-operators/tests/plan_properties.rs b/crates/asap-physical-operators/tests/plan_properties.rs index 231aecf93..b34b9f4b3 100644 --- a/crates/asap-physical-operators/tests/plan_properties.rs +++ b/crates/asap-physical-operators/tests/plan_properties.rs @@ -152,6 +152,43 @@ fn kll_kernel_rejects_a_k_above_the_library_cap() { assert!(validate_summary_kernel(&kll(26_603), &update, &grouping).is_err()); } +/// A plain CountSketch reads the median of its rows, so plan validation and +/// kernel construction both require a positive odd depth. +#[test] +fn count_sketch_kernel_requires_an_odd_depth() { + use asap_physical_operators::{ + capability::validate_summary_kernel, factory::create_planner_accumulator, + }; + use planner_types::post_asap::{ + GroupingStrategy, SketchAlgorithm, SketchKind, SketchParams, SummaryInputExpr, + SummaryUpdate, + }; + use planner_types::pre_asap::ColumnRef; + let grouping = GroupingStrategy::default(); + let mut update = SummaryUpdate::column(ColumnRef::SampleValue); + update.item = Some(SummaryInputExpr::Column(ColumnRef::Named("host".into()))); + let count_sketch = |depth| { + FieldDataType::Sketch( + SketchKind::new( + SketchAlgorithm::CountSketch, + SketchParams::CountSketch { width: 64, depth }, + ), + grouping.clone(), + ) + }; + for (depth, valid) in [(5, true), (4, false)] { + let family = count_sketch(depth); + assert_eq!( + validate_summary_kernel(&family, &update, &grouping).is_ok(), + valid + ); + assert_eq!( + create_planner_accumulator(&family, &update, &grouping).is_ok(), + valid + ); + } +} + // Kernel support must not be mistaken for executable native state/evaluation support. #[test] fn summary_capability_levels_are_distinct() { From 16facc759402b88aef844c3997b95ea6e491f8e3 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Thu, 15 Oct 2026 14:17:53 +0000 Subject: [PATCH 3/5] fix(asap-physical-operators): build top-k sketches on the per-row-seeded kernel create_planner_accumulator built CmsWithHeap and CountSketchWithHeap on asap_sketchlib's portable CMSHeap and CSHeap. Those hash row r with SEEDLIST[r % 20], so row r + 20 copies row r and the planner's depths (23 to 149 rows for CountSketch) are not independent. They also round weights and drop nonpositive ones, and the DAG cannot load their states, since it expects the weighted frequency kernel. Both families, and the plain CountSketch, now build through one WeightedFrequencyUpdater on sketchlib's per-row-seeded weighted frequency kernel, with the DAG keyed build's Utf8 identities and exact Float64 weights. A negative Count-Min input fails validation instead of being dropped. The portable heap updaters stay for callers that construct them directly. Tests: 25-row Count-Min and CountSketch heap states keep every pair of rows independent; factory heap states load as typed DAG summary values and keep fractional weights exactly; a negative Count-Min input is rejected. --- .../src/summary_kernels/factory.rs | 148 ++++++++---------- .../tests/deployment.rs | 135 +++++++++++++--- 2 files changed, 181 insertions(+), 102 deletions(-) diff --git a/crates/asap-physical-operators/src/summary_kernels/factory.rs b/crates/asap-physical-operators/src/summary_kernels/factory.rs index 3b92fbe73..1d9da341e 100644 --- a/crates/asap-physical-operators/src/summary_kernels/factory.rs +++ b/crates/asap-physical-operators/src/summary_kernels/factory.rs @@ -261,8 +261,8 @@ pub enum TopkWeight { /// * `Value` → `inner.update(key, value)` adds the datapoint value (Σ value). /// * `Count` → `inner.update(key, 1.0)` adds one per event (Σ count). /// -/// Both `CountMinSketchWithHeap` and `CountSketchWithHeap` raw-input policies -/// route here; the heap is the shared distinguishing payload. +/// `create_planner_accumulator` builds Planner `CmsWithHeap` and +/// `CountSketchWithHeap` families with [`WeightedFrequencyUpdater`] instead. pub struct CmsHeapAccumulatorUpdater { acc: CountMinSketchWithHeapAccumulator, row_num: usize, @@ -324,36 +324,57 @@ impl AccumulatorUpdater for CmsHeapAccumulatorUpdater { } // --------------------------------------------------------------------------- -// CountSketchAccumulatorUpdater (median-of-signed-rows CountSketch, no heap) +// WeightedFrequencyUpdater (CountSketch, Count-Min with heap, CountSketch with heap) // --------------------------------------------------------------------------- -/// Keyed point-frequency updater backed by sketchlib's per-row-seeded -/// weighted frequency kernel in CountSketch mode with no candidate heap -/// (signed rows, median-of-rows estimator) — distinct math from -/// `CmsAccumulatorUpdater`'s CMS (min-of-rows). Each label value of the key -/// is one `Utf8` identity, as in the keyed heap build. +/// Keyed frequency updater backed by sketchlib's per-row-seeded weighted +/// frequency kernel, the state the DAG's keyed summary build produces: plain +/// CountSketch (no candidate heap), Count-Min with a top-k heap, or +/// CountSketch with a top-k heap. Each row hashes with its own seed, so rows +/// stay independent at any depth. Each label value of the key is one `Utf8` +/// identity, as in the DAG's keyed build. /// -/// As with bare CMS, each raw Prometheus sample contributes its `value`. -/// Unit event counting must be selected explicitly by a future typed plan -/// contract rather than being implied by `SketchAlgorithm::CountSketch`. -pub struct CountSketchAccumulatorUpdater { +/// Each raw Prometheus sample contributes its `value` as a Float64 weight, +/// kept exactly (no rounding). CountSketch accepts signed weights; Count-Min +/// accepts nonnegative ones and rejects a negative input at validation. +pub struct WeightedFrequencyUpdater { acc: WeightedFrequency, - row_num: usize, - col_num: usize, -} - -impl CountSketchAccumulatorUpdater { - /// Fails for a zero or even `row_num`: the median needs an odd row count. - pub fn new(row_num: usize, col_num: usize) -> Result { + algorithm: FrequencyAlgorithm, + width: usize, + depth: usize, + capacity: usize, +} + +impl WeightedFrequencyUpdater { + /// Fails for a zero dimension, or an even CountSketch `depth` (the median + /// needs an odd row count). + pub fn new( + algorithm: FrequencyAlgorithm, + width: usize, + depth: usize, + capacity: usize, + ) -> Result { Ok(Self { - acc: WeightedFrequency::new(FrequencyAlgorithm::CountSketch, col_num, row_num, 0)?, - row_num, - col_num, + acc: WeightedFrequency::new(algorithm, width, depth, capacity)?, + algorithm, + width, + depth, + capacity, }) } } -impl AccumulatorUpdater for CountSketchAccumulatorUpdater { +impl AccumulatorUpdater for WeightedFrequencyUpdater { + fn validate_single_input(&self, value: f64) -> Result<(), String> { + if !value.is_finite() { + Err("accumulator input must be finite".into()) + } else if self.algorithm == FrequencyAlgorithm::Cms && value < 0.0 { + Err("Count-Min input must be nonnegative".into()) + } else { + Ok(()) + } + } + fn update_single(&mut self, _value: f64, _timestamp_ms: i64) { debug_assert!( false, @@ -367,23 +388,18 @@ impl AccumulatorUpdater for CountSketchAccumulatorUpdater { .iter() .map(|label| crate::values::Value::Utf8(label.as_str().into())) .collect::>(); - // Inputs pass `validate_single_input` (finite) first; only a sum past + // Inputs pass `validate_single_input` first; only a sum past // `f64::MAX` can fail here. self.acc .update(&identity, value) - .expect("CountSketch counter overflow"); + .expect("weighted frequency counter overflow"); } impl_clone_accumulator_methods!(acc); fn reset(&mut self) { - self.acc = WeightedFrequency::new( - FrequencyAlgorithm::CountSketch, - self.col_num, - self.row_num, - 0, - ) - .expect("dimensions validated at construction"); + self.acc = WeightedFrequency::new(self.algorithm, self.width, self.depth, self.capacity) + .expect("dimensions validated at construction"); } fn is_keyed(&self) -> bool { @@ -399,6 +415,9 @@ impl AccumulatorUpdater for CountSketchAccumulatorUpdater { // CountSketchWithHeapAccumulatorUpdater (real CountSketch + top-k heap) // --------------------------------------------------------------------------- +/// `create_planner_accumulator` builds Planner `CountSketchWithHeap` families +/// with [`WeightedFrequencyUpdater`] instead. +/// /// Keyed top-k updater backed by a real `CountSketchWithHeap` (signed-row /// CountSketch matrix PLUS a size-`heap_size` top-k heap). Distinct math from /// `CmsHeapAccumulatorUpdater`'s CMS-with-heap (min-of-rows); shares the same @@ -521,26 +540,6 @@ fn cms_dims(params: &SketchParams) -> (usize, usize) { } } -/// Read `(rows = depth, columns = width, heap_size)` out of `SketchParams::CmsWithHeap` -/// or `::CountSketchWithHeap`. -fn cms_heap_dims(params: &SketchParams) -> (usize, usize, usize) { - match params { - SketchParams::CmsWithHeap { - width, - depth, - heap_size, - } - | SketchParams::CountSketchWithHeap { - width, - depth, - heap_size, - } => (*depth as usize, *width as usize, *heap_size as usize), - other => unreachable!( - "accumulator_spec() paired a WithHeap SketchAlgorithm with unexpected params: {other:?}" - ), - } -} - /// Construct the kernel declared by a Planner SummaryAgg. No deployment config /// tags participate in this dispatch and unsupported payloads are errors. pub fn create_planner_accumulator( @@ -599,23 +598,18 @@ pub fn create_planner_accumulator( } (SketchAlgorithm::CountSketch, params @ SketchParams::CountSketch { .. }) => { let (r, c) = cms_dims(params); - Box::new(CountSketchAccumulatorUpdater::new(r, c).map_err(|e| e.to_string())?) - } - (SketchAlgorithm::CmsWithHeap, params @ SketchParams::CmsWithHeap { .. }) => { - let (r, c, h) = cms_heap_dims(params); - Box::new(CmsHeapAccumulatorUpdater::new(r, c, h, TopkWeight::Value)) + Box::new( + WeightedFrequencyUpdater::new(FrequencyAlgorithm::CountSketch, c, r, 0) + .map_err(|e| e.to_string())?, + ) } - ( - SketchAlgorithm::CountSketchWithHeap, - params @ SketchParams::CountSketchWithHeap { .. }, - ) => { - let (r, c, h) = cms_heap_dims(params); - Box::new(CountSketchWithHeapAccumulatorUpdater::new( - r, - c, - h, - TopkWeight::Value, - )) + (SketchAlgorithm::CmsWithHeap | SketchAlgorithm::CountSketchWithHeap, _) => { + let (algorithm, width, depth, capacity) = + WeightedFrequency::configuration(kind).map_err(|e| e.to_string())?; + Box::new( + WeightedFrequencyUpdater::new(algorithm, width, depth, capacity) + .map_err(|e| e.to_string())?, + ) } (SketchAlgorithm::Hll, SketchParams::Hll { precision }) => Box::new(HllUpdater { acc: HllSketchAccumulator::new( @@ -806,7 +800,9 @@ mod planner_parameter_regression { .unwrap(); (s.inner.rows(), s.inner.cols()) } - SketchAlgorithm::CountSketch => { + SketchAlgorithm::CountSketch + | SketchAlgorithm::CmsWithHeap + | SketchAlgorithm::CountSketchWithHeap => { let (width, depth, _) = state .as_any() .downcast_ref::() @@ -814,20 +810,6 @@ mod planner_parameter_regression { .shape(); (depth, width) } - SketchAlgorithm::CmsWithHeap => { - let s = state - .as_any() - .downcast_ref::() - .unwrap(); - (s.inner.rows(), s.inner.cols()) - } - SketchAlgorithm::CountSketchWithHeap => { - let s = state - .as_any() - .downcast_ref::() - .unwrap(); - (s.inner.rows(), s.inner.cols()) - } _ => unreachable!(), }; assert_eq!(dims, (3, 128), "{algorithm:?}"); diff --git a/crates/asap-physical-operators/tests/deployment.rs b/crates/asap-physical-operators/tests/deployment.rs index e87b55749..65e0f1752 100644 --- a/crates/asap-physical-operators/tests/deployment.rs +++ b/crates/asap-physical-operators/tests/deployment.rs @@ -91,9 +91,14 @@ fn native_count_sketch_dimensions_are_not_packed_wire_dimensions() { let state = operator.into_accumulator(); let state = state .as_any() - .downcast_ref::() + .downcast_ref::() .unwrap(); - assert_eq!(state.query_key(&key), 7.0); + assert_eq!( + state + .point_estimate(&[asap_physical_operators::values::Value::Utf8("a".into())]) + .unwrap(), + 7.0 + ); } fn plain_count_sketch(width: u32, depth: u32) -> FieldDataType { @@ -115,11 +120,11 @@ fn build_plain_count_sketch( family: &FieldDataType, events: impl IntoIterator, ) -> Box { - build_weighted_plain_count_sketch(family, events.into_iter().map(|i| (i, 1.0))) + build_weighted_frequency(family, events.into_iter().map(|i| (i, 1.0))) } -/// Build a plain CountSketch over `(key index, weight)` updates. -fn build_weighted_plain_count_sketch( +/// Build a keyed frequency summary over `(key index, weight)` updates. +fn build_weighted_frequency( family: &FieldDataType, updates: impl IntoIterator, ) -> Box { @@ -153,28 +158,25 @@ fn zipf_events(keys: usize) -> Vec { .collect() } -// A deep plain CountSketch, sized from a small δ, builds and keeps every row -// independent: each row hashes with its own seed, so no row geometry limit -// applies and two rows rarely put a key in the same column. -#[test] -fn deep_plain_count_sketch_rows_hash_independently() { - let (width, depth) = (30_000usize, 15usize); - let family = plain_count_sketch(width as u32, depth as u32); - let keys = 2_000; - // Key `base + p` of a batch carries weight 3^p, so a cell holds a signed - // sum whose balanced-ternary digits name the keys that hit it. +/// Decode, for `keys` keys, the column each row of `family` (a keyed +/// frequency sketch of `width` x `depth`) puts the key in, and assert every +/// pair of rows shares a column for few keys, as independent rows do. +fn assert_rows_hash_independently(family: &FieldDataType, width: usize, depth: usize, keys: usize) { + // Key `base + p` of a batch carries weight 3^p, so a cell holds a sum + // whose balanced-ternary digits name the keys that hit it (CountSketch + // signs give digits -1 or 1, Count-Min only 1). const BATCH: usize = 33; let mut columns = vec![vec![None; depth]; keys]; for base in (0..keys).step_by(BATCH) { let batch = base..keys.min(base + BATCH); - let state = build_weighted_plain_count_sketch( - &family, + let state = build_weighted_frequency( + family, batch.clone().map(|i| (i, 3f64.powi((i - base) as i32))), ); let state = state .as_any() .downcast_ref::() - .unwrap(); + .expect("the per-row-seeded weighted frequency kernel"); let cells = serde_json::to_value(state).unwrap()["cells"].clone(); for (index, cell) in cells.as_array().unwrap().iter().enumerate() { let (row, column) = (index / width, index % width); @@ -204,12 +206,107 @@ fn deep_plain_count_sketch_rows_hash_independently() { for a in 0..depth { for b in a + 1..depth { let shared = columns.iter().filter(|c| c[a] == c[b]).count(); - // Independent rows share a column for 2000 / 30000 < 1 key per pair. + // Independent rows share a column for keys / width < 1 key per pair. assert!(shared <= 5, "rows {a} and {b} share {shared} columns"); } } } +// A deep plain CountSketch, sized from a small δ, builds and keeps every row +// independent: each row hashes with its own seed, so no row geometry limit +// applies and two rows rarely put a key in the same column. +#[test] +fn deep_plain_count_sketch_rows_hash_independently() { + assert_rows_hash_independently(&plain_count_sketch(30_000, 15), 30_000, 15, 2_000); +} + +fn heap_sketch( + algorithm: SketchAlgorithm, + width: u32, + depth: u32, + heap_size: u32, +) -> FieldDataType { + let params = match algorithm { + SketchAlgorithm::CmsWithHeap => SketchParams::CmsWithHeap { + width, + depth, + heap_size, + }, + _ => SketchParams::CountSketchWithHeap { + width, + depth, + heap_size, + }, + }; + FieldDataType::Sketch(SketchKind::new(algorithm, params), Default::default()) +} + +// The factory's top-k sketches use the same per-row-seeded kernel: past 20 +// rows (where a fixed seed list would repeat) every row stays independent. +#[test] +fn deep_heap_sketch_rows_hash_independently() { + for algorithm in [ + SketchAlgorithm::CmsWithHeap, + SketchAlgorithm::CountSketchWithHeap, + ] { + assert_rows_hash_independently(&heap_sketch(algorithm, 30_000, 25, 4), 30_000, 25, 1_000); + } +} + +// A factory-built top-k state is the DAG's state type, so it loads as a typed +// summary value, and it keeps fractional weights exactly. +#[test] +fn heap_sketch_states_load_as_dag_summary_values() { + use asap_physical_operators::{ + planner::post_asap::{Field, Schema}, + summary_kernels::weighted_frequency::WeightedFrequency, + values::{Batch, Value}, + }; + for algorithm in [ + SketchAlgorithm::CmsWithHeap, + SketchAlgorithm::CountSketchWithHeap, + ] { + let family = heap_sketch(algorithm, 1_024, 5, 4); + let state = build_weighted_frequency(&family, [(0, 0.25), (0, 0.5), (1, 2.0)]); + let estimate = state + .as_any() + .downcast_ref::() + .expect("the DAG's weighted frequency state") + .point_estimate(&[Value::Utf8("host-0".into())]) + .unwrap(); + assert_eq!(estimate, 0.75); + let schema = std::sync::Arc::new(Schema { + fields: vec![Field { + name: "state".into(), + dtype: family.clone(), + nullable: false, + table: None, + }], + unique_keys: vec![], + closed: false, + time_index: None, + }); + let row = vec![Value::Summary { + family, + state: std::sync::Arc::from(state), + }]; + assert!(Batch::try_new(schema, vec![row]).is_ok()); + } +} + +// Count-Min needs nonnegative weights; a negative input is an error at +// validation, never silently dropped. +#[test] +fn count_min_heap_rejects_negative_inputs() { + use asap_physical_operators::planner::post_asap::SummaryInputExpr; + let mut update = SummaryUpdate::column(ColumnRef::SampleValue); + update.item = Some(SummaryInputExpr::Column(ColumnRef::Named("host".into()))); + let family = heap_sketch(SketchAlgorithm::CmsWithHeap, 64, 3, 4); + let operator = create_planner_accumulator(&family, &update, &Default::default()).unwrap(); + assert!(operator.validate_single_input(-1.0).is_err()); + assert!(operator.validate_single_input(0.5).is_ok()); +} + // The planner certifies `|estimate − truth| ≤ √(3/w) ‖f‖₂` for every point // query of a plain CountSketch; a deep shape meets it on a skewed stream. #[test] From c79e519d5e30019d29f13cab0729ce3fed5a74b2 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Thu, 15 Oct 2026 14:18:48 +0000 Subject: [PATCH 4/5] fix(asap-physical-operators): count KMV's reserved hashes in its memory asap_sketchlib's KMV reserves all k hash slots at construction, but approx_memory_bytes counted only the hashes held so far, so a fresh or sparse state reported far less than it holds. It now counts k * 8 bytes. --- .../src/summary_kernels/kmv.rs | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) diff --git a/crates/asap-physical-operators/src/summary_kernels/kmv.rs b/crates/asap-physical-operators/src/summary_kernels/kmv.rs index 10e28ad3d..543794ed0 100644 --- a/crates/asap-physical-operators/src/summary_kernels/kmv.rs +++ b/crates/asap-physical-operators/src/summary_kernels/kmv.rs @@ -74,8 +74,9 @@ impl AggregateCore for KmvAccumulator { } fn approx_memory_bytes(&self) -> usize { + // sketchlib reserves all k hash slots at construction. std::mem::size_of::() - .saturating_add(self.inner.k_vals.len() * std::mem::size_of::()) + .saturating_add(self.inner.k.saturating_mul(std::mem::size_of::())) } } @@ -195,4 +196,17 @@ mod tests { assert_eq!(cardinality(kmv.take_accumulator().as_ref()), 2.0); assert_eq!(cardinality(&kmv), 0.0); } + + // sketchlib reserves all k hash slots at construction, so an empty state + // already holds k * 8 bytes. + #[test] + fn memory_counts_the_k_reserved_hashes() { + let k = 4096; + let empty = KmvAccumulator::new(k); + assert!(empty.approx_memory_bytes() >= k as usize * 8); + assert_eq!( + empty.approx_memory_bytes(), + build(k, 0..10).approx_memory_bytes() + ); + } } From 6cf3454a55355b5153921cd14487ff949dce65fb Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Thu, 15 Oct 2026 14:20:07 +0000 Subject: [PATCH 5/5] fix(asap-aware-mapping): reject empirical cost evidence for even CountSketch depths A CountSketch reads the median of its rows, and the executor builds only odd depths, so evidence measured at an even CountSketch or CountSketch-with-heap depth describes no executable shape. valid_params now rejects it; Count-Min depths stay unrestricted. --- .../asap-aware-mapping/src/empirical_cost.rs | 46 ++++++++++++++++--- 1 file changed, 40 insertions(+), 6 deletions(-) diff --git a/crates/asap-aware-mapping/src/empirical_cost.rs b/crates/asap-aware-mapping/src/empirical_cost.rs index 1724a24ea..3a4e54447 100644 --- a/crates/asap-aware-mapping/src/empirical_cost.rs +++ b/crates/asap-aware-mapping/src/empirical_cost.rs @@ -372,9 +372,10 @@ fn valid_params(algorithm: &SketchAlgorithm, params: &SketchParams) -> bool { (SketchAlgorithm::DDSketch, SketchParams::DDSketch { alpha }) => { alpha.is_finite() && *alpha > 0.0 && *alpha < 1.0 } - (SketchAlgorithm::Cms, SketchParams::Cms { width, depth }) - | (SketchAlgorithm::CountSketch, SketchParams::CountSketch { width, depth }) => { - *width > 0 && *depth > 0 + (SketchAlgorithm::Cms, SketchParams::Cms { width, depth }) => *width > 0 && *depth > 0, + // A CountSketch reads the median of its rows: the depth must be odd. + (SketchAlgorithm::CountSketch, SketchParams::CountSketch { width, depth }) => { + *width > 0 && depth % 2 == 1 } ( SketchAlgorithm::CmsWithHeap, @@ -383,15 +384,15 @@ fn valid_params(algorithm: &SketchAlgorithm, params: &SketchParams) -> bool { depth, heap_size, }, - ) - | ( + ) => *width > 0 && *depth > 0 && *heap_size > 0, + ( SketchAlgorithm::CountSketchWithHeap, SketchParams::CountSketchWithHeap { width, depth, heap_size, }, - ) => *width > 0 && *depth > 0 && *heap_size > 0, + ) => *width > 0 && depth % 2 == 1 && *heap_size > 0, _ => false, } } @@ -692,4 +693,37 @@ mod tests { Err(EvidenceError::Ambiguous) ); } + + /// A CountSketch reads the median of its rows, so evidence for an even + /// depth describes no executable shape. + #[test] + fn count_sketch_evidence_requires_an_odd_depth() { + for (depth, valid) in [(5, true), (4, false)] { + assert_eq!( + valid_params( + &SketchAlgorithm::CountSketch, + &SketchParams::CountSketch { width: 64, depth } + ), + valid + ); + assert_eq!( + valid_params( + &SketchAlgorithm::CountSketchWithHeap, + &SketchParams::CountSketchWithHeap { + width: 64, + depth, + heap_size: 8 + } + ), + valid + ); + } + assert!(valid_params( + &SketchAlgorithm::Cms, + &SketchParams::Cms { + width: 64, + depth: 4 + } + )); + } }