diff --git a/Cargo.lock b/Cargo.lock index bce4c50f..adfa58ef 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-aware-mapping/src/empirical_cost.rs b/crates/asap-aware-mapping/src/empirical_cost.rs index 1724a24e..3a4e5444 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 + } + )); + } } diff --git a/crates/asap-physical-operators/Cargo.toml b/crates/asap-physical-operators/Cargo.toml index bdc93848..77d01f9c 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/capability.rs b/crates/asap-physical-operators/src/capability.rs index 8865396a..c8441313 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 c3aaa40c..00000000 --- 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 4e216bdd..1d9da341 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}; @@ -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,33 +324,57 @@ impl AccumulatorUpdater for CmsHeapAccumulatorUpdater { } // --------------------------------------------------------------------------- -// CountSketchAccumulatorUpdater (real median-of-signed-rows CountSketch) +// WeightedFrequencyUpdater (CountSketch, Count-Min with heap, CountSketch with heap) // --------------------------------------------------------------------------- -/// Keyed point-frequency updater backed by a real `asap_sketchlib::CountSketch` -/// (signed rows, median-of-rows estimator) — distinct math from -/// `CmsAccumulatorUpdater`'s CMS (min-of-rows). +/// 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 { - acc: CountSketchAccumulator, - row_num: usize, - col_num: usize, +/// 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, + 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(algorithm, width, depth, capacity)?, + algorithm, + width, + depth, + capacity, + }) + } } -impl CountSketchAccumulatorUpdater { - pub fn new(row_num: usize, col_num: usize) -> Self { - Self { - acc: CountSketchAccumulator::new(row_num, col_num), - row_num, - col_num, +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(()) } } -} -impl AccumulatorUpdater for CountSketchAccumulatorUpdater { fn update_single(&mut self, _value: f64, _timestamp_ms: i64) { debug_assert!( false, @@ -359,13 +383,23 @@ 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` first; only a sum past + // `f64::MAX` can fail here. + self.acc + .update(&identity, value) + .expect("weighted frequency 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(self.algorithm, self.width, self.depth, self.capacity) + .expect("dimensions validated at construction"); } fn is_keyed(&self) -> bool { @@ -373,8 +407,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() } } @@ -382,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 @@ -504,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( @@ -582,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)) - } - (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( @@ -789,26 +800,15 @@ mod planner_parameter_regression { .unwrap(); (s.inner.rows(), s.inner.cols()) } - SketchAlgorithm::CountSketch => { - let s = state - .as_any() - .downcast_ref::() - .unwrap(); - (s.inner.rows, s.inner.cols) - } - SketchAlgorithm::CmsWithHeap => { - let s = state - .as_any() - .downcast_ref::() - .unwrap(); - (s.inner.rows(), s.inner.cols()) - } - SketchAlgorithm::CountSketchWithHeap => { - let s = state + SketchAlgorithm::CountSketch + | SketchAlgorithm::CmsWithHeap + | SketchAlgorithm::CountSketchWithHeap => { + let (width, depth, _) = state .as_any() - .downcast_ref::() - .unwrap(); - (s.inner.rows(), s.inner.cols()) + .downcast_ref::() + .unwrap() + .shape(); + (depth, width) } _ => unreachable!(), }; diff --git a/crates/asap-physical-operators/src/summary_kernels/kmv.rs b/crates/asap-physical-operators/src/summary_kernels/kmv.rs index 10e28ad3..543794ed 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() + ); + } } diff --git a/crates/asap-physical-operators/src/summary_kernels/mod.rs b/crates/asap-physical-operators/src/summary_kernels/mod.rs index b30e87fc..5efef891 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/src/summary_kernels/weighted_frequency.rs b/crates/asap-physical-operators/src/summary_kernels/weighted_frequency.rs index 7c112230..39b7be95 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) diff --git a/crates/asap-physical-operators/tests/deployment.rs b/crates/asap-physical-operators/tests/deployment.rs index 42c85ca5..65e0f175 100644 --- a/crates/asap-physical-operators/tests/deployment.rs +++ b/crates/asap-physical-operators/tests/deployment.rs @@ -91,7 +91,272 @@ 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 { + 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_frequency(family, events.into_iter().map(|i| (i, 1.0))) +} + +/// Build a keyed frequency summary over `(key index, weight)` updates. +fn build_weighted_frequency( + 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() +} + +/// 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_frequency( + family, + batch.clone().map(|i| (i, 3f64.powi((i - base) as i32))), + ); + let state = state + .as_any() + .downcast_ref::() + .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); + 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 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] +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 231aecf9..b34b9f4b 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() {