diff --git a/Cargo.lock b/Cargo.lock index adfa58ef1..063da9b90 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -319,7 +319,7 @@ version = "0.1.0" dependencies = [ "asap-frontend-promql", "asap-types", - "asap_sketchlib 0.3.0 (git+https://github.com/ProjectASAP/asap_sketchlib?rev=831a5045847010a3a65f5a8e614e0208581d4a05)", + "asap_sketchlib", "serde", "serde_json", "thiserror 2.0.18", @@ -394,7 +394,7 @@ dependencies = [ "asap-physical-operators", "asap-planner", "asap-types", - "asap_sketchlib 0.3.0 (git+https://github.com/ProjectASAP/asap_sketchlib?rev=831a5045847010a3a65f5a8e614e0208581d4a05)", + "asap_sketchlib", "futures", "serde_json", "tokio", @@ -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=9415c5b161237ad238e101ed9166d4dbaace3031)", + "asap_sketchlib", "chrono", "futures", "regex", @@ -447,24 +447,7 @@ dependencies = [ [[package]] name = "asap_sketchlib" version = "0.3.0" -source = "git+https://github.com/ProjectASAP/asap_sketchlib?rev=831a5045847010a3a65f5a8e614e0208581d4a05#831a5045847010a3a65f5a8e614e0208581d4a05" -dependencies = [ - "bytes", - "prost", - "rand 0.9.5", - "rmp-serde", - "serde", - "serde-big-array", - "serde_bytes", - "smallvec", - "twox-hash", - "xxhash-rust", -] - -[[package]] -name = "asap_sketchlib" -version = "0.3.0" -source = "git+https://github.com/ProjectASAP/asap_sketchlib?rev=9415c5b161237ad238e101ed9166d4dbaace3031#9415c5b161237ad238e101ed9166d4dbaace3031" +source = "git+https://github.com/ProjectASAP/asap_sketchlib?rev=3f44cf1c2d823b844c17796f24408e03863149ee#3f44cf1c2d823b844c17796f24408e03863149ee" dependencies = [ "bincode", "bytes", diff --git a/Cargo.toml b/Cargo.toml index 1d2a19198..a0d811d05 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,4 +17,4 @@ exclude = ["crates/metricsql-parser-vendored"] resolver = "2" [workspace.dependencies] -asap_sketchlib = { git = "https://github.com/ProjectASAP/asap_sketchlib", rev = "831a5045847010a3a65f5a8e614e0208581d4a05" } +asap_sketchlib = { git = "https://github.com/ProjectASAP/asap_sketchlib", rev = "3f44cf1c2d823b844c17796f24408e03863149ee" } diff --git a/crates/asap-aware-mapping/src/accuracy/estimators/univmon.rs b/crates/asap-aware-mapping/src/accuracy/estimators/univmon.rs index 8dffc8a62..ece5b028d 100644 --- a/crates/asap-aware-mapping/src/accuracy/estimators/univmon.rs +++ b/crates/asap-aware-mapping/src/accuracy/estimators/univmon.rs @@ -20,7 +20,7 @@ use asap_types::pre_asap::AggIntent; /// Layers `asap_sketchlib` builds: layer `i` hashes at seed index `i`, and /// seed 19 picks the layers a key reaches. -pub(crate) const MAX_LAYERS: u8 = 19; +pub(crate) const MAX_LAYERS: u8 = asap_sketchlib::sketch_framework::univmon::MAX_LAYER_SIZE as u8; /// Rows the executor builds an L2 sketch with: the deepest odd shape under /// its 20-row limit. const MAX_L2_ROWS: u32 = 19; @@ -91,19 +91,13 @@ pub(super) fn guarantee( } } -/// Whether `asap_sketchlib` builds this shape with independent hashes: at -/// most [`MAX_LAYERS`] layers, and `rows · (⌈log2 cols⌉ + 1) ≤ 128`. +/// Whether `asap_sketchlib` builds this shape, with independent hashes: at +/// most [`MAX_LAYERS`] layers and 20 rows, and `rows · (⌈log2 cols⌉ + 1) ≤ +/// 128` ([`asap_sketchlib::UnivMon::check_shape`]; the heap size is not part +/// of the hash rules). pub(crate) fn hash_independent(rows: u32, cols: u32, layers: u8) -> bool { - rows >= 1 - && cols >= 1 - && (1..=MAX_LAYERS).contains(&layers) - && rows.saturating_mul(column_bits(cols) + 1) <= HASH_BITS -} - -/// `⌈log2 cols⌉`, the column bits a row reads. -fn column_bits(cols: u32) -> u32 { - cols.checked_next_power_of_two() - .map_or(u32::BITS, u32::trailing_zeros) + asap_sketchlib::UnivMon::check_shape(1, rows as usize, cols as usize, usize::from(layers)) + .is_ok() } /// The relative error of the L2 norm at `cols` columns. Per row, the @@ -495,6 +489,16 @@ mod tests { } } + /// The library builds at most 20 rows, so a 21-row UnivMon certifies + /// nothing even though its hashes fit the budget (21 · (3 + 1) = 84 ≤ + /// 128 at 8 columns). + #[test] + fn univmon_past_twenty_rows_is_not_hash_independent() { + assert!(hash_independent(20, 8, 1)); + assert!(!hash_independent(21, 8, 1)); + assert!(l2(&univmon(1, 21, 8, 1, None)).is_none()); + } + /// Each distinct-count heap meets its target once the source's bound /// is bound, and one fewer key does not. #[test] diff --git a/crates/asap-aware-mapping/src/empirical_cost.rs b/crates/asap-aware-mapping/src/empirical_cost.rs index 9596a6363..176a49c28 100644 --- a/crates/asap-aware-mapping/src/empirical_cost.rs +++ b/crates/asap-aware-mapping/src/empirical_cost.rs @@ -373,9 +373,9 @@ fn valid_params(algorithm: &SketchAlgorithm, params: &SketchParams) -> bool { alpha.is_finite() && *alpha > 0.0 && *alpha < 1.0 } (SketchAlgorithm::Cms, SketchParams::Cms { width, depth }) => *width > 0 && *depth > 0, - // A CountSketch reads the median of its rows: the depth must be odd. + // The kernel's shape rules (an odd depth, for the median). (SketchAlgorithm::CountSketch, SketchParams::CountSketch { width, depth }) => { - *width > 0 && depth % 2 == 1 + count_sketch_builds(*width, *depth) } ( SketchAlgorithm::CmsWithHeap, @@ -392,11 +392,22 @@ fn valid_params(algorithm: &SketchAlgorithm, params: &SketchParams) -> bool { depth, heap_size, }, - ) => *width > 0 && depth % 2 == 1 && *heap_size > 0, + ) => count_sketch_builds(*width, *depth) && *heap_size > 0, _ => false, } } +/// Whether `asap_sketchlib`'s CountSketch kernel builds a `width x depth` +/// shape, so evidence for it describes an executable shape. +fn count_sketch_builds(width: u32, depth: u32) -> bool { + asap_sketchlib::WeightedFrequency::check_shape( + asap_sketchlib::FrequencyAlgorithm::CountSketch, + width as usize, + depth as usize, + ) + .is_ok() +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/asap-physical-operators/Cargo.toml b/crates/asap-physical-operators/Cargo.toml index 77d01f9c1..2a159ef31 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 = "9415c5b161237ad238e101ed9166d4dbaace3031", features = ["experimental"] } +asap_sketchlib = { workspace = true, 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 b17435298..c02310bbf 100644 --- a/crates/asap-physical-operators/src/capability.rs +++ b/crates/asap-physical-operators/src/capability.rs @@ -62,9 +62,15 @@ 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 }) if valid_matrix(*width, *depth) => true, - // The median needs an odd row count. + // The kernel's shape rules (an odd depth, for the median). (A::CountSketch, P::CountSketch { width, depth }) - if valid_matrix(*width, *depth) && depth % 2 == 1 => + if valid_matrix(*width, *depth) + && asap_sketchlib::WeightedFrequency::check_shape( + asap_sketchlib::FrequencyAlgorithm::CountSketch, + *width as usize, + *depth as usize, + ) + .is_ok() => { true } @@ -93,7 +99,7 @@ pub fn validate_summary_kernel( layers, .. }, - ) if crate::summary_kernels::univmon::check_dimensions( + ) if asap_sketchlib::UnivMon::check_shape( *heap_size as usize, *sketch_rows as usize, *sketch_cols as usize, diff --git a/crates/asap-physical-operators/src/summary_kernels/count_min_sketch_with_heap.rs b/crates/asap-physical-operators/src/summary_kernels/count_min_sketch_with_heap.rs deleted file mode 100644 index f2d8786c2..000000000 --- a/crates/asap-physical-operators/src/summary_kernels/count_min_sketch_with_heap.rs +++ /dev/null @@ -1,237 +0,0 @@ -use crate::{AggregateCore, KeyByLabelValues}; -use asap_sketchlib::CountMinSketchWithHeap; - -/// Count-Min Sketch with a top-k heap over `asap_sketchlib::CountMinSketchWithHeap`. -#[derive(Debug, Clone)] -pub struct CountMinSketchWithHeapAccumulator { - pub inner: CountMinSketchWithHeap, -} - -impl CountMinSketchWithHeapAccumulator { - pub fn new(row_num: usize, col_num: usize, heap_size: usize) -> Self { - Self { - inner: CountMinSketchWithHeap::new(row_num, col_num, heap_size), - } - } - - pub fn query_key(&self, key: &KeyByLabelValues) -> f64 { - let key_string = key.labels.join(";"); - self.inner.estimate(&key_string) - } - - /// VALUE-WEIGHTED heavy-hitter update (FIX: CountSketch/CMS topk - /// recall-0). The default ingest path inserts `+1` per occurrence keyed - /// by the raw `item`, so the heap ranks groups by OCCURRENCE COUNT — the - /// wrong answer for `topk(k, sum by (label) (metric))`, which asks for - /// the top groups by SUM OF VALUE. This update adds the sample `value` - /// (not `+1`) into both the CMS matrix and the top-k heap, keyed by the - /// GROUP LABEL (e.g. the `host` / `zone` value), so the heap's ranking is - /// by summed value. Repeated calls for the same `group_label` accumulate, - /// so after folding a window the heap holds Σvalue per group. - /// - /// Delegates to the library's value-weighted `CountMinSketchWithHeap:: - /// update(key, value)` (`sketchlib_cms_heap_update` → `insert_many(key, - /// round(value))`), which is the "separate update path" the evaluation - /// plan (Fig 3c) called for. - pub fn insert_value(&mut self, group_label: &str, value: f64) { - self.inner.update(group_label, value); - } - - /// Read the top-`k` GROUPS ranked by summed VALUE (descending), keyed by - /// the group label. Pairs with [`Self::insert_value`]: the heap built by - /// value-weighted updates ranks by Σvalue, so this returns the - /// value-weighted top-k (not the occurrence-count top-k the raw `item` - /// heap would give). Sorted descending by value; ties broken by key for - /// determinism; truncated to `k`. - pub fn topk_by_value(&self, k: usize) -> Vec<(String, f64)> { - let mut items: Vec<(String, f64)> = self - .inner - .topk_heap_items() - .into_iter() - .map(|it| (it.key, it.value)) - .collect(); - items.sort_by(|a, b| { - b.1.partial_cmp(&a.1) - .unwrap_or(std::cmp::Ordering::Equal) - .then_with(|| a.0.cmp(&b.0)) - }); - items.truncate(k); - items - } - - /// Get all keys from the top-k heap. - pub fn get_topk_keys(&self) -> Vec { - self.inner - .topk_heap_items() - .iter() - .map(|item| { - let labels: Vec = item.key.split(';').map(|s| s.to_string()).collect(); - KeyByLabelValues { labels } - }) - .collect() - } -} - -impl AggregateCore for CountMinSketchWithHeapAccumulator { - 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_cms = other - .as_any() - .downcast_ref::() - .ok_or("Failed to downcast to CountMinSketchWithHeapAccumulator")?; - - let mut merged = self.clone(); - merged.inner.merge(&other_cms.inner)?; - Ok(Box::new(merged)) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_count_min_sketch_with_heap_creation() { - let cms = CountMinSketchWithHeapAccumulator::new(4, 1000, 20); - assert_eq!(cms.inner.rows(), 4); - assert_eq!(cms.inner.cols(), 1000); - assert_eq!(cms.inner.heap_size, 20); - assert_eq!(cms.inner.topk_heap_items().len(), 0); - } - - #[test] - fn test_get_topk_keys() { - let mut cms = CountMinSketchWithHeapAccumulator::new(2, 3, 5); - cms.inner.update("label1;label2", 100.0); - cms.inner.update("label3;label4", 50.0); - - let keys = cms.get_topk_keys(); - assert_eq!(keys.len(), 2); - // Heap order is not part of the contract; compare as a set. - let label_sets: std::collections::HashSet<_> = - keys.iter().map(|k| k.labels.clone()).collect(); - assert!(label_sets.contains(&vec!["label1".to_string(), "label2".to_string()])); - assert!(label_sets.contains(&vec!["label3".to_string(), "label4".to_string()])); - } - - // ---------------------------------------------------------------- - // FIX 1 — VALUE-WEIGHTED top-k (recall 0 → correct). - // - // `topk(k, sum by (host) (cpu_load))` asks for the top-k hosts by - // SUM OF VALUE. The heavy-hitter heap built by the default `+1`-per- - // occurrence update ranks by COUNT keyed by `item`, so its recall - // against the value-weighted ground truth is 0 when the busiest host - // (most samples) is NOT the heaviest host (largest Σvalue). - // `insert_value(group_label, value)` adds the sample VALUE keyed by the - // GROUP LABEL, so `topk_by_value` ranks by Σvalue — correct recall. - // ---------------------------------------------------------------- - - /// Crafted adversarial dataset: the host with the MOST samples - /// (`h_chatty`, 100 tiny samples) is NOT the host with the largest - /// value-sum (`h_heavy`, a handful of huge samples). A COUNT-ranked - /// heap would surface `h_chatty`; the value-weighted top-k must surface - /// the true heavy hitters by Σvalue, giving recall 1.0 against the - /// ground-truth top-k-by-value-sum. - #[test] - fn value_weighted_topk_has_full_recall_vs_count_topk() { - // (host, per-sample value, sample count) → true Σvalue: - // h_heavy : 1000 × 3 = 3000 (few samples, huge value) - // h_mid : 200 × 5 = 1000 - // h_small : 50 × 6 = 300 - // h_chatty: 1 × 100 = 100 (MOST samples, tiny value) - let data: &[(&str, f64, usize)] = &[ - ("h_heavy", 1000.0, 3), - ("h_mid", 200.0, 5), - ("h_small", 50.0, 6), - ("h_chatty", 1.0, 100), - ]; - - // Wide CMS + heap large enough to hold every group exactly (4 groups) - // so the estimate equals the true Σvalue with no hash collisions. - let mut acc = CountMinSketchWithHeapAccumulator::new(5, 4096, 16); - let mut truth: std::collections::HashMap<&str, f64> = std::collections::HashMap::new(); - for (host, value, count) in data { - for _ in 0..*count { - acc.insert_value(host, *value); - } - *truth.entry(*host).or_insert(0.0) += value * (*count as f64); - } - - // Ground-truth top-2 by value-sum: h_heavy (3000), h_mid (1000). - let mut truth_ranked: Vec<(&str, f64)> = truth.into_iter().collect(); - truth_ranked.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap()); - let truth_top2: std::collections::HashSet<&str> = - truth_ranked.iter().take(2).map(|(k, _)| *k).collect(); - assert!( - truth_top2.contains("h_heavy") && truth_top2.contains("h_mid"), - "ground-truth top-2 by value-sum should be h_heavy + h_mid" - ); - - // Value-weighted top-2 from the heap. - let got = acc.topk_by_value(2); - assert_eq!(got.len(), 2, "k=2 → two groups: {got:?}"); - let got_keys: std::collections::HashSet<&str> = - got.iter().map(|(k, _)| k.as_str()).collect(); - - // RECALL = |got ∩ truth| / |truth| must be 1.0. - let hits = got_keys.intersection(&truth_top2).count(); - let recall = hits as f64 / truth_top2.len() as f64; - assert_eq!( - recall, 1.0, - "value-weighted top-k recall must be 1.0 (count-ranked heap would \ - surface h_chatty and miss h_heavy → recall < 1): got={got:?}" - ); - - // The busiest-by-count host (h_chatty) must NOT be in the top-2, - // proving we rank by value-sum, not occurrence count. - assert!( - !got_keys.contains("h_chatty"), - "h_chatty (most samples, smallest value-sum) must be excluded: {got:?}" - ); - - // Estimates are exact here (no collisions, heap holds all groups): - // top-1 must be h_heavy with Σvalue 3000. - assert_eq!(got[0].0, "h_heavy"); - assert!( - (got[0].1 - 3000.0).abs() < 1e-6, - "h_heavy value-sum estimate ≈ 3000, got {}", - got[0].1 - ); - assert_eq!(got[1].0, "h_mid"); - assert!( - (got[1].1 - 1000.0).abs() < 1e-6, - "h_mid value-sum estimate ≈ 1000, got {}", - got[1].1 - ); - } - - /// A single value-weighted insert must put the full value (not +1) into - /// the heap, and repeated inserts for the same group must accumulate. - #[test] - fn insert_value_accumulates_summed_value_in_heap() { - let mut acc = CountMinSketchWithHeapAccumulator::new(4, 1024, 8); - acc.insert_value("g", 10.0); - acc.insert_value("g", 25.0); - let top = acc.topk_by_value(1); - assert_eq!(top.len(), 1); - assert_eq!(top[0].0, "g"); - assert!( - (top[0].1 - 35.0).abs() < 1e-6, - "summed value should be 35 (10+25), got {}", - top[0].1 - ); - } -} diff --git a/crates/asap-physical-operators/src/summary_kernels/count_sketch_with_heap.rs b/crates/asap-physical-operators/src/summary_kernels/count_sketch_with_heap.rs deleted file mode 100644 index 9f15e5042..000000000 --- a/crates/asap-physical-operators/src/summary_kernels/count_sketch_with_heap.rs +++ /dev/null @@ -1,146 +0,0 @@ -//! CountSketch with a top-k heap over `asap_sketchlib::CountSketchWithHeap` -//! (median-of-signed-rows), distinct from the Count-Min heap variant. - -use crate::{AggregateCore, KeyByLabelValues}; -use asap_sketchlib::CountSketchWithHeap; - -#[derive(Debug, Clone)] -pub struct CountSketchWithHeapAccumulator { - pub inner: CountSketchWithHeap, -} - -impl CountSketchWithHeapAccumulator { - pub fn new(row_num: usize, col_num: usize, heap_size: usize) -> Self { - Self { - inner: CountSketchWithHeap::new(row_num, col_num, heap_size), - } - } - - pub fn query_key(&self, key: &KeyByLabelValues) -> f64 { - let key_string = key.labels.join(";"); - self.inner.estimate(&key_string) - } - - /// Value-weighted heavy-hitter update -- see - /// `CountMinSketchWithHeapAccumulator::insert_value`'s doc for why - /// this (not a `+1`-per-occurrence update) is the correct semantics - /// for `topk(k, sum by (label) (metric))`-shaped queries. - pub fn insert_value(&mut self, group_label: &str, value: f64) { - self.inner.update(group_label, value); - } - - /// Read the top-`k` groups ranked by summed value (descending, tie-broken - /// by key for determinism). Mirrors `CountMinSketchWithHeapAccumulator::topk_by_value`. - pub fn topk_by_value(&self, k: usize) -> Vec<(String, f64)> { - let mut items: Vec<(String, f64)> = self - .inner - .topk_heap_items() - .into_iter() - .map(|it| (it.key, it.value)) - .collect(); - items.sort_by(|a, b| { - b.1.partial_cmp(&a.1) - .unwrap_or(std::cmp::Ordering::Equal) - .then_with(|| a.0.cmp(&b.0)) - }); - items.truncate(k); - items - } - - /// Get all keys from the top-k heap. - pub fn get_topk_keys(&self) -> Vec { - self.inner - .topk_heap_items() - .iter() - .map(|item| { - let labels: Vec = item.key.split(';').map(|s| s.to_string()).collect(); - KeyByLabelValues { labels } - }) - .collect() - } -} - -impl AggregateCore for CountSketchWithHeapAccumulator { - 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 CountSketchWithHeapAccumulator")?; - - let mut merged = self.clone(); - merged.inner.merge(&other_cs.inner)?; - Ok(Box::new(merged)) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_count_sketch_with_heap_creation() { - let cs = CountSketchWithHeapAccumulator::new(4, 1000, 20); - assert_eq!(cs.inner.rows(), 4); - assert_eq!(cs.inner.cols(), 1000); - assert_eq!(cs.inner.heap_size, 20); - assert_eq!(cs.inner.topk_heap_items().len(), 0); - } - - #[test] - fn test_get_topk_keys() { - let mut cs = CountSketchWithHeapAccumulator::new(2, 3, 5); - cs.inner.update("label1;label2", 100.0); - cs.inner.update("label3;label4", 50.0); - - let keys = cs.get_topk_keys(); - assert_eq!(keys.len(), 2); - let label_sets: std::collections::HashSet<_> = - keys.iter().map(|k| k.labels.clone()).collect(); - assert!(label_sets.contains(&vec!["label1".to_string(), "label2".to_string()])); - assert!(label_sets.contains(&vec!["label3".to_string(), "label4".to_string()])); - } - - #[test] - fn insert_value_accumulates_summed_value_in_heap() { - let mut acc = CountSketchWithHeapAccumulator::new(4, 1024, 8); - acc.insert_value("g", 10.0); - acc.insert_value("g", 25.0); - let top = acc.topk_by_value(1); - assert_eq!(top.len(), 1); - assert_eq!(top[0].0, "g"); - assert!( - (top[0].1 - 35.0).abs() < 1e-6, - "summed value should be 35 (10+25), got {}", - top[0].1 - ); - } - - /// CountSketch and Count-Min heap states are distinct families and never merge. - #[test] - fn test_rejects_merge_with_cms_family_accumulator() { - use crate::summary_kernels::count_min_sketch_with_heap::CountMinSketchWithHeapAccumulator; - - let cs = CountSketchWithHeapAccumulator::new(4, 64, 10); - let cms = CountMinSketchWithHeapAccumulator::new(4, 64, 10); - let result = cs.merge_with(&cms); - assert!( - result.is_err(), - "CountSketchWithHeapAccumulator must not merge with CountMinSketchWithHeapAccumulator \ - -- different algorithms sharing only a storage shape" - ); - } -} diff --git a/crates/asap-physical-operators/src/summary_kernels/datasketches_kll.rs b/crates/asap-physical-operators/src/summary_kernels/datasketches_kll.rs index 14ed5a9ad..bc60f8586 100644 --- a/crates/asap-physical-operators/src/summary_kernels/datasketches_kll.rs +++ b/crates/asap-physical-operators/src/summary_kernels/datasketches_kll.rs @@ -33,11 +33,6 @@ impl std::fmt::Debug for DatasketchesKLLAccumulator { } } -// SAFETY: `KllSketch` owns its buffers and has no interior mutability; the -// accumulator is only mutated through `&mut self`. -unsafe impl Send for DatasketchesKLLAccumulator {} -unsafe impl Sync for DatasketchesKLLAccumulator {} - impl AggregateCore for DatasketchesKLLAccumulator { fn clone_boxed_core(&self) -> Box { Box::new(self.clone()) diff --git a/crates/asap-physical-operators/src/summary_kernels/factory.rs b/crates/asap-physical-operators/src/summary_kernels/factory.rs index 2f6422c6f..2827a777e 100644 --- a/crates/asap-physical-operators/src/summary_kernels/factory.rs +++ b/crates/asap-physical-operators/src/summary_kernels/factory.rs @@ -2,8 +2,8 @@ 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, CountSketchWithHeapAccumulator, - DDSketchAccumulator, DatasketchesKLLAccumulator, HydraCms, HydraKllSketchAccumulator, + CountMinSketchAccumulator, DDSketchAccumulator, DatasketchesKLLAccumulator, HydraCms, + HydraKllSketchAccumulator, }; use crate::{AggregateCore, KeyByLabelValues}; use planner_types::post_asap::{FieldDataType as SummaryFamilyType, SketchAlgorithm, SketchParams}; @@ -240,104 +240,6 @@ impl AccumulatorUpdater for CmsAccumulatorUpdater { } } -// --------------------------------------------------------------------------- -// CmsHeapAccumulatorUpdater — value-weighted / count-weighted top-k -// --------------------------------------------------------------------------- - -/// What quantity the top-k heap ranks keys by. -/// -/// These are DIFFERENT query semantics and must be chosen explicitly: -/// -/// * [`TopkWeight::Value`] — accumulate **Σ of the datapoint value** per key. -/// This answers "top-k by total " (e.g. "top-k hosts by -/// total CPU"). The heap value is the summed metric value, so the read-side -/// reducer's "sort heap descending by value" yields the correct ranking. -/// -/// * [`TopkWeight::Count`] — accumulate **+1 per event** per key (occurrence -/// frequency), the textbook heavy-hitter / frequency-top-k semantics -/// ("which keys appear most often"). -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum TopkWeight { - /// Σ datapoint value per key (value-weighted top-k). - Value, - /// +1 per event per key (count-weighted / frequency top-k). - Count, -} - -/// Keyed top-k updater backed by a real `CountMinSketchWithHeap` (a CMS -/// matrix PLUS a size-`heap_size` top-k heap). Unlike the heap-LESS -/// `CmsAccumulatorUpdater`, this enumerates top-k keys at read time -/// (`get_topk_keys` / `topk_heap_items`), which is what `topk(...)` queries -/// need. -/// -/// The key is the frequency item supplied by the operator (e.g. a `host` -/// value), not a group-by population. The accumulated quantity is selected -/// by [`TopkWeight`]: -/// * `Value` → `inner.update(key, value)` adds the datapoint value (Σ value). -/// * `Count` → `inner.update(key, 1.0)` adds one per event (Σ count). -/// -/// `create_planner_accumulator` builds Planner `CmsWithHeap` and -/// `CountSketchWithHeap` families with [`WeightedFrequencyUpdater`] instead. -pub struct CmsHeapAccumulatorUpdater { - acc: CountMinSketchWithHeapAccumulator, - row_num: usize, - col_num: usize, - heap_size: usize, - weight: TopkWeight, -} - -impl CmsHeapAccumulatorUpdater { - pub fn new(row_num: usize, col_num: usize, heap_size: usize, weight: TopkWeight) -> Self { - Self { - acc: CountMinSketchWithHeapAccumulator::new(row_num, col_num, heap_size), - row_num, - col_num, - heap_size, - weight, - } - } -} - -impl AccumulatorUpdater for CmsHeapAccumulatorUpdater { - fn update_single(&mut self, _value: f64, _timestamp_ms: i64) { - debug_assert!( - false, - "update_single called on keyed updater; use update_keyed" - ); - } - - fn update_keyed(&mut self, key: &KeyByLabelValues, value: f64, _timestamp_ms: i64) { - // Heap key = the group-by label-value vector (e.g. `host`), joined the - // same way the read-side `get_topk_keys` splits it back apart (`;`). - let weighted = match self.weight { - // Σ value: feed the datapoint value. sketchlib's CMS-heap - // `update(key, w)` adds `w.round()` occurrences of `key`, so the - // heap value accumulates the (rounded) summed metric value. - TopkWeight::Value => value, - // Σ count: one occurrence per event, regardless of value. - TopkWeight::Count => 1.0, - }; - self.acc.inner.update(&key.to_semicolon_str(), weighted); - } - - impl_clone_accumulator_methods!(acc); - - fn reset(&mut self) { - self.acc = - CountMinSketchWithHeapAccumulator::new(self.row_num, self.col_num, self.heap_size); - } - - fn is_keyed(&self) -> bool { - true - } - - fn memory_usage_bytes(&self) -> usize { - std::mem::size_of::() - + self.row_num * self.col_num * std::mem::size_of::() - + self.heap_size * (std::mem::size_of::() + 32) - } -} - // --------------------------------------------------------------------------- // WeightedFrequencyUpdater (CountSketch, Count-Min with heap, CountSketch with heap) // --------------------------------------------------------------------------- @@ -354,15 +256,11 @@ impl AccumulatorUpdater for CmsHeapAccumulatorUpdater { /// 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). + /// Fails for a shape `asap_sketchlib` refuses: a zero dimension, or an + /// even CountSketch `depth` (the median needs an odd row count). pub fn new( algorithm: FrequencyAlgorithm, width: usize, @@ -371,23 +269,16 @@ impl WeightedFrequencyUpdater { ) -> Result { Ok(Self { acc: WeightedFrequency::new(algorithm, width, depth, capacity)?, - algorithm, - width, - depth, - capacity, }) } } impl AccumulatorUpdater for WeightedFrequencyUpdater { + /// The kernel's own weight rule: finite, and nonnegative for Count-Min. 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(()) - } + self.acc + .check_weight(value) + .map_err(|error| error.to_string()) } fn update_single(&mut self, _value: f64, _timestamp_ms: i64) { @@ -398,23 +289,17 @@ impl AccumulatorUpdater for WeightedFrequencyUpdater { } fn update_keyed(&mut self, key: &KeyByLabelValues, value: f64, _timestamp_ms: i64) { - 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) + .update_labels(&key.labels, value) .expect("weighted frequency counter overflow"); } impl_clone_accumulator_methods!(acc); fn reset(&mut self) { - self.acc = WeightedFrequency::new(self.algorithm, self.width, self.depth, self.capacity) - .expect("dimensions validated at construction"); + self.acc.clear(); } fn is_keyed(&self) -> bool { @@ -426,70 +311,6 @@ impl AccumulatorUpdater for WeightedFrequencyUpdater { } } -// --------------------------------------------------------------------------- -// 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 -/// [`TopkWeight`] semantics and heap payload shape. -pub struct CountSketchWithHeapAccumulatorUpdater { - acc: CountSketchWithHeapAccumulator, - row_num: usize, - col_num: usize, - heap_size: usize, - weight: TopkWeight, -} - -impl CountSketchWithHeapAccumulatorUpdater { - pub fn new(row_num: usize, col_num: usize, heap_size: usize, weight: TopkWeight) -> Self { - Self { - acc: CountSketchWithHeapAccumulator::new(row_num, col_num, heap_size), - row_num, - col_num, - heap_size, - weight, - } - } -} - -impl AccumulatorUpdater for CountSketchWithHeapAccumulatorUpdater { - fn update_single(&mut self, _value: f64, _timestamp_ms: i64) { - debug_assert!( - false, - "update_single called on keyed updater; use update_keyed" - ); - } - - fn update_keyed(&mut self, key: &KeyByLabelValues, value: f64, _timestamp_ms: i64) { - let weighted = match self.weight { - TopkWeight::Value => value, - TopkWeight::Count => 1.0, - }; - self.acc.inner.update(&key.to_semicolon_str(), weighted); - } - - impl_clone_accumulator_methods!(acc); - - fn reset(&mut self) { - self.acc = CountSketchWithHeapAccumulator::new(self.row_num, self.col_num, self.heap_size); - } - - fn is_keyed(&self) -> bool { - true - } - - fn memory_usage_bytes(&self) -> usize { - std::mem::size_of::() - + self.row_num * self.col_num * std::mem::size_of::() - + self.heap_size * (std::mem::size_of::() + 32) - } -} - // --------------------------------------------------------------------------- // HydraKllAccumulatorUpdater // --------------------------------------------------------------------------- diff --git a/crates/asap-physical-operators/src/summary_kernels/kmv.rs b/crates/asap-physical-operators/src/summary_kernels/kmv.rs index e6ae02e5d..a1153b9de 100644 --- a/crates/asap-physical-operators/src/summary_kernels/kmv.rs +++ b/crates/asap-physical-operators/src/summary_kernels/kmv.rs @@ -58,25 +58,22 @@ impl AggregateCore for KmvAccumulator { .into()); } let mut merged = self.clone(); - merged.inner.merge(&mut other.inner.clone()); + merged.inner.merge(&other.inner); Ok(Box::new(merged)) } /// Distinct count. A bare `PointCount` over a KMV also reads the distinct count. fn estimate(&self, query: &SketchStatistic) -> Result { match query { - // `KMV::estimate` takes `&mut self` without mutating. SketchStatistic::Cardinality | SketchStatistic::PointCount { value: None, .. } => { - Ok(self.inner.clone().estimate()) + Ok(self.inner.estimate()) } other => Err(format!("KMV does not answer {other:?}").into()), } } fn approx_memory_bytes(&self) -> usize { - // sketchlib reserves all k hash slots at construction. - std::mem::size_of::() - .saturating_add(self.inner.k.saturating_mul(std::mem::size_of::())) + self.inner.approx_memory_bytes() } } @@ -212,16 +209,14 @@ mod tests { assert_eq!(cardinality(&kmv), 0.0); } - // sketchlib reserves all k hash slots at construction, so an empty state - // already holds k * 8 bytes. + // The footprint is the library's: it covers every retained hash and + // grows as the state fills toward k. #[test] - fn memory_counts_the_k_reserved_hashes() { + fn memory_covers_the_retained_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() - ); + let (empty, full) = (KmvAccumulator::new(k), build(k, 0..20_000)); + assert_eq!(full.inner.k_vals.len(), k as usize); + assert!(full.approx_memory_bytes() >= k as usize * 8); + assert!(full.approx_memory_bytes() > empty.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 5efef8913..919d1e768 100644 --- a/crates/asap-physical-operators/src/summary_kernels/mod.rs +++ b/crates/asap-physical-operators/src/summary_kernels/mod.rs @@ -1,7 +1,5 @@ //! 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_with_heap; pub mod datasketches_kll; pub mod dd_sketch; pub mod exact; @@ -13,8 +11,6 @@ pub mod kmv; pub mod univmon; pub use count_min_sketch::*; -pub use count_min_sketch_with_heap::*; -pub use count_sketch_with_heap::*; pub use datasketches_kll::*; pub use dd_sketch::*; pub use hll_sketch::*; diff --git a/crates/asap-physical-operators/src/summary_kernels/univmon.rs b/crates/asap-physical-operators/src/summary_kernels/univmon.rs index bd8db2027..c647be0a1 100644 --- a/crates/asap-physical-operators/src/summary_kernels/univmon.rs +++ b/crates/asap-physical-operators/src/summary_kernels/univmon.rs @@ -1,42 +1,11 @@ //! One frequency state shared by count, distinct, L2 and entropy evaluations. use crate::AggregateCore; -use asap_sketchlib::sketch_framework::univmon::MAX_LAYER_SIZE; use asap_sketchlib::{DataInput, UnivMon}; use planner_types::{post_asap::SketchStatistic, pre_asap::ColumnRef}; type Error = Box; -/// Rows beyond the executor's own limit, and shapes `asap_sketchlib` -/// rejects because their hashes would not be independent: more than -/// [`MAX_LAYER_SIZE`] (19) layers, or a layer whose rows read more than -/// one 128-bit hash holds. Each row reads `⌈log2 cols⌉` column bits and one -/// sign bit, so `rows · (⌈log2 cols⌉ + 1) ≤ 128`. -pub(crate) fn check_dimensions( - heap_size: usize, - rows: usize, - cols: usize, - layers: usize, -) -> Result<(), Error> { - if heap_size == 0 - || cols == 0 - || !(1..=20).contains(&rows) - || !(1..=MAX_LAYER_SIZE).contains(&layers) - { - return Err("invalid UnivMon dimensions".into()); - } - let column_bits = cols - .checked_next_power_of_two() - .map_or(usize::BITS, usize::trailing_zeros) as usize; - if rows * (column_bits + 1) > 128 { - return Err("UnivMon rows and columns exceed one 128-bit hash".into()); - } - rows.checked_mul(cols) - .and_then(|n| n.checked_mul(layers)) - .ok_or("UnivMon dimensions overflow")?; - Ok(()) -} - #[derive(Debug, Clone)] pub struct UnivMonAccumulator { inner: UnivMon, @@ -60,7 +29,7 @@ impl UnivMonAccumulator { } pub fn new(heap_size: usize, rows: usize, cols: usize, layers: usize) -> Result { - check_dimensions(heap_size, rows, cols, layers)?; + UnivMon::check_shape(heap_size, rows, cols, layers)?; Ok(Self { inner: UnivMon::init_univmon(heap_size, rows, cols, layers), max_key_bytes: 0, @@ -72,7 +41,7 @@ impl UnivMonAccumulator { /// bytes. Rejects dimensions [`Self::new`] rejects and terminal-mode /// sketches, which do not accept this accumulator's updates. pub fn from_sketch(sketch: UnivMon) -> Result { - check_dimensions( + UnivMon::check_shape( sketch.heap_size, sketch.sketch_row, sketch.sketch_col, @@ -299,7 +268,10 @@ mod tests { assert!(UnivMonAccumulator::from_sketch(terminal.clone()).is_err()); terminal.free(); assert!(UnivMonAccumulator::from_sketch(terminal).is_ok()); - assert!(UnivMonAccumulator::from_sketch(UnivMon::init_univmon(4, 21, 16, 2)).is_err()); + // `init_univmon` refuses 21 rows; a state claiming them is refused too. + let mut misshapen = UnivMon::init_univmon(4, 3, 16, 2); + misshapen.sketch_row = 21; + assert!(UnivMonAccumulator::from_sketch(misshapen).is_err()); } // Shapes whose hashes are not independent are errors, not library 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 39b7be953..b6134db31 100644 --- a/crates/asap-physical-operators/src/summary_kernels/weighted_frequency.rs +++ b/crates/asap-physical-operators/src/summary_kernels/weighted_frequency.rs @@ -64,19 +64,22 @@ impl WeightedFrequency { depth, heap_size, }, - ) if depth % 2 == 1 => (FrequencyAlgorithm::CountSketch, *width, *depth, *heap_size), + ) => (FrequencyAlgorithm::CountSketch, *width, *depth, *heap_size), _ => { return Err(Error::Invalid( - "unsupported weighted frequency family or depth".into(), + "unsupported weighted frequency family".into(), )) } }; - if width == 0 || depth == 0 || capacity == 0 { + // A heap family needs a heap; the shape rules are the kernel's. + if capacity == 0 { return Err(Error::Invalid( - "invalid weighted frequency dimensions".into(), + "weighted frequency heap capacity is 0".into(), )); } - Ok((algorithm, width as usize, depth as usize, capacity as usize)) + let (width, depth) = (width as usize, depth as usize); + Kernel::check_shape(algorithm, width, depth).map_err(adapt_error)?; + Ok((algorithm, width, depth, capacity as usize)) } /// Algorithm and `(width, depth, capacity)` shape, so a deployment can check @@ -97,6 +100,23 @@ impl WeightedFrequency { .map(|inner| Self { inner }) .map_err(adapt_error) } + /// Adds `weight` to the identity tuple of `labels`, one `Utf8` identity + /// per label, as the DAG's keyed build keys a label vector. + pub(crate) fn update_labels(&mut self, labels: &[String], weight: f64) -> Result<(), Error> { + let identity = labels + .iter() + .map(|label| FrequencyIdentity::Utf8(label.clone())) + .collect::>(); + self.inner.update(&identity, weight).map_err(adapt_error) + } + /// The kernel's weight rule: finite, and nonnegative for Count-Min. + pub(crate) fn check_weight(&self, weight: f64) -> Result<(), WeightedFrequencyError> { + self.inner.check_weight(weight) + } + /// Empties the state in place, keeping its algorithm and shape. + pub(crate) fn clear(&mut self) { + self.inner.clear(); + } pub fn update(&mut self, values: &[Value], weight: f64) -> Result<(), Error> { let values = values.iter().map(identity).collect::, _>>()?; self.inner.update(&values, weight).map_err(adapt_error)