diff --git a/Cargo.lock b/Cargo.lock index 640ce0e23..cab4d40d0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -485,20 +485,15 @@ dependencies = [ [[package]] name = "asap_sketchlib" -version = "0.3.0" -source = "git+https://github.com/ProjectASAP/asap_sketchlib?rev=3f44cf1c2d823b844c17796f24408e03863149ee#3f44cf1c2d823b844c17796f24408e03863149ee" +version = "0.4.0" +source = "git+https://github.com/ProjectASAP/asap_sketchlib?rev=eb04e7f243c1d5e5c346cd19842f2402d796b620#eb04e7f243c1d5e5c346cd19842f2402d796b620" dependencies = [ - "bincode", - "bytes", - "prost", "rand 0.9.5", "rmp-serde", "serde", - "serde-big-array", "serde_bytes", "smallvec", "twox-hash", - "xxhash-rust", ] [[package]] @@ -922,7 +917,7 @@ dependencies = [ "flate2", "futures", "indexmap", - "itertools 0.14.0", + "itertools", "liblzma", "log", "object_store", @@ -954,7 +949,7 @@ dependencies = [ "datafusion-physical-plan", "datafusion-session", "futures", - "itertools 0.14.0", + "itertools", "log", "object_store", "parking_lot", @@ -979,7 +974,7 @@ dependencies = [ "datafusion-physical-expr-common", "datafusion-physical-plan", "futures", - "itertools 0.14.0", + "itertools", "log", "object_store", ] @@ -998,7 +993,7 @@ dependencies = [ "half", "hashbrown 0.17.1", "indexmap", - "itertools 0.14.0", + "itertools", "libc", "log", "object_store", @@ -1045,7 +1040,7 @@ dependencies = [ "flate2", "futures", "glob", - "itertools 0.14.0", + "itertools", "liblzma", "log", "object_store", @@ -1076,7 +1071,7 @@ dependencies = [ "datafusion-physical-plan", "datafusion-session", "futures", - "itertools 0.14.0", + "itertools", "object_store", "tokio", ] @@ -1150,7 +1145,7 @@ dependencies = [ "datafusion-pruning", "datafusion-session", "futures", - "itertools 0.14.0", + "itertools", "log", "object_store", "parking_lot", @@ -1203,7 +1198,7 @@ dependencies = [ "datafusion-functions-window-common", "datafusion-physical-expr-common", "indexmap", - "itertools 0.14.0", + "itertools", "recursive", "serde_json", "sqlparser", @@ -1218,7 +1213,7 @@ dependencies = [ "arrow", "datafusion-common", "indexmap", - "itertools 0.14.0", + "itertools", ] [[package]] @@ -1242,7 +1237,7 @@ dependencies = [ "datafusion-macros", "datafusion-physical-expr-common", "hex", - "itertools 0.14.0", + "itertools", "log", "md-5", "memchr", @@ -1305,7 +1300,7 @@ dependencies = [ "datafusion-macros", "datafusion-physical-expr-common", "hashbrown 0.17.1", - "itertools 0.14.0", + "itertools", "itoa", "log", "memchr", @@ -1378,7 +1373,7 @@ dependencies = [ "datafusion-expr-common", "datafusion-physical-expr", "indexmap", - "itertools 0.14.0", + "itertools", "log", "recursive", "regex", @@ -1400,7 +1395,7 @@ dependencies = [ "half", "hashbrown 0.17.1", "indexmap", - "itertools 0.14.0", + "itertools", "parking_lot", "petgraph", "recursive", @@ -1419,7 +1414,7 @@ dependencies = [ "datafusion-functions", "datafusion-physical-expr", "datafusion-physical-expr-common", - "itertools 0.14.0", + "itertools", ] [[package]] @@ -1434,7 +1429,7 @@ dependencies = [ "datafusion-expr-common", "hashbrown 0.17.1", "indexmap", - "itertools 0.14.0", + "itertools", "parking_lot", "pin-project", ] @@ -1454,7 +1449,7 @@ dependencies = [ "datafusion-physical-expr-common", "datafusion-physical-plan", "datafusion-pruning", - "itertools 0.14.0", + "itertools", "recursive", ] @@ -1483,7 +1478,7 @@ dependencies = [ "half", "hashbrown 0.17.1", "indexmap", - "itertools 0.14.0", + "itertools", "log", "num-traits", "parking_lot", @@ -2064,15 +2059,6 @@ version = "3.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8bb03732005da905c88227371639bf1ad885cc712789c011c31c5fb3ab3ccf02" -[[package]] -name = "itertools" -version = "0.13.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "413ee7dfc52ee1a4949ceeb7dbc8a33f2d6c088194d9f922fb8318faf1f01186" -dependencies = [ - "either", -] - [[package]] name = "itertools" version = "0.14.0" @@ -2471,7 +2457,7 @@ dependencies = [ "futures-util", "http", "humantime", - "itertools 0.14.0", + "itertools", "parking_lot", "percent-encoding", "thiserror 2.0.18", @@ -2739,29 +2725,6 @@ dependencies = [ "regex", ] -[[package]] -name = "prost" -version = "0.13.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2796faa41db3ec313a31f7624d9286acf277b52de526150b7e69f3debf891ee5" -dependencies = [ - "bytes", - "prost-derive", -] - -[[package]] -name = "prost-derive" -version = "0.13.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d" -dependencies = [ - "anyhow", - "itertools 0.13.0", - "proc-macro2", - "quote", - "syn", -] - [[package]] name = "psm" version = "0.1.32" @@ -2985,15 +2948,6 @@ dependencies = [ "serde_derive", ] -[[package]] -name = "serde-big-array" -version = "0.5.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "11fc7cc2c76d73e0f27ee52abbd64eec84d46f370c88371120433196934e4b7f" -dependencies = [ - "serde", -] - [[package]] name = "serde_bytes" version = "0.11.19" diff --git a/Cargo.toml b/Cargo.toml index ae333ca2b..003cf11e1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -23,4 +23,4 @@ resolver = "2" arrow = { version = "58.4", default-features = false } # The hasher Arrow already uses, for the executor's hot interning maps. ahash = { version = "0.8", default-features = false, features = ["std", "runtime-rng"] } -asap_sketchlib = { git = "https://github.com/ProjectASAP/asap_sketchlib", rev = "3f44cf1c2d823b844c17796f24408e03863149ee" } +asap_sketchlib = { git = "https://github.com/ProjectASAP/asap_sketchlib", rev = "eb04e7f243c1d5e5c346cd19842f2402d796b620" } diff --git a/crates/executor/src/capability.rs b/crates/executor/src/capability.rs index b0925666c..b97d14e1c 100644 --- a/crates/executor/src/capability.rs +++ b/crates/executor/src/capability.rs @@ -8,6 +8,8 @@ //! //! Stored-state encodings belong to deployments. Full plan acceptance is //! owned by `binding`, which also validates schemas, expressions and inputs. +use crate::summary_kernels::hll_sketch::HLL_PRECISIONS; +use crate::summary_kernels::weighted_frequency::{self, FrequencyAlgorithm}; use crate::Error; use planner_types::ir::schema::{ ExactKind, ExactParams, FieldDataType as SummaryFamilyType, GroupingStrategy, SketchAlgorithm, @@ -59,14 +61,14 @@ pub fn validate_summary_kernel( { false } - (A::Hll, P::Hll { precision, .. }) if (4..=18).contains(precision) => false, + (A::Hll, P::Hll { precision, .. }) if HLL_PRECISIONS.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 kernel's shape rules (an odd depth, for the median). (A::CountSketch, P::CountSketch { width, depth }) if valid_matrix(*width, *depth) - && asap_sketchlib::WeightedFrequency::check_shape( - asap_sketchlib::FrequencyAlgorithm::CountSketch, + && weighted_frequency::check_shape( + FrequencyAlgorithm::CountSketch, *width as usize, *depth as usize, ) @@ -99,7 +101,7 @@ pub fn validate_summary_kernel( layers, .. }, - ) if asap_sketchlib::UnivMon::check_shape( + ) if crate::summary_kernels::univmon::check_shape( *heap_size as usize, *sketch_rows as usize, *sketch_cols as usize, @@ -306,8 +308,7 @@ pub fn validate_native_family(family: &SummaryFamilyType) -> Result<(), Error> { }; } if matches!(kind.algorithm(), A::CmsWithHeap | A::CountSketchWithHeap) { - let (_, width, depth, _) = - crate::summary_kernels::weighted_frequency::WeightedFrequency::configuration(kind)?; + let (_, width, depth, _) = weighted_frequency::WeightedFrequency::configuration(kind)?; return if valid_matrix(width as u32, depth as u32) && grouping == &Default::default() { Ok(()) } else { diff --git a/crates/executor/src/operators/row_reference/mod.rs b/crates/executor/src/operators/row_reference/mod.rs index 80c547970..12b9c6469 100644 --- a/crates/executor/src/operators/row_reference/mod.rs +++ b/crates/executor/src/operators/row_reference/mod.rs @@ -127,7 +127,11 @@ pub(crate) mod harness { ); } if let Some(frequency) = any.downcast_ref::() { - return serde_json::to_string(frequency).unwrap(); + return format!( + "{:?} {:?}", + frequency.counters(), + frequency.rows(usize::MAX) + ); } panic!("no fingerprint for this state kind") } diff --git a/crates/executor/src/summary_kernels/count_min_sketch.rs b/crates/executor/src/summary_kernels/count_min_sketch.rs index 20f49f092..873b3ab1c 100644 --- a/crates/executor/src/summary_kernels/count_min_sketch.rs +++ b/crates/executor/src/summary_kernels/count_min_sketch.rs @@ -1,23 +1,31 @@ -//! Count-Min Sketch frequency summary over `asap_sketchlib::CountMinSketch`. +//! Count-Min Sketch frequency summary over `asap_sketchlib::CountMinF64`. use crate::{AggregateCore, KernelError, KeyByLabelValues}; -use asap_sketchlib::CountMinSketch; +use asap_sketchlib::{CountMinF64, CountMinSketch, DataInput}; use planner_types::ir::schema::SketchStatistic; #[derive(Debug, Clone)] pub struct CountMinSketchAccumulator { - pub inner: CountMinSketch, + pub inner: CountMinF64, } impl CountMinSketchAccumulator { - pub fn new(row_num: usize, col_num: usize) -> Self { - Self { - inner: CountMinSketch::new(row_num, col_num), - } + /// Fails for a shape `asap_sketchlib` refuses (a zero dimension). + pub fn new(rows: usize, cols: usize) -> Result { + CountMinSketch::check_shape(rows, cols)?; + Ok(Self { + inner: CountMinF64::with_dimensions(rows, cols), + }) + } + + /// Adds `weight` to `item`, whatever its sign. + pub fn update(&mut self, item: &str, weight: f64) { + self.inner.update_weighted(&DataInput::Str(item), weight); } /// Estimated frequency of one item. pub fn query_key(&self, key: &KeyByLabelValues) -> f64 { - self.inner.estimate(&key.to_semicolon_str()) + self.inner + .estimate(&DataInput::Str(&key.to_semicolon_str())) } } @@ -38,29 +46,31 @@ impl AggregateCore for CountMinSketchAccumulator { .as_any() .downcast_ref::() .ok_or("Count-Min Sketch merges only with Count-Min Sketch")?; - Ok(Box::new(Self { - inner: CountMinSketch::merge_refs(&[&self.inner, &other.inner])?, - })) + // sketchlib's merge panics on a shape mismatch. + if (self.inner.rows(), self.inner.cols()) != (other.inner.rows(), other.inner.cols()) { + return Err("Count-Min Sketch merges only with the same shape".into()); + } + let mut merged = self.clone(); + merged.inner.merge(&other.inner); + Ok(Box::new(merged)) } /// A bare point count reads the total update weight: every Count-Min row /// receives each update exactly once, so one row's mass survives collisions. fn estimate(&self, query: &SketchStatistic) -> Result { match query { - SketchStatistic::PointCount { value: None, .. } => Ok(row_mass(&self.inner.sketch())), + SketchStatistic::PointCount { value: None, .. } => { + Ok(self.inner.as_storage().row_slice(0).iter().sum()) + } _ => Err(format!("{query:?} is not supported by Count-Min Sketch").into()), } } fn approx_memory_bytes(&self) -> usize { - 16 * 1024 + self.inner.memory_bytes() } } -pub(crate) fn row_mass(matrix: &[Vec]) -> f64 { - matrix.first().map_or(0.0, |row| row.iter().sum()) -} - #[cfg(test)] mod tests { use super::*; @@ -69,12 +79,12 @@ mod tests { #[test] fn merged_point_counts_add() { let (mut a, mut b) = ( - CountMinSketchAccumulator::new(3, 128), - CountMinSketchAccumulator::new(3, 128), + CountMinSketchAccumulator::new(3, 128).unwrap(), + CountMinSketchAccumulator::new(3, 128).unwrap(), ); let key = KeyByLabelValues::new_with_labels(vec!["checkout".into()]); - a.inner.update(&key.to_semicolon_str(), 2.0); - b.inner.update(&key.to_semicolon_str(), 3.0); + a.update(&key.to_semicolon_str(), 2.0); + b.update(&key.to_semicolon_str(), 3.0); let merged = a.merge_with(&b).unwrap(); let merged = merged .as_any() @@ -90,22 +100,45 @@ mod tests { key: planner_types::ir::scalar::ColumnRef::SampleValue, value: None, }; - let mut state = CountMinSketchAccumulator::new(2, 1); - state.inner.update("a", 3.0); - state.inner.update("b", 7.0); + let mut state = CountMinSketchAccumulator::new(2, 1).unwrap(); + state.update("a", 3.0); + state.update("b", 7.0); assert_eq!(state.estimate(&bare_count).unwrap(), 10.0); let merged = state.merge_with(&state).unwrap(); assert_eq!(merged.estimate(&bare_count).unwrap(), 20.0); - let empty = CountMinSketchAccumulator::new(2, 1); + let empty = CountMinSketchAccumulator::new(2, 1).unwrap(); assert_eq!(empty.estimate(&bare_count).unwrap(), 0.0); assert!(state.estimate(&SketchStatistic::Cardinality).is_err()); } + // Every weight reaches the counters: a zero adds nothing and a negative + // one subtracts. + #[test] + fn non_positive_weights_are_added() { + let mut state = CountMinSketchAccumulator::new(3, 128).unwrap(); + let key = KeyByLabelValues::new_with_labels(vec!["checkout".into()]); + state.update(&key.to_semicolon_str(), 5.0); + state.update(&key.to_semicolon_str(), 0.0); + state.update(&key.to_semicolon_str(), -2.0); + assert_eq!(state.query_key(&key), 3.0); + } + // Merge rejects a different summary family. #[test] fn rejects_foreign_merge() { - let cms = CountMinSketchAccumulator::new(3, 128); + let cms = CountMinSketchAccumulator::new(3, 128).unwrap(); let kll = crate::summary_kernels::DatasketchesKLLAccumulator::new(200); assert!(cms.merge_with(&kll).is_err()); } + + // States of different rows or columns do not merge. + #[test] + fn merge_rejects_a_different_shape() { + let cms = CountMinSketchAccumulator::new(3, 128).unwrap(); + for (rows, cols) in [(3, 64), (4, 128)] { + let other = CountMinSketchAccumulator::new(rows, cols).unwrap(); + assert!(cms.merge_with(&other).is_err()); + } + assert!(cms.merge_with(&cms).is_ok()); + } } diff --git a/crates/executor/src/summary_kernels/datasketches_kll.rs b/crates/executor/src/summary_kernels/datasketches_kll.rs index f1288e070..e5dcf0848 100644 --- a/crates/executor/src/summary_kernels/datasketches_kll.rs +++ b/crates/executor/src/summary_kernels/datasketches_kll.rs @@ -1,41 +1,37 @@ -//! KLL quantile summary over `asap_sketchlib::KllSketch`. +//! KLL quantile summary over `asap_sketchlib::Kll`. use crate::{AggregateCore, KernelError}; -use asap_sketchlib::KllSketch; +use asap_sketchlib::Kll; use planner_types::ir::schema::SketchStatistic; #[derive(Clone)] pub struct DatasketchesKLLAccumulator { - pub inner: KllSketch, + pub inner: Kll, } impl DatasketchesKLLAccumulator { pub fn new(k: u16) -> Self { Self { - inner: KllSketch::new(k), + inner: Kll::with_k(i32::from(k)), } } pub fn update(&mut self, value: f64) { - self.inner.update(value); + self.inner.update(&value); } + /// The value at `quantile`; 0 for an empty sketch. pub fn get_quantile(&self, quantile: f64) -> f64 { + if self.is_empty() { + return 0.0; + } self.inner.quantile(quantile) } } -/// The memory one KLL sketch of parameter `k` holds: asap_sketchlib -/// allocates its compactor buffers up front, about 4.5 KiB plus 32 bytes -/// per unit of `k` (10 KiB at k = 200, 36 KiB at k = 1024), and they do -/// not grow with updates. -pub(crate) fn kll_memory_bytes(k: u16) -> usize { - 4_608 + 32 * usize::from(k) -} - impl std::fmt::Debug for DatasketchesKLLAccumulator { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("DatasketchesKLLAccumulator") - .field("k", &self.inner.k) + .field("k", &self.inner.k()) .field("sketch_n", &self.inner.count()) .finish() } @@ -58,9 +54,17 @@ impl AggregateCore for DatasketchesKLLAccumulator { .as_any() .downcast_ref::() .ok_or("KLL merges only with KLL")?; - Ok(Box::new(Self { - inner: KllSketch::merge_refs(&[&self.inner, &other.inner])?, - })) + if self.inner.k() != other.inner.k() { + return Err(format!( + "KLL merges only with the same k ({} vs {})", + self.inner.k(), + other.inner.k() + ) + .into()); + } + let mut merged = self.clone(); + merged.inner.merge(&other.inner); + Ok(Box::new(merged)) } fn estimate(&self, query: &SketchStatistic) -> Result { @@ -78,8 +82,10 @@ impl AggregateCore for DatasketchesKLLAccumulator { self.inner.count() == 0 } + /// sketchlib's count of what the sketch holds now (its struct + /// included); the item buffer grows with the stream. fn approx_memory_bytes(&self) -> usize { - kll_memory_bytes(self.inner.k()) + self.inner.memory_bytes() } } @@ -115,4 +121,30 @@ mod tests { assert!(kll.estimate(&SketchStatistic::Quantile { q: 1.5 }).is_err()); assert!(kll.estimate(&SketchStatistic::Cardinality).is_err()); } + + // The charge is sketchlib's count of what the sketch holds, the struct + // counted once: it starts small and grows with the stream. + #[test] + fn memory_charge_is_the_library_count() { + let mut kll = DatasketchesKLLAccumulator::new(200); + assert_eq!(kll.approx_memory_bytes(), kll.inner.memory_bytes()); + let empty = kll.approx_memory_bytes(); + for v in 0..10_000 { + kll.update(f64::from(v)); + } + assert_eq!(kll.approx_memory_bytes(), kll.inner.memory_bytes()); + assert!(kll.approx_memory_bytes() > empty); + } + + // States of different k do not merge. + #[test] + fn merge_rejects_a_different_k() { + let kll = DatasketchesKLLAccumulator::new(200); + assert!(kll + .merge_with(&DatasketchesKLLAccumulator::new(100)) + .is_err()); + assert!(kll + .merge_with(&DatasketchesKLLAccumulator::new(200)) + .is_ok()); + } } diff --git a/crates/executor/src/summary_kernels/dd_sketch.rs b/crates/executor/src/summary_kernels/dd_sketch.rs index 735effce7..3dd30e837 100644 --- a/crates/executor/src/summary_kernels/dd_sketch.rs +++ b/crates/executor/src/summary_kernels/dd_sketch.rs @@ -33,13 +33,13 @@ impl AggregateCore for DDSketchAccumulator { .as_any() .downcast_ref::() .ok_or("DDSketch merges only with DDSketch")?; - Ok(Box::new(Self { - inner: DdSketch::merge_refs(&[&self.inner, &other.inner])?, - })) + let mut merged = self.clone(); + merged.inner.merge(&other.inner)?; + Ok(Box::new(merged)) } fn is_empty(&self) -> bool { - self.inner.total_count() == 0 + self.inner.get_count() == 0 } /// Quantiles, and the total sample count as a bare `PointCount`. @@ -47,10 +47,10 @@ impl AggregateCore for DDSketchAccumulator { match query { SketchStatistic::Quantile { q } if (0.0..=1.0).contains(q) => self .inner - .quantile(*q) + .get_value_at_quantile(*q) .ok_or_else(|| "DDSketch quantile of an empty population".into()), SketchStatistic::Quantile { .. } => Err("quantile must be in [0, 1]".into()), - SketchStatistic::PointCount { value: None, .. } => Ok(self.inner.total_count() as f64), + SketchStatistic::PointCount { value: None, .. } => Ok(self.inner.get_count() as f64), other => Err(format!("DDSketch does not answer {other:?}").into()), } } @@ -76,8 +76,8 @@ mod tests { DDSketchAccumulator::new(0.01), ); for v in 1..=50 { - a.inner.update(f64::from(v)); - b.inner.update(f64::from(v + 50)); + a.inner.add(&f64::from(v)); + b.inner.add(&f64::from(v + 50)); } let merged = a.merge_with(&b).unwrap(); assert_eq!(merged.estimate(&bare_count()).unwrap(), 100.0); @@ -94,4 +94,19 @@ mod tests { assert!(dd.estimate(&SketchStatistic::Quantile { q: 0.5 }).is_err()); assert!(dd.estimate(&SketchStatistic::Cardinality).is_err()); } + + // The quantile at q reads the value of rank ⌈q·n⌉; q = 0 and q = 1 read + // the exact minimum and maximum. + #[test] + fn quantile_reads_rank_ceil_q_n() { + let mut dd = DDSketchAccumulator::new(0.01); + for v in [1.0, 2.0, 3.0, 4.0] { + dd.inner.add(&v); + } + let at = |q| dd.estimate(&SketchStatistic::Quantile { q }).unwrap(); + assert_eq!(at(0.0), 1.0); + assert!((at(0.5) - 2.0).abs() <= 0.01 * 2.0, "{}", at(0.5)); + assert_eq!(at(0.9), 4.0); + assert_eq!(at(1.0), 4.0); + } } diff --git a/crates/executor/src/summary_kernels/factory.rs b/crates/executor/src/summary_kernels/factory.rs index 93324d057..a5808cbf4 100644 --- a/crates/executor/src/summary_kernels/factory.rs +++ b/crates/executor/src/summary_kernels/factory.rs @@ -172,8 +172,7 @@ impl AccumulatorUpdater for KllAccumulatorUpdater { } fn memory_usage_bytes(&self) -> usize { - std::mem::size_of::() - + crate::summary_kernels::datasketches_kll::kll_memory_bytes(self.k) + self.acc.approx_memory_bytes() } } @@ -206,7 +205,7 @@ impl AccumulatorUpdater for DDSketchAccumulatorUpdater { } fn update_single(&mut self, value: f64, _timestamp_ms: i64) -> Result<(), String> { - self.acc.inner.update(value); + self.acc.inner.add(&value); Ok(()) } @@ -230,7 +229,7 @@ impl AccumulatorUpdater for DDSketchAccumulatorUpdater { } fn memory_usage_bytes(&self) -> usize { - // Bucket store is variable; rough estimate matches KLL. + // A fixed estimate; the bucket stores grow with the values' range. std::mem::size_of::() + 4096 } } @@ -247,17 +246,13 @@ impl AccumulatorUpdater for DDSketchAccumulatorUpdater { /// not be inferred from the sketch algorithm alone. pub struct CmsAccumulatorUpdater { acc: CountMinSketchAccumulator, - row_num: usize, - col_num: usize, } impl CmsAccumulatorUpdater { - pub fn new(row_num: usize, col_num: usize) -> Self { - Self { - acc: CountMinSketchAccumulator::new(row_num, col_num), - row_num, - col_num, - } + pub fn new(rows: usize, cols: usize) -> Result { + Ok(Self { + acc: CountMinSketchAccumulator::new(rows, cols).map_err(|e| e.to_string())?, + }) } } @@ -272,14 +267,14 @@ impl AccumulatorUpdater for CmsAccumulatorUpdater { value: f64, _timestamp_ms: i64, ) -> Result<(), String> { - self.acc.inner.update(&key.to_semicolon_str(), value); + self.acc.update(&key.to_semicolon_str(), value); Ok(()) } impl_clone_accumulator_methods!(acc); fn reset(&mut self) { - self.acc = CountMinSketchAccumulator::new(self.row_num, self.col_num); + self.acc.inner.clear(); } fn is_keyed(&self) -> bool { @@ -287,8 +282,7 @@ impl AccumulatorUpdater for CmsAccumulatorUpdater { } 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() } } @@ -372,17 +366,17 @@ impl AccumulatorUpdater for WeightedFrequencyUpdater { pub struct HydraKllAccumulatorUpdater { acc: HydraKllSketchAccumulator, - row_num: usize, - col_num: usize, + rows: usize, + cols: usize, k: u16, } impl HydraKllAccumulatorUpdater { - pub fn new(row_num: usize, col_num: usize, k: u16) -> Self { + pub fn new(rows: usize, cols: usize, k: u16) -> Self { Self { - acc: HydraKllSketchAccumulator::new(row_num, col_num, k), - row_num, - col_num, + acc: HydraKllSketchAccumulator::new(rows, cols, k), + rows, + cols, k, } } @@ -406,7 +400,7 @@ impl AccumulatorUpdater for HydraKllAccumulatorUpdater { impl_clone_accumulator_methods!(acc); fn reset(&mut self) { - self.acc = HydraKllSketchAccumulator::new(self.row_num, self.col_num, self.k); + self.acc = HydraKllSketchAccumulator::new(self.rows, self.cols, self.k); } fn is_keyed(&self) -> bool { @@ -414,8 +408,7 @@ impl AccumulatorUpdater for HydraKllAccumulatorUpdater { } fn memory_usage_bytes(&self) -> usize { - // Rough estimate: each cell is a KLL sketch - std::mem::size_of::() + self.row_num * self.col_num * 4096 + self.acc.approx_memory_bytes() } } @@ -488,7 +481,7 @@ pub fn create_planner_accumulator( } (SketchAlgorithm::Cms, params @ SketchParams::Cms { .. }) => { let (r, c) = cms_dims(params); - Box::new(CmsAccumulatorUpdater::new(r, c)) + Box::new(CmsAccumulatorUpdater::new(r, c)?) } (SketchAlgorithm::CountSketch, params @ SketchParams::CountSketch { .. }) => { let (r, c) = cms_dims(params); @@ -506,10 +499,7 @@ pub fn create_planner_accumulator( ) } (SketchAlgorithm::Hll, SketchParams::Hll { precision, .. }) => Box::new(HllUpdater { - acc: HllSketchAccumulator::new( - asap_sketchlib::HllVariant::Regular, - u32::from(*precision), - ), + acc: HllSketchAccumulator::new(*precision).map_err(|e| e.to_string())?, }), (SketchAlgorithm::Kmv, SketchParams::Kmv { k, .. }) => { Box::new(crate::summary_kernels::kmv::KmvAccumulator::new(*k)) @@ -605,7 +595,7 @@ impl AccumulatorUpdater for HllUpdater { fn update_single(&mut self, value: f64, _: i64) -> Result<(), String> { if !value.is_nan() { let bits = if value == 0.0 { 0 } else { value.to_bits() }; - self.acc.inner.update(&bits.to_le_bytes()); + self.acc.update(&bits.to_le_bytes()); } Ok(()) } @@ -619,8 +609,7 @@ impl AccumulatorUpdater for HllUpdater { } impl_clone_accumulator_methods!(acc); fn reset(&mut self) { - self.acc.inner = - asap_sketchlib::HllSketch::new(self.acc.inner.variant, self.acc.inner.precision); + self.acc.clear(); } } diff --git a/crates/executor/src/summary_kernels/hll_sketch.rs b/crates/executor/src/summary_kernels/hll_sketch.rs index 166d8c2db..8697ef8b3 100644 --- a/crates/executor/src/summary_kernels/hll_sketch.rs +++ b/crates/executor/src/summary_kernels/hll_sketch.rs @@ -1,18 +1,123 @@ -//! HyperLogLog distinct-count summary over `asap_sketchlib::HllSketch`. +//! HyperLogLog distinct-count summary over `asap_sketchlib`'s classic +//! `HyperLogLogImpl`, at a precision chosen at run time. use crate::{AggregateCore, KernelError}; -use asap_sketchlib::{HllSketch, HllVariant}; +use asap_sketchlib::DataInput; use planner_types::ir::schema::SketchStatistic; +use registers::Registers; + +/// The precisions a state can be built at. +pub const HLL_PRECISIONS: std::ops::RangeInclusive = 4..=18; + +/// `asap_sketchlib` fixes an HLL's precision in its register type: one type +/// per precision in [`HLL_PRECISIONS`], and [`Registers`] holds any of them. +mod registers { + use asap_sketchlib::sketches::hll::HyperLogLogImpl; + use asap_sketchlib::{Classic, DataInput}; + + macro_rules! precisions { + ($($variant:ident($storage:ident, $precision:literal)),* $(,)?) => { + $(asap_sketchlib::impl_hll_registers!($storage, $precision, 1_usize << $precision);)* + + #[derive(Debug, Clone)] + pub enum Registers { + $($variant(HyperLogLogImpl),)* + } + + impl Registers { + pub fn new(precision: u8) -> Option { + match precision { + $($precision => Some(Self::$variant(HyperLogLogImpl::::new())),)* + _ => None, + } + } + + pub fn precision(&self) -> u8 { + match self { + $(Self::$variant(_) => $precision,)* + } + } + + pub fn update(&mut self, item: &DataInput) { + match self { + $(Self::$variant(hll) => hll.update(item),)* + } + } + + /// False, and `self` unchanged, when the precisions differ. + pub fn merge(&mut self, other: &Self) -> bool { + match (self, other) { + $((Self::$variant(hll), Self::$variant(theirs)) => { + hll.merge(theirs); + true + })* + _ => false, + } + } + + pub fn estimate(&self) -> usize { + match self { + $(Self::$variant(hll) => hll.estimate(),)* + } + } + + pub fn registers(&self) -> &[u8] { + match self { + $(Self::$variant(hll) => hll.registers_as_slice(),)* + } + } + } + }; + } + + precisions!( + P4(HllRegistersP4, 4), + P5(HllRegistersP5, 5), + P6(HllRegistersP6, 6), + P7(HllRegistersP7, 7), + P8(HllRegistersP8, 8), + P9(HllRegistersP9, 9), + P10(HllRegistersP10, 10), + P11(HllRegistersP11, 11), + P12(HllRegistersP12, 12), + P13(HllRegistersP13, 13), + P14(HllRegistersP14, 14), + P15(HllRegistersP15, 15), + P16(HllRegistersP16, 16), + P17(HllRegistersP17, 17), + P18(HllRegistersP18, 18), + ); +} #[derive(Debug, Clone)] pub struct HllSketchAccumulator { - pub inner: HllSketch, + inner: Registers, } impl HllSketchAccumulator { - pub fn new(variant: HllVariant, precision: u32) -> Self { - Self { - inner: HllSketch::new(variant, precision), - } + /// Fails for a precision outside [`HLL_PRECISIONS`]. + pub fn new(precision: u8) -> Result { + Registers::new(precision) + .map(|inner| Self { inner }) + .ok_or_else(|| format!("HLL precision {precision} is outside 4..=18").into()) + } + + pub fn precision(&self) -> u8 { + self.inner.precision() + } + + /// Observe one item, hashed by its bytes. + pub fn update(&mut self, item: &[u8]) { + self.inner.update(&DataInput::Bytes(item)); + } + + /// One rank per register. + pub fn registers(&self) -> &[u8] { + self.inner.registers() + } + + /// Empty the state in place, keeping its precision. + pub fn clear(&mut self) { + self.inner = Registers::new(self.precision()).expect("a built precision"); } } @@ -33,23 +138,31 @@ impl AggregateCore for HllSketchAccumulator { .as_any() .downcast_ref::() .ok_or("HLL merges only with HLL")?; - Ok(Box::new(Self { - inner: HllSketch::merge_refs(&[&self.inner, &other.inner])?, - })) + let mut merged = self.clone(); + if !merged.inner.merge(&other.inner) { + return Err(format!( + "HLL merges only with the same precision ({} vs {})", + self.precision(), + other.precision() + ) + .into()); + } + Ok(Box::new(merged)) } - /// Distinct count. A bare `PointCount` over an HLL also reads the distinct count. + /// Distinct count, rounded down to an integer. A bare `PointCount` over + /// an HLL also reads the distinct count. fn estimate(&self, query: &SketchStatistic) -> Result { match query { SketchStatistic::Cardinality | SketchStatistic::PointCount { value: None, .. } => { - Ok(self.inner.estimate()) + Ok(self.inner.estimate() as f64) } other => Err(format!("HLL does not answer {other:?}").into()), } } fn approx_memory_bytes(&self) -> usize { - std::mem::size_of::().saturating_add(self.inner.registers.capacity()) + std::mem::size_of::().saturating_add(self.registers().len()) } } @@ -61,12 +174,12 @@ mod tests { #[test] fn merged_cardinality_deduplicates_overlap() { let (mut a, mut b) = ( - HllSketchAccumulator::new(HllVariant::Regular, 12), - HllSketchAccumulator::new(HllVariant::Regular, 12), + HllSketchAccumulator::new(12).unwrap(), + HllSketchAccumulator::new(12).unwrap(), ); for v in 0..1000u32 { - a.inner.update(&v.to_le_bytes()); - b.inner.update(&(v + 500).to_le_bytes()); + a.update(&v.to_le_bytes()); + b.update(&(v + 500).to_le_bytes()); } let merged = a.merge_with(&b).unwrap(); let estimate = merged.estimate(&SketchStatistic::Cardinality).unwrap(); @@ -76,7 +189,37 @@ mod tests { // HLL does not answer quantiles. #[test] fn rejects_quantile() { - let hll = HllSketchAccumulator::new(HllVariant::Regular, 12); + let hll = HllSketchAccumulator::new(12).unwrap(); assert!(hll.estimate(&SketchStatistic::Quantile { q: 0.5 }).is_err()); } + + // Every precision in 4..=18 builds 2^p registers, others fail, and + // states of different precisions do not merge. + #[test] + fn precision_range_and_mismatched_merge() { + for precision in HLL_PRECISIONS { + let hll = HllSketchAccumulator::new(precision).unwrap(); + assert_eq!(hll.precision(), precision); + assert_eq!(hll.registers().len(), 1 << precision); + } + assert!(HllSketchAccumulator::new(3).is_err()); + assert!(HllSketchAccumulator::new(19).is_err()); + let (p12, p14) = ( + HllSketchAccumulator::new(12).unwrap(), + HllSketchAccumulator::new(14).unwrap(), + ); + assert!(p12.merge_with(&p14).is_err()); + } + + // The cardinality is an integer: 998 for 1,000 distinct items at p = 12. + #[test] + fn cardinality_is_an_integer() { + let mut hll = HllSketchAccumulator::new(12).unwrap(); + for v in 0..1000u32 { + hll.update(&v.to_le_bytes()); + } + let estimate = hll.estimate(&SketchStatistic::Cardinality).unwrap(); + assert_eq!(estimate.fract(), 0.0); + assert_eq!(estimate, 998.0); + } } diff --git a/crates/executor/src/summary_kernels/hydra_cms.rs b/crates/executor/src/summary_kernels/hydra_cms.rs index 811920ce8..4f835d991 100644 --- a/crates/executor/src/summary_kernels/hydra_cms.rs +++ b/crates/executor/src/summary_kernels/hydra_cms.rs @@ -9,7 +9,7 @@ //! A group's total needs a group-mass query that `asap_sketchlib`'s Hydra //! does not offer, so it is not answered here. use crate::{AggregateCore, KernelError}; -use asap_sketchlib::{input::HydraCounter, CountMin, DataInput, FastPath, Hydra, Vector2D}; +use asap_sketchlib::{input::HydraCounter, CountMinSketch, DataInput, FastPath, Hydra, Vector2D}; use planner_types::ir::schema::SketchStatistic; use std::sync::Arc; @@ -38,9 +38,9 @@ impl HydraCms { if [shared_rows, shared_columns, width, depth].contains(&0) { return Err("HydraCms dimensions must be positive".into()); } - let cell = HydraCounter::CM(CountMin::, FastPath>::with_dimensions( - depth, width, - )); + let cell = HydraCounter::CountMinSketch( + CountMinSketch::, FastPath>::with_dimensions(depth, width), + ); Ok(Self { inner: Hydra::with_schema(shared_rows, shared_columns, [GROUP], cell)?, width, @@ -51,12 +51,7 @@ impl HydraCms { /// `(shared_rows, shared_columns, width, depth)`. pub fn shape(&self) -> (usize, usize, usize, usize) { - ( - self.inner.row_num, - self.inner.col_num, - self.width, - self.depth, - ) + (self.inner.rows, self.inner.cols, self.width, self.depth) } /// Add `weight` occurrences of `item` in `group` (1 for a row count). diff --git a/crates/executor/src/summary_kernels/hydra_kll.rs b/crates/executor/src/summary_kernels/hydra_kll.rs index 8d6f6a795..66e5eae57 100644 --- a/crates/executor/src/summary_kernels/hydra_kll.rs +++ b/crates/executor/src/summary_kernels/hydra_kll.rs @@ -1,25 +1,39 @@ use crate::{AggregateCore, KeyByLabelValues}; -use asap_sketchlib::HydraKllSketch; +use asap_sketchlib::input::{HydraCounter, HydraQuery}; +use asap_sketchlib::{DataInput, Hydra, Kll}; -/// HydraKLL (shared-grouping quantiles) over `asap_sketchlib::HydraKllSketch`. +/// The single Hydra key column: a key's whole label tuple is one +/// subpopulation. +const KEY: &str = "key"; + +/// HydraKLL (shared-grouping quantiles): `asap_sketchlib::Hydra` over KLL cells. #[derive(Debug, Clone)] pub struct HydraKllSketchAccumulator { - pub inner: HydraKllSketch, + pub inner: Hydra, } impl HydraKllSketchAccumulator { - pub fn new(row_num: usize, col_num: usize, k: u16) -> Self { + pub fn new(rows: usize, cols: usize, k: u16) -> Self { + let cell = HydraCounter::Kll(Kll::with_k(i32::from(k))); Self { - inner: HydraKllSketch::new(row_num, col_num, k), + inner: Hydra::with_schema(rows, cols, [KEY], cell).expect("a one-column schema"), } } pub fn update(&mut self, key: &KeyByLabelValues, value: f64) { - self.inner.update(&key.to_semicolon_str(), value); + self.inner + .update(&[&key.to_semicolon_str()], &DataInput::F64(value), None) + .expect("one key per schema column"); } + /// The median over rows of each cell's KLL quantile. pub fn query_key(&self, key: &KeyByLabelValues, quantile: f64) -> f64 { - self.inner.quantile(&key.to_semicolon_str(), quantile) + self.inner + .query_key( + &[Some(&key.to_semicolon_str())], + &HydraQuery::Quantile(quantile), + ) + .expect("KLL cells answer quantiles") } } @@ -49,10 +63,49 @@ impl AggregateCore for HydraKllSketchAccumulator { Ok(Box::new(merged)) } + /// The KLL cells' memory, walking every cell. fn approx_memory_bytes(&self) -> usize { - // HydraKLL is a row*col grid of KLL sketches; typical instances - // are on the order of tens of KiB. 32 KiB is a conservative - // per-instance default. - 32 * 1024 + self.inner + .cells + .as_slice() + .iter() + .map(|cell| match cell { + HydraCounter::Kll(kll) => kll.memory_bytes(), + _ => 0, + }) + .fold(std::mem::size_of::(), usize::saturating_add) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + // A key's quantile reads its own values, and merge keeps both inputs'. + #[test] + fn quantiles_per_key_survive_merge() { + let (mut a, mut b) = ( + HydraKllSketchAccumulator::new(3, 64, 200), + HydraKllSketchAccumulator::new(3, 64, 200), + ); + let (x, y) = ( + KeyByLabelValues::new_with_labels(vec!["x".into()]), + KeyByLabelValues::new_with_labels(vec!["y".into()]), + ); + for v in 1..=50 { + a.update(&x, f64::from(v)); + b.update(&x, f64::from(v + 50)); + a.update(&y, 1000.0); + } + let merged = a.merge_with(&b).unwrap(); + let merged = merged + .as_any() + .downcast_ref::() + .unwrap(); + assert!((merged.query_key(&x, 1.0) - 100.0).abs() <= 1.0); + assert!(merged.query_key(&x, 0.0) <= 2.0); + assert!(a + .merge_with(&HydraKllSketchAccumulator::new(3, 32, 200)) + .is_err()); } } diff --git a/crates/executor/src/summary_kernels/kmv.rs b/crates/executor/src/summary_kernels/kmv.rs index 82e9adbc1..ac9b1f135 100644 --- a/crates/executor/src/summary_kernels/kmv.rs +++ b/crates/executor/src/summary_kernels/kmv.rs @@ -1,11 +1,11 @@ -//! KMV (k minimum values) distinct-count summary over `asap_sketchlib::KMV`. +//! KMV (k minimum values) distinct-count summary over `asap_sketchlib::Kmv`. //! -//! Every item is hashed with the library's shared `CANONICAL_HASH_SEED`, and +//! Every item is hashed at the library's `DEFAULT_CANONICAL_SEED_INDEX`, and //! a merge keeps the `k` smallest distinct hashes of both states, so merging //! two states equals building one over the union of their inputs. use crate::summary_kernels::factory::AccumulatorUpdater; use crate::{AggregateCore, KernelError, KeyByLabelValues}; -use asap_sketchlib::{DataInput, KMV}; +use asap_sketchlib::{DataInput, Kmv}; use planner_types::ir::schema::SketchStatistic; /// The smallest `k` the estimate `(k - 1) / U(k)` is defined for. @@ -13,13 +13,13 @@ pub const KMV_MIN_K: u32 = 2; #[derive(Debug, Clone)] pub struct KmvAccumulator { - pub inner: KMV, + pub inner: Kmv, } impl KmvAccumulator { pub fn new(k: u32) -> Self { Self { - inner: KMV::new(k as usize), + inner: Kmv::new(k as usize), } } @@ -28,7 +28,7 @@ impl KmvAccumulator { pub fn insert_sample(&mut self, value: f64) { if !value.is_nan() { let bits = if value == 0.0 { 0 } else { value.to_bits() }; - self.inner.insert(&DataInput::Bytes(&bits.to_le_bytes())); + self.inner.update(&DataInput::Bytes(&bits.to_le_bytes())); } } } @@ -73,7 +73,7 @@ impl AggregateCore for KmvAccumulator { } fn approx_memory_bytes(&self) -> usize { - self.inner.approx_memory_bytes() + self.inner.memory_bytes() } } @@ -116,7 +116,7 @@ impl AccumulatorUpdater for KmvAccumulator { Ok(()) } fn reset(&mut self) { - self.inner = KMV::new(self.inner.k); + self.inner = Kmv::new(self.inner.k); } } @@ -172,7 +172,7 @@ mod tests { let k = 256; for exact in [k, 5_000] { let kmv = build(k, 0..exact); - let kth = kmv.inner.k_vals.iter().copied().max().unwrap(); + let kth = kmv.inner.retained_hashes.iter().copied().max().unwrap(); let u = (kth as f64 + 1.0) / 18_446_744_073_709_551_616.0; assert_eq!(cardinality(&kmv), f64::from(k - 1) / u); } @@ -189,7 +189,7 @@ mod tests { let merged = a.merge_with(&b).unwrap(); let merged = merged.as_any().downcast_ref::().unwrap(); let sorted = |s: &KmvAccumulator| { - let mut hashes: Vec = s.inner.k_vals.iter().copied().collect(); + let mut hashes: Vec = s.inner.retained_hashes.iter().copied().collect(); hashes.sort_unstable(); hashes }; @@ -228,7 +228,7 @@ mod tests { fn memory_covers_the_retained_hashes() { let k = 4096; let (empty, full) = (KmvAccumulator::new(k), build(k, 0..20_000)); - assert_eq!(full.inner.k_vals.len(), k as usize); + assert_eq!(full.inner.retained_hashes.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/executor/src/summary_kernels/state_column_updater.rs b/crates/executor/src/summary_kernels/state_column_updater.rs index 47af5a00e..96ecc4fb0 100644 --- a/crates/executor/src/summary_kernels/state_column_updater.rs +++ b/crates/executor/src/summary_kernels/state_column_updater.rs @@ -48,8 +48,9 @@ enum GroupStates { /// A keyed frequency sketch (Count-Min or Count Sketch with a heap) per /// group. Keyed { - /// An empty state of the family's shape, cloned for each new group. - empty: WeightedFrequency, + /// An empty state of the family's shape, cloned for each new group + /// (boxed: much larger than the other kinds' fields). + empty: Box, /// The states, by group. states: Vec, }, @@ -94,7 +95,7 @@ impl StateColumnUpdater { { let (algorithm, width, depth, capacity) = WeightedFrequency::configuration(kind)?; GroupStates::Keyed { - empty: WeightedFrequency::new(algorithm, width, depth, capacity)?, + empty: Box::new(WeightedFrequency::new(algorithm, width, depth, capacity)?), states: Vec::new(), } } @@ -233,7 +234,7 @@ impl StateColumnUpdater { bytes } GroupStates::Keyed { empty, states } => { - states.push(empty.clone()); + states.push(empty.as_ref().clone()); empty.approx_memory_bytes() } GroupStates::Hydra { .. } => { diff --git a/crates/executor/src/summary_kernels/traits.rs b/crates/executor/src/summary_kernels/traits.rs index 2bbf707ea..6705cbb99 100644 --- a/crates/executor/src/summary_kernels/traits.rs +++ b/crates/executor/src/summary_kernels/traits.rs @@ -67,7 +67,7 @@ mod tests { .as_any_mut() .downcast_mut::() .unwrap(); - dd.inner.update(3.0); + dd.inner.add(&3.0); let count = SketchStatistic::PointCount { key: planner_types::ir::scalar::ColumnRef::SampleValue, value: None, diff --git a/crates/executor/src/summary_kernels/univmon.rs b/crates/executor/src/summary_kernels/univmon.rs index e47e2c846..2ca679eff 100644 --- a/crates/executor/src/summary_kernels/univmon.rs +++ b/crates/executor/src/summary_kernels/univmon.rs @@ -7,6 +7,20 @@ use planner_types::ir::schema::SketchStatistic; type Error = Box; +/// The shapes the executor builds: sketchlib's ([`UnivMon::check_shape`]), +/// with at most `MATRIX_MAX_ROWS` rows, the rows its ASAPv1 encoding takes, +/// so a deployment can store every state the executor builds. +pub fn check_shape(heap_size: usize, rows: usize, cols: usize, layers: usize) -> Result<(), Error> { + if rows > asap_sketchlib::MATRIX_MAX_ROWS { + return Err(format!( + "UnivMon rows {rows} exceeds MATRIX_MAX_ROWS {}", + asap_sketchlib::MATRIX_MAX_ROWS + ) + .into()); + } + Ok(UnivMon::check_shape(heap_size, rows, cols, layers)?) +} + #[derive(Debug, Clone)] pub struct UnivMonAccumulator { inner: UnivMon, @@ -24,15 +38,15 @@ pub struct UnivMonAccumulator { impl UnivMonAccumulator { /// Empty the sketch in place, keeping its shape. pub(crate) fn clear(&mut self) { - self.inner.free(); + self.inner.clear(); self.max_key_bytes = 0; self.inserted_key_bytes = 0; } pub fn new(heap_size: usize, rows: usize, cols: usize, layers: usize) -> Result { - UnivMon::check_shape(heap_size, rows, cols, layers)?; + check_shape(heap_size, rows, cols, layers)?; Ok(Self { - inner: UnivMon::init_univmon(heap_size, rows, cols, layers), + inner: UnivMon::new(heap_size, rows, cols, layers), max_key_bytes: 0, inserted_key_bytes: 0, }) @@ -42,21 +56,16 @@ 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 { - UnivMon::check_shape( - sketch.heap_size, - sketch.sketch_row, - sketch.sketch_col, - sketch.layer_size, - )?; + check_shape(sketch.heap_size, sketch.rows, sketch.cols, sketch.layers)?; if !sketch.accepts_standard_updates() { return Err("terminal-mode UnivMon state cannot accept standard updates".into()); } // One walk of the decoded heaps, not one per charge. - let keys: Vec = (0..sketch.layer_size) - .flat_map(|layer| sketch.hh_layers[layer].heap()) + let keys: Vec = (0..sketch.layers) + .flat_map(|layer| sketch.layer_heaps[layer].heap()) .map(|item| match &item.key { - asap_sketchlib::HeapItem::String(key) => key.capacity(), - asap_sketchlib::HeapItem::Bytes(key) => key.capacity(), + asap_sketchlib::OwnedDataInput::String(key) => key.capacity(), + asap_sketchlib::OwnedDataInput::Bytes(key) => key.capacity(), _ => 0, }) .collect(); @@ -79,11 +88,11 @@ impl UnivMonAccumulator { return Ok(()); } self.inner - .bucket_size + .total_weight .checked_add(1) .ok_or("UnivMon count overflow")?; let bits = if value == 0.0 { 0 } else { value.to_bits() }; - self.inner.insert(&DataInput::U64(bits), 1); + self.inner.update_weighted(&DataInput::U64(bits), 1); Ok(()) } @@ -101,10 +110,10 @@ impl UnivMonAccumulator { } let key = value.key()?; self.inner - .bucket_size + .total_weight .checked_add(1) .ok_or("UnivMon count overflow")?; - self.inner.insert(&DataInput::Bytes(&key), 1); + self.inner.update_weighted(&DataInput::Bytes(&key), 1); self.max_key_bytes = self.max_key_bytes.max(key.len()); self.inserted_key_bytes = self.inserted_key_bytes.saturating_add(key.len()); Ok(()) @@ -113,23 +122,23 @@ impl UnivMonAccumulator { fn compatible(&self, other: &Self) -> bool { ( self.inner.heap_size, - self.inner.sketch_row, - self.inner.sketch_col, - self.inner.layer_size, + self.inner.rows, + self.inner.cols, + self.inner.layers, ) == ( other.inner.heap_size, - other.inner.sketch_row, - other.inner.sketch_col, - other.inner.layer_size, + other.inner.rows, + other.inner.cols, + other.inner.layers, ) } pub fn dimensions(&self) -> (usize, usize, usize, usize) { ( self.inner.heap_size, - self.inner.sketch_row, - self.inner.sketch_col, - self.inner.layer_size, + self.inner.rows, + self.inner.cols, + self.inner.layers, ) } @@ -138,8 +147,8 @@ impl UnivMonAccumulator { return Err("incompatible UnivMon dimensions".into()); } self.inner - .bucket_size - .checked_add(other.inner.bucket_size) + .total_weight + .checked_add(other.inner.total_weight) .ok_or("UnivMon count overflow")?; self.inner.merge(&other.inner); self.max_key_bytes = self.max_key_bytes.max(other.max_key_bytes); @@ -158,7 +167,7 @@ impl AggregateCore for UnivMonAccumulator { /// every row; the library exposes no eviction hook for an exact running /// total, so a stream of long keys is still overcounted. fn approx_memory_bytes(&self) -> usize { - let layers = self.inner.layer_size; + let layers = self.inner.layers; let key_bytes = layers .saturating_mul(self.inner.heap_size) .saturating_mul(self.max_key_bytes) @@ -166,10 +175,10 @@ impl AggregateCore for UnivMonAccumulator { key_bytes .saturating_add(std::mem::size_of::()) .saturating_add( - self.inner.layer_size.saturating_mul( + self.inner.layers.saturating_mul( self.inner - .sketch_row - .saturating_mul(self.inner.sketch_col) + .rows + .saturating_mul(self.inner.cols) .saturating_mul(16) .saturating_add(self.inner.heap_size.saturating_mul(256)), ), @@ -202,10 +211,10 @@ impl AggregateCore for UnivMonAccumulator { SketchStatistic::PointCount { key: ColumnRef::SampleValue, value: None, - } => self.inner.calc_l1(), - SketchStatistic::Cardinality => self.inner.calc_card(), - SketchStatistic::FrequencyL2 => self.inner.calc_l2(), - SketchStatistic::FrequencyEntropy => self.inner.calc_entropy(), + } => self.inner.estimate_l1(), + SketchStatistic::Cardinality => self.inner.estimate_cardinality(), + SketchStatistic::FrequencyL2 => self.inner.estimate_l2(), + SketchStatistic::FrequencyEntropy => self.inner.estimate_entropy(), other => return Err(format!("UnivMon does not answer {other:?}").into()), }) } @@ -264,17 +273,31 @@ mod tests { // Adoption rejects terminal-mode sketches and dimensions `new` rejects. #[test] fn adoption_rejects_terminal_or_invalid_sketches() { - let mut terminal = UnivMon::init_univmon(4, 3, 16, 2); - terminal.fast_insert(&DataInput::U64(1), 1); + let mut terminal = UnivMon::new(4, 3, 16, 2); + terminal.update_weighted_terminal(&DataInput::U64(1), 1); assert!(UnivMonAccumulator::from_sketch(terminal.clone()).is_err()); - terminal.free(); + terminal.clear(); assert!(UnivMonAccumulator::from_sketch(terminal).is_ok()); - // `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; + // `UnivMon::new` refuses 26 rows of 16 columns (130 hash bits); a + // state claiming them is refused too. + let mut misshapen = UnivMon::new(4, 3, 16, 2); + misshapen.rows = 26; assert!(UnivMonAccumulator::from_sketch(misshapen).is_err()); } + // A 21-row UnivMon fits the hash budget (21 · (5 + 1) = 126 at 32 + // columns) and sketchlib builds it, but its ASAPv1 encoding refuses more + // than MATRIX_MAX_ROWS rows, so a deployment could not store it: the + // executor refuses it, building or adopting. + #[test] + fn rejects_rows_past_what_the_state_encodes() { + assert!(UnivMon::new(4, 21, 32, 1).to_asapv1().is_err()); + assert!(UnivMon::new(4, 20, 32, 1).to_asapv1().is_ok()); + assert!(UnivMonAccumulator::new(4, 21, 32, 1).is_err()); + assert!(UnivMonAccumulator::new(4, 20, 32, 1).is_ok()); + assert!(UnivMonAccumulator::from_sketch(UnivMon::new(4, 21, 32, 1)).is_err()); + } + // Shapes whose hashes are not independent are errors, not library // panics: more than 19 layers, or rows · (⌈log2 cols⌉ + 1) > 128. #[test] diff --git a/crates/executor/src/summary_kernels/weighted_frequency.rs b/crates/executor/src/summary_kernels/weighted_frequency.rs index b3a5e4b8e..93f7fd7d6 100644 --- a/crates/executor/src/summary_kernels/weighted_frequency.rs +++ b/crates/executor/src/summary_kernels/weighted_frequency.rs @@ -1,48 +1,103 @@ -//! ASAP type and trait adapter for sketchlib's Float64 weighted frequency kernel. +//! Float64 weighted Count-Min and Count Sketch over `asap_sketchlib`'s `f64` +//! regular-path matrices, with a top-k heap or, at capacity 0, without one. //! -//! 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. +//! An identity tuple is one `TupleKey`, hashed and retained as its bytes. use crate::AggregateCore; use crate::{values::Value, Error}; -pub use asap_sketchlib::FrequencyAlgorithm; -use asap_sketchlib::{FrequencyIdentity, WeightedFrequency as Kernel, WeightedFrequencyError}; -use serde::{Deserialize, Serialize}; +use asap_sketchlib::{ + CountMinSketch, CountMinSketchWithHeap, CountSketch, CountSketchWithHeap, DataInput, KeyColumn, + OwnedDataInput, RegularPath, TupleKey, Vector2D, +}; -fn adapt_error(error: WeightedFrequencyError) -> Error { - match error { - WeightedFrequencyError::Invalid(message) => Error::Invalid(message), - WeightedFrequencyError::Update(message) => Error::Operator(message), - } +/// Which matrix the state keeps. +#[derive(Copy, Clone, Debug, PartialEq, Eq)] +pub enum FrequencyAlgorithm { + Cms, + CountSketch, } -fn identity(value: &Value) -> Result { - Ok(match value { - Value::Null => FrequencyIdentity::Null, - Value::Bool(v) => FrequencyIdentity::Bool(*v), - Value::Int64(v) => FrequencyIdentity::Int64(*v), - Value::Float64(v) => FrequencyIdentity::Float64(*v), - Value::Utf8(v) => FrequencyIdentity::Utf8(v.to_string()), - _ => { - return Err(Error::Invalid( - "unsupported weighted frequency identity".into(), - )) + +type Matrix = Vector2D; + +#[derive(Clone)] +enum Sketch { + Cms(CountMinSketch), + CountSketch(CountSketch), + CmsHeap(CountMinSketchWithHeap), + CountSketchHeap(CountSketchWithHeap), +} + +/// The shapes a state is built with: those `asap_sketchlib` builds (positive +/// `width` and `depth`), and an odd Count Sketch `depth`, so its median is +/// one row's estimate. +pub fn check_shape(algorithm: FrequencyAlgorithm, width: usize, depth: usize) -> Result<(), Error> { + let shape = match algorithm { + FrequencyAlgorithm::Cms => CountMinSketch::check_shape(depth, width), + FrequencyAlgorithm::CountSketch if depth.is_multiple_of(2) => { + Err(format!("Count Sketch depth {depth} is not odd")) } - }) + FrequencyAlgorithm::CountSketch => CountSketch::check_shape(depth, width), + }; + shape.map_err(|problem| Error::Invalid(format!("invalid weighted frequency shape: {problem}"))) } -fn value(identity: FrequencyIdentity) -> Value { - match identity { - FrequencyIdentity::Null => Value::Null, - FrequencyIdentity::Bool(v) => Value::Bool(v), - FrequencyIdentity::Int64(v) => Value::Int64(v), - FrequencyIdentity::Float64(v) => Value::Float64(v), - FrequencyIdentity::Utf8(v) => Value::Utf8(v.into()), - } + +/// The canonical key of an identity tuple. A Float64 identity must be +/// finite; `-0.0` and `0.0` are one identity. +fn identity_key(values: &[Value]) -> Result, Error> { + let columns = values + .iter() + .map(|value| match value { + Value::Null => Ok(KeyColumn::Null), + Value::Bool(v) => Ok(KeyColumn::Bool(*v)), + Value::Int64(v) => Ok(KeyColumn::I64(*v)), + Value::Float64(v) if v.is_finite() => Ok(KeyColumn::F64(*v)), + Value::Float64(_) => Err(Error::Invalid( + "non-finite weighted frequency identity".into(), + )), + Value::Utf8(v) => Ok(KeyColumn::String(v.to_string())), + _ => Err(Error::Invalid( + "unsupported weighted frequency identity".into(), + )), + }) + .collect::, _>>()?; + TupleKey::new(columns).to_bytes().map_err(Error::Invalid) } -#[derive(Clone, Debug, Serialize, Deserialize)] -#[serde(transparent)] + +fn identity_values(key: &[u8]) -> Vec { + let tuple = TupleKey::from_bytes(key).expect("a retained key is a tuple key"); + tuple + .columns() + .iter() + .map(|column| match column { + KeyColumn::Null => Value::Null, + KeyColumn::Bool(v) => Value::Bool(*v), + KeyColumn::I64(v) => Value::Int64(*v), + KeyColumn::F64(v) => Value::Float64(*v), + KeyColumn::String(v) => Value::Utf8(v.as_str().into()), + }) + .collect() +} + +#[derive(Clone)] pub struct WeightedFrequency { - inner: Kernel, + sketch: Sketch, + width: usize, + depth: usize, + capacity: usize, + /// The sum of every weight's magnitude, which bounds every counter's: + /// while it is finite no counter can overflow. + total_magnitude: f64, } + +impl std::fmt::Debug for WeightedFrequency { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("WeightedFrequency") + .field("algorithm", &self.algorithm()) + .field("shape", &self.shape()) + .field("total_magnitude", &self.total_magnitude) + .finish() + } +} + impl WeightedFrequency { pub(crate) fn configuration( kind: &planner_types::ir::schema::SketchKind, @@ -71,74 +126,203 @@ impl WeightedFrequency { )) } }; - // A heap family needs a heap; the shape rules are the kernel's. + // A heap family needs a heap. if capacity == 0 { return Err(Error::Invalid( "weighted frequency heap capacity is 0".into(), )); } let (width, depth) = (width as usize, depth as usize); - Kernel::check_shape(algorithm, width, depth).map_err(adapt_error)?; + check_shape(algorithm, width, depth)?; Ok((algorithm, width, depth, capacity as usize)) } /// Algorithm and `(width, depth, capacity)` shape, so a deployment can check /// a decoded state against its declared family. pub fn algorithm(&self) -> FrequencyAlgorithm { - self.inner.algorithm() + match self.sketch { + Sketch::Cms(_) | Sketch::CmsHeap(_) => FrequencyAlgorithm::Cms, + Sketch::CountSketch(_) | Sketch::CountSketchHeap(_) => FrequencyAlgorithm::CountSketch, + } } pub fn shape(&self) -> (usize, usize, usize) { - self.inner.shape() + (self.width, self.depth, self.capacity) } + + /// Fails for a shape [`check_shape`] refuses. pub fn new( algorithm: FrequencyAlgorithm, width: usize, depth: usize, capacity: usize, ) -> Result { - Kernel::new(algorithm, width, depth, capacity) - .map(|inner| Self { inner }) - .map_err(adapt_error) + check_shape(algorithm, width, depth)?; + let sketch = match (algorithm, capacity) { + (FrequencyAlgorithm::Cms, 0) => { + Sketch::Cms(CountMinSketch::with_dimensions(depth, width)) + } + (FrequencyAlgorithm::CountSketch, 0) => { + Sketch::CountSketch(CountSketch::with_dimensions(depth, width)) + } + (FrequencyAlgorithm::Cms, _) => { + Sketch::CmsHeap(CountMinSketchWithHeap::new(depth, width, capacity)) + } + (FrequencyAlgorithm::CountSketch, _) => { + Sketch::CountSketchHeap(CountSketchWithHeap::new(depth, width, capacity)) + } + }; + Ok(Self { + sketch, + width, + depth, + capacity, + total_magnitude: 0.0, + }) } + /// 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 + let columns = labels .iter() - .map(|label| FrequencyIdentity::Utf8(label.clone())) - .collect::>(); - self.inner.update(&identity, weight).map_err(adapt_error) + .map(|label| KeyColumn::String(label.clone())) + .collect(); + let key = TupleKey::new(columns).to_bytes().map_err(Error::Invalid)?; + self.update_key(&key, weight) } - /// 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) + + /// The weight rule: finite, and nonnegative for Count-Min. + pub(crate) fn check_weight(&self, weight: f64) -> Result<(), String> { + if !weight.is_finite() || (self.algorithm() == FrequencyAlgorithm::Cms && weight < 0.0) { + return Err( + "weighted frequency requires finite weights; CMS additionally requires nonnegative weights".into(), + ); + } + Ok(()) } + /// Empties the state in place, keeping its algorithm and shape. pub(crate) fn clear(&mut self) { - self.inner.clear(); + match &mut self.sketch { + Sketch::Cms(s) => s.clear(), + Sketch::CountSketch(s) => s.clear(), + Sketch::CmsHeap(s) => s.clear(), + Sketch::CountSketchHeap(s) => s.clear(), + } + self.total_magnitude = 0.0; } + + /// Adds `weight` to the identity tuple `values`. An error, and the state + /// unchanged, for a weight [`Self::check_weight`] refuses, an identity + /// that is not Null, Bool, Int64, finite Float64 or Utf8, or a total + /// weight past `f64::MAX`. 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) + let key = identity_key(values)?; + self.update_key(&key, weight) + } + + fn update_key(&mut self, key: &[u8], weight: f64) -> Result<(), Error> { + self.check_weight(weight).map_err(Error::Operator)?; + let total = self.total_magnitude + weight.abs(); + if !total.is_finite() { + return Err(Error::Operator("weighted frequency sum overflow".into())); + } + let key = DataInput::Bytes(key); + match &mut self.sketch { + Sketch::Cms(s) => s.update_weighted(&key, weight), + Sketch::CountSketch(s) => s.update_weighted(&key, weight), + Sketch::CmsHeap(s) => s.update_weighted(&key, weight), + Sketch::CountSketchHeap(s) => s.update_weighted(&key, weight), + } + self.total_magnitude = total; + Ok(()) + } + + fn estimate_key(&self, key: &[u8]) -> f64 { + let key = DataInput::Bytes(key); + match &self.sketch { + Sketch::Cms(s) => s.estimate(&key), + Sketch::CountSketch(s) => s.estimate(&key), + Sketch::CmsHeap(s) => s.estimate(&key), + Sketch::CountSketchHeap(s) => s.estimate(&key), + } } + /// 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) + Ok(self.estimate_key(&identity_key(values)?)) + } + + /// The keys the heap retains; none at capacity 0. + fn retained_keys(&self) -> Vec<&[u8]> { + let keys: Vec<&OwnedDataInput> = match &self.sketch { + Sketch::Cms(_) | Sketch::CountSketch(_) => return Vec::new(), + Sketch::CmsHeap(s) => s.heap().heap().iter().map(|e| &e.key).collect(), + Sketch::CountSketchHeap(s) => s.heap().heap().iter().map(|e| &e.key).collect(), + }; + keys.into_iter() + .map(|key| match key { + OwnedDataInput::Bytes(bytes) => bytes.as_slice(), + _ => unreachable!("the state retains tuple-key bytes"), + }) + .collect() } + + /// Up to `n` retained identity tuples, each followed by its current + /// estimate as a Float64, by estimate descending, ties by key bytes + /// ascending. Retention alone does not establish top-k completeness. pub fn rows(&self, n: usize) -> Vec> { - self.inner - .topk(n) + let mut scored = self + .retained_keys() .into_iter() - .map(|(items, score)| { - let mut row = items.into_iter().map(value).collect::>(); + .map(|key| (self.estimate_key(key), key)) + .collect::>(); + scored.sort_by(|a, b| b.0.total_cmp(&a.0).then_with(|| a.1.cmp(b.1))); + scored + .into_iter() + .take(n) + .map(|(score, key)| { + let mut row = identity_values(key); row.push(Value::Float64(score)); row }) .collect() } + + /// The `depth x width` counters, row by row. + pub fn counters(&self) -> &[f64] { + match &self.sketch { + Sketch::Cms(s) => s.as_storage().as_slice(), + Sketch::CountSketch(s) => s.as_storage().as_slice(), + Sketch::CmsHeap(s) => s.count_min_sketch().as_storage().as_slice(), + Sketch::CountSketchHeap(s) => s.count_sketch().as_storage().as_slice(), + } + } + + /// Merge with a state of the same algorithm and shape, leaving both unchanged. + pub fn merge(&self, other: &Self) -> Result { + // sketchlib's merges panic on a shape mismatch. + if self.algorithm() != other.algorithm() || self.shape() != other.shape() { + return Err(Error::Invalid("weighted frequency shape mismatch".into())); + } + let total = self.total_magnitude + other.total_magnitude; + if !total.is_finite() { + return Err(Error::Operator("weighted frequency merge overflow".into())); + } + let mut merged = self.clone(); + match (&mut merged.sketch, &other.sketch) { + (Sketch::Cms(s), Sketch::Cms(o)) => s.merge(o), + (Sketch::CountSketch(s), Sketch::CountSketch(o)) => s.merge(o), + (Sketch::CmsHeap(s), Sketch::CmsHeap(o)) => s.merge(o), + (Sketch::CountSketchHeap(s), Sketch::CountSketchHeap(o)) => s.merge(o), + _ => unreachable!("one algorithm and capacity is one kind of sketch"), + } + merged.total_magnitude = total; + Ok(merged) + } } + impl AggregateCore for WeightedFrequency { fn clone_boxed_core(&self) -> Box { Box::new(self.clone()) @@ -157,12 +341,18 @@ impl AggregateCore for WeightedFrequency { .as_any() .downcast_ref::() .ok_or("weighted frequency state type mismatch")?; - Ok(Box::new(Self { - inner: self.inner.merge(&other.inner)?, - })) + Ok(Box::new(self.merge(other)?)) } + /// This adapter's fields plus sketchlib's count of the sketch, which + /// includes the sketch's own struct. fn approx_memory_bytes(&self) -> usize { - self.inner.approx_memory_bytes() + std::mem::size_of::() - std::mem::size_of::() + + match &self.sketch { + Sketch::Cms(s) => s.memory_bytes(), + Sketch::CountSketch(s) => s.memory_bytes(), + Sketch::CmsHeap(s) => s.memory_bytes(), + Sketch::CountSketchHeap(s) => s.memory_bytes(), + } } } @@ -170,6 +360,21 @@ impl AggregateCore for WeightedFrequency { mod tests { use super::*; + // The charge is sketchlib's count of the sketch plus this adapter's own + // fields, each struct counted once. + #[test] + fn memory_charge_counts_the_sketch_once() { + let state = WeightedFrequency::new(FrequencyAlgorithm::Cms, 64, 3, 4).unwrap(); + let Sketch::CmsHeap(sketch) = &state.sketch else { + unreachable!() + }; + assert_eq!( + state.approx_memory_bytes(), + std::mem::size_of::() - std::mem::size_of::() + + sketch.memory_bytes() + ); + } + // Merge uses the same Float64 state representation and rejects other shapes. #[test] fn compatible_merge_preserves_fractional_weights() { @@ -184,4 +389,79 @@ mod tests { .merge_with(&WeightedFrequency::new(FrequencyAlgorithm::Cms, 32, 5, 8).unwrap()) .is_err()); } + + // Count Sketch keeps signed weights and ranks by signed estimate; typed + // identities do not collide by formatting. + #[test] + fn signed_weights_and_typed_identities() { + let mut state = + WeightedFrequency::new(FrequencyAlgorithm::CountSketch, 4096, 5, 8).unwrap(); + state.update(&[Value::Int64(1)], -10.5).unwrap(); + state.update(&[Value::Utf8("1".into())], 0.125).unwrap(); + state.update(&[Value::Null], 0.3125).unwrap(); + let rows = state.rows(3); + assert!(matches!(rows[0][0], Value::Null)); + assert!(matches!(rows[1][0], Value::Utf8(_))); + assert!(matches!(rows[2][..], [Value::Int64(1), Value::Float64(score)] if score == -10.5)); + } + + // Weights outside the rule, non-finite or unsupported identities and a + // total weight past f64::MAX are errors that leave the state unchanged. + #[test] + fn invalid_updates_leave_the_state_unchanged() { + let mut cms = WeightedFrequency::new(FrequencyAlgorithm::Cms, 64, 3, 4).unwrap(); + cms.update(&[Value::Int64(1)], f64::MAX).unwrap(); + let snapshot = + |s: &WeightedFrequency| (s.counters().to_vec(), format!("{:?}", s.rows(usize::MAX))); + let before = snapshot(&cms); + for weight in [-1.0, f64::INFINITY, f64::NAN, f64::MAX] { + assert!(cms.update(&[Value::Null], weight).is_err()); + } + for identity in [ + Value::Float64(f64::NAN), + Value::Float64(f64::INFINITY), + Value::Date(1), + ] { + assert!(cms.update(&[identity], 1.0).is_err()); + } + assert_eq!(snapshot(&cms), before); + let mut cs = WeightedFrequency::new(FrequencyAlgorithm::CountSketch, 64, 3, 4).unwrap(); + assert!(cs.update(&[Value::Null], -1.0).is_ok()); + cms.clear(); + assert!(cms.rows(4).is_empty()); + } + + // Signed zeros are one identity; an even Count Sketch depth and a zero + // dimension are refused, and deep shapes are built. + #[test] + fn signed_zero_and_shape_rules() { + let mut state = WeightedFrequency::new(FrequencyAlgorithm::Cms, 64, 3, 4).unwrap(); + state.update(&[Value::Float64(-0.0)], 1.0).unwrap(); + state.update(&[Value::Float64(0.0)], 2.0).unwrap(); + let rows = state.rows(4); + assert_eq!(rows.len(), 1); + assert!(matches!(rows[0][..], [Value::Float64(z), Value::Float64(3.0)] if z == 0.0)); + assert!(WeightedFrequency::new(FrequencyAlgorithm::CountSketch, 8, 2, 2).is_err()); + assert!(WeightedFrequency::new(FrequencyAlgorithm::Cms, 0, 3, 2).is_err()); + assert!(WeightedFrequency::new(FrequencyAlgorithm::Cms, 8, 0, 2).is_err()); + assert!(WeightedFrequency::new(FrequencyAlgorithm::Cms, 8, 32, 2).is_ok()); + assert!(WeightedFrequency::new(FrequencyAlgorithm::CountSketch, 8, 255, 0).is_ok()); + } + + // States deeper than ASAPv1's 20 rows build, update, merge and read. + #[test] + fn deep_states_update_merge_and_read() { + for (algorithm, width, depth, capacity) in [ + (FrequencyAlgorithm::Cms, 272, 32, 4), + (FrequencyAlgorithm::CountSketch, 64, 21, 0), + (FrequencyAlgorithm::CountSketch, 64, 21, 4), + ] { + let mut state = WeightedFrequency::new(algorithm, width, depth, capacity).unwrap(); + state.update(&[Value::Int64(1)], 2.0).unwrap(); + let merged = state.merge(&state).unwrap(); + assert_eq!(merged.point_estimate(&[Value::Int64(1)]).unwrap(), 4.0); + assert_eq!(merged.counters().len(), width * depth); + assert_eq!(merged.rows(usize::MAX).len(), capacity.min(1)); + } + } } diff --git a/crates/executor/src/values/mod.rs b/crates/executor/src/values/mod.rs index 7a5d41bd2..a75bdc957 100644 --- a/crates/executor/src/values/mod.rs +++ b/crates/executor/src/values/mod.rs @@ -262,15 +262,15 @@ pub(super) fn validate_state( SketchParams::Kll { k, .. } => state .as_any() .downcast_ref::() - .is_some_and(|s| u32::from(s.inner.k()) == *k), + .is_some_and(|s| s.inner.k() == *k as usize), SketchParams::DDSketch { alpha } => state .as_any() .downcast_ref::() - .is_some_and(|s| s.inner.alpha == *alpha), + .is_some_and(|s| s.inner.alpha() == *alpha), SketchParams::Hll { precision, .. } => state .as_any() .downcast_ref::() - .is_some_and(|s| s.inner.precision == u32::from(*precision)), + .is_some_and(|s| s.precision() == *precision), SketchParams::Kmv { k, .. } => state .as_any() .downcast_ref::() diff --git a/crates/executor/tests/deployment.rs b/crates/executor/tests/deployment.rs index 4e43b2c4e..c039c847e 100644 --- a/crates/executor/tests/deployment.rs +++ b/crates/executor/tests/deployment.rs @@ -173,10 +173,9 @@ fn assert_rows_hash_independently(family: &FieldDataType, width: usize, depth: u .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() { + for (index, cell) in state.counters().iter().enumerate() { let (row, column) = (index / width, index % width); - let mut rest = cell.as_f64().unwrap() as i64; + let mut rest = *cell as i64; let mut position = 0; while rest != 0 { let digit = rest.rem_euclid(3); @@ -295,8 +294,8 @@ fn heap_sketch_states_load_as_dag_summary_values() { // interchangeable: merged either way, they equal one build over both inputs. #[test] fn factory_and_dag_heap_states_are_interchangeable() { + use asap_executor::summary_kernels::weighted_frequency::FrequencyAlgorithm; use asap_executor::{summary_kernels::weighted_frequency::WeightedFrequency, values::Value}; - use asap_sketchlib::FrequencyAlgorithm; let estimate = |state: &dyn AggregateCore, i: usize| { state .as_any() @@ -377,14 +376,12 @@ fn plain_count_sketch_merge_equals_union_build() { .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() + state + .as_any() + .downcast_ref::() + .unwrap() + .counters() + .to_vec() }; assert_eq!(cells(&*merged), cells(&*union)); for i in 0..500 { diff --git a/crates/executor/tests/deployment_computation.rs b/crates/executor/tests/deployment_computation.rs index 3c7225fe3..8b1a60a2c 100644 --- a/crates/executor/tests/deployment_computation.rs +++ b/crates/executor/tests/deployment_computation.rs @@ -667,9 +667,9 @@ fn stored_count_min_bare_count_compiles_to_a_evaluation() { .unwrap(); let program: CompiledPhysicalDAG = serde_json::from_slice(&serde_json::to_vec(&program).unwrap()).unwrap(); - let mut sketch = CountMinSketchAccumulator::new(*depth as usize, *width as usize); - sketch.inner.update("a", 3.0); - sketch.inner.update("b", 7.0); + let mut sketch = CountMinSketchAccumulator::new(*depth as usize, *width as usize).unwrap(); + sketch.update("a", 3.0); + sketch.update("b", 7.0); let row = schema .fields .iter() diff --git a/crates/executor/tests/state_column_updater.rs b/crates/executor/tests/state_column_updater.rs index 2998146ee..4cfcad8e6 100644 --- a/crates/executor/tests/state_column_updater.rs +++ b/crates/executor/tests/state_column_updater.rs @@ -213,7 +213,7 @@ fn fingerprint(state: &dyn AggregateCore) -> String { .join(","); } if let Some(s) = any.downcast_ref::() { - return serde_json::to_string(s).unwrap(); + return format!("{:?} {:?}", s.counters(), s.rows(usize::MAX)); } if let Some(s) = any.downcast_ref::() { // The group's name and its count of every item the samples use. @@ -380,13 +380,15 @@ fn kernel_build(family: &FieldDataType, keyed: bool, sharded: bool) -> BTreeMap< ), _ => unreachable!(), }; - Err(WeightedFrequency::new( - algorithm, - *width as usize, - *depth as usize, - *capacity as usize, - ) - .unwrap()) + Err(Box::new( + WeightedFrequency::new( + algorithm, + *width as usize, + *depth as usize, + *capacity as usize, + ) + .unwrap(), + )) } else { Ok(create_planner_accumulator(family, &update, &Default::default()).unwrap()) } @@ -405,7 +407,7 @@ fn kernel_build(family: &FieldDataType, keyed: bool, sharded: bool) -> BTreeMap< series, match state { Ok(updater) => Arc::::from(updater.into_accumulator()), - Err(keyed) => Arc::new(keyed) as Arc, + Err(keyed) => Arc::new(*keyed) as Arc, }, ) }) diff --git a/crates/executor/tests/stored_pane_memory_limit.rs b/crates/executor/tests/stored_pane_memory_limit.rs index c7fd9e18f..f428900d2 100644 --- a/crates/executor/tests/stored_pane_memory_limit.rs +++ b/crates/executor/tests/stored_pane_memory_limit.rs @@ -16,6 +16,8 @@ use std::sync::Arc; const SERIES: u32 = 1_000; const PANES: usize = 12; +/// Samples per state: past 1,024 a k = 200 KLL holds its whole item buffer. +const SAMPLES: u32 = 1_100; /// One pane of `SERIES` KLL states (k = 200, about 11 kB charged each, so /// about 11 MB per pane), keyed by series. @@ -57,7 +59,11 @@ fn pane_of(series_count: u32) -> (Arc, Batch) { let mut updater = StateColumnUpdater::new(&family).unwrap(); updater.ensure_groups(series_count as usize).unwrap(); for series in 0..series_count { - updater.update(series, f64::from(series), 0).unwrap(); + for sample in 0..SAMPLES { + updater + .update(series, f64::from(series) + f64::from(sample), 0) + .unwrap(); + } } let names = (0..series_count) .map(|series| format!("s{series}")) diff --git a/crates/integration-tests/tests/joint_search_equivalence.rs b/crates/integration-tests/tests/joint_search_equivalence.rs index 3eabe1da9..8b22fc043 100644 --- a/crates/integration-tests/tests/joint_search_equivalence.rs +++ b/crates/integration-tests/tests/joint_search_equivalence.rs @@ -548,21 +548,23 @@ fn a_query_with_no_valid_candidate_is_named_in_the_failure() { #[test] fn a_calibrated_cost_model_selects_the_exhaustive_plan() { use asap_plan_selection::{MeasuredOperationCosts, OperatorCost, OperatorKind}; - let mut settings = PlanSelectionSettings::default(); // Scans are priced from the table; every other node is calibrated by // the factor fitted over the scans. - settings.measured_operation_costs = MeasuredOperationCosts { - operators: vec![OperatorCost { - operator: OperatorKind::Scan, - phase: None, - sketch: None, - exact: None, - measure: None, - ns_per_unit: 300.0, - ns_per_invocation: 0.0, - memory_bytes_per_state: None, - }], - ..MeasuredOperationCosts::default() + let settings = PlanSelectionSettings { + measured_operation_costs: MeasuredOperationCosts { + operators: vec![OperatorCost { + operator: OperatorKind::Scan, + phase: None, + sketch: None, + exact: None, + measure: None, + ns_per_unit: 300.0, + ns_per_invocation: 0.0, + memory_bytes_per_state: None, + }], + ..MeasuredOperationCosts::default() + }, + ..PlanSelectionSettings::default() }; let mut searched = 0; for seed in 1..=30u64 { @@ -581,7 +583,7 @@ fn a_calibrated_cost_model_selects_the_exhaustive_plan() { &format!("seed {seed}"), &workload, &roots, - &[settings.clone()], + std::slice::from_ref(&settings), ); } assert!( diff --git a/crates/integration-tests/tests/promql_numeric_regressions.rs b/crates/integration-tests/tests/promql_numeric_regressions.rs index 6d93b1a8a..0ec2483e5 100644 --- a/crates/integration-tests/tests/promql_numeric_regressions.rs +++ b/crates/integration-tests/tests/promql_numeric_regressions.rs @@ -105,6 +105,22 @@ fn sum_rate_and_increase_have_real_exact_accumulator_nodes() { } } +/// The PromQL quantile over a DDSketch: the linear interpolation at rank +/// `q (n - 1)` between the sketch's two neighbouring order statistics. +fn interpolated_quantile(sketch: &asap_sketchlib::DdSketch, q: f64) -> f64 { + let n = sketch.get_count(); + // The `j`-th smallest (from 0): the sketch's value at rank `ceil(p n) = j + 1`. + let order_statistic = |j: u64| { + sketch + .get_value_at_quantile((j as f64 + 0.5) / n as f64) + .unwrap() + }; + let rank = q * (n - 1) as f64; + let (lower, upper) = (rank.floor() as u64, rank.ceil() as u64); + let fraction = rank - lower as f64; + order_statistic(lower) * (1. - fraction) + order_statistic(upper) * fraction +} + /// A finite, nonzero approximate denominator does not certify a ratio bound. #[test] fn checked_ratio_must_not_certify_cross_zero_interpolation() { @@ -121,13 +137,16 @@ fn checked_ratio_must_not_certify_cross_zero_interpolation() { post_asap_dag(&node); // Keep the actual signed-sketch counterexample: division guards alone pass // even though the quantile interpolation does not preserve relative error. + // The sketch reads its minimum and maximum exactly, so the interpolated + // order statistics are interior ones. let alpha = (0.01 - 8.0 * f64::EPSILON) / 2.01; + let values = [-2., -1., 1.011, 2.]; let estimate = |q| { let mut sketch = asap_sketchlib::DdSketch::new(alpha); - for value in [-1., 1.011] { - sketch.try_update(value).unwrap(); + for value in values { + sketch.try_add(&value).unwrap(); } - sketch.quantile_interpolated(q).unwrap() + interpolated_quantile(&sketch, q) }; let numerator = estimate(0.5); let denominator = estimate(0.9); @@ -135,7 +154,11 @@ fn checked_ratio_must_not_certify_cross_zero_interpolation() { assert!( numerator.is_finite() && denominator.is_finite() && denominator != 0. && got.is_normal() ); - let exact_quantile = |q: f64| -(1. - q) + 1.011 * q; + let exact_quantile = |q: f64| { + let rank = q * (values.len() - 1) as f64; + let (lower, upper) = (rank.floor() as usize, rank.ceil() as usize); + values[lower] * (1. - rank.fract()) + values[upper] * rank.fract() + }; let want = exact_quantile(0.5) / exact_quantile(0.9); let error = (got - want).abs() / want.abs(); assert!( @@ -238,26 +261,25 @@ fn sketch_counts_use_unit_weights_and_signed_sums_keep_value_weights() { } } for value in [1., 0., 3., -3.] { - let mut cms = - asap_sketchlib::message_pack_format::portable::countminsketch::new_sketchlib_cms( - 5, 128, - ); + let key = asap_sketchlib::DataInput::Str("a"); + let mut cms: asap_sketchlib::CountMinF64 = + asap_sketchlib::CountMinSketch::with_dimensions(5, 128); let mut cs = - asap_sketchlib::message_pack_format::portable::countsketch::CountSketch::new( + asap_sketchlib::CountSketch::>::with_dimensions( 5, 128, ); for _ in 0..10 { let weight = contribution(family, update, value); if is_count { - cms.insert_many(&asap_sketchlib::DataInput::String("a".into()), weight); + cms.update_weighted(&key, weight); } else { - cs.update("a", weight); + cs.update_weighted(&key, weight); } } let got = if is_count { - cms.estimate(&asap_sketchlib::DataInput::String("a".into())) + cms.estimate(&key) } else { - cs.estimate("a") + cs.estimate(&key) }; assert_eq!(got, if is_count { 10. } else { 10. * value }); } diff --git a/crates/integration-tests/tests/promql_to_post_asap.rs b/crates/integration-tests/tests/promql_to_post_asap.rs index 4cfd6f159..1d06bea51 100644 --- a/crates/integration-tests/tests/promql_to_post_asap.rs +++ b/crates/integration-tests/tests/promql_to_post_asap.rs @@ -836,11 +836,23 @@ fn ddsketch_ratio_bound_holds_for_signed_pinned_sketch_evaluations() { }; let mut sketch = asap_sketchlib::DdSketch::new(alpha); for v in values { - sketch.try_update(v).unwrap(); + sketch.try_add(&v).unwrap(); } + // PromQL's quantile: interpolated between neighbouring order + // statistics, the `j`-th smallest read at rank `j + 1`. + let n = sketch.get_count(); + let interpolated = |q: f64| { + let order_statistic = |j: u64| { + sketch + .get_value_at_quantile((j as f64 + 0.5) / n as f64) + .unwrap() + }; + let r = q * (n - 1) as f64; + order_statistic(r.floor() as u64) * (1. - r.fract()) + + order_statistic(r.ceil() as u64) * r.fract() + }; let want = exact(0.9) / exact(0.5); - let got = sketch.quantile_interpolated(0.9).unwrap() - / sketch.quantile_interpolated(0.5).unwrap(); + let got = interpolated(0.9) / interpolated(0.5); assert!((got - want).abs() / want.abs() <= bound + 1e-12); } } diff --git a/crates/logical-asap-optimization/src/accuracy/estimators/hll.rs b/crates/logical-asap-optimization/src/accuracy/estimators/hll.rs index 38bf9c77a..65ba001da 100644 --- a/crates/logical-asap-optimization/src/accuracy/estimators/hll.rs +++ b/crates/logical-asap-optimization/src/accuracy/estimators/hll.rs @@ -434,11 +434,11 @@ mod tests { h = (h ^ (h >> 30)).wrapping_mul(0xbf58476d1ce4e5b9); h = (h ^ (h >> 27)).wrapping_mul(0x94d049bb133111eb); h ^= h >> 31; - single.insert_with_hash(h); + single.update_with_hash(h); if n % 2 == 0 { - left.insert_with_hash(h); + left.update_with_hash(h); } else { - right.insert_with_hash(h); + right.update_with_hash(h); } } left.merge(&right); diff --git a/crates/logical-asap-optimization/src/accuracy/estimators/univmon.rs b/crates/logical-asap-optimization/src/accuracy/estimators/univmon.rs index 78e4027a9..da36047c6 100644 --- a/crates/logical-asap-optimization/src/accuracy/estimators/univmon.rs +++ b/crates/logical-asap-optimization/src/accuracy/estimators/univmon.rs @@ -1,6 +1,7 @@ //! UnivMon's readouts and the guarantees `asap_sketchlib`'s standard-mode -//! estimators carry (the derivations are in the library's `calc_l2` and -//! `calc_card` doc comments): +//! estimators carry (the derivations are in the library's +//! `docs/api/api_univmon.md`, the bounds `estimate_l2` and +//! `estimate_cardinality` cite): //! - the total (a bare `PointCount`) is the exact unit-update count; //! - L2 reads layer 0's Count Sketch directly: per row, Chebyshev with //! `Var ≤ 2q F2²`, and the median of the rows fails only when at least @@ -20,9 +21,12 @@ use asap_types::ir::operator::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 = 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. +pub(crate) const MAX_LAYERS: u8 = + asap_sketchlib::sketch_framework::univmon::UNIVMON_MAX_LAYERS as u8; +/// The most rows an L2 sketch is sized with: the deepest odd shape within +/// the 20 rows the executor builds ([`hash_independent`]), so the row median +/// is one row's estimate, and within the hash budget at the more than 20 +/// columns an L2 guarantee needs (5 column bits: 19 · 6 = 114 ≤ 128). const MAX_L2_ROWS: u32 = 19; /// Each row of a layer reads its column and sign bits from one hash. const HASH_BITS: u32 = 128; @@ -30,7 +34,7 @@ const HASH_BITS: u32 = 128; /// more than `ε₂ F2` with probability at most `2q/ε₂² = 0.1`. Fixing it /// lets a shape state one `(ε, δ)`: the columns give `ε`, the rows `δ`. const L2_ROW_FAILURE: f64 = 0.1; -/// `Var[calc_card] ≤ 1.35 F0²/k` for `k ≥ 8`, the constant the library +/// `Var[estimate_cardinality] ≤ 1.35 F0²/k` for `k ≥ 8`, the constant the library /// proves (it also states the looser `2 F0²/k`). const DISTINCT_VARIANCE: f64 = 1.35; /// The smallest heap the distinct-count variance bound holds for. @@ -91,12 +95,19 @@ pub(super) fn guarantee( } } -/// 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). +/// Whether the executor builds this shape, with independent hashes: at +/// most [`MAX_LAYERS`] layers and `MATRIX_MAX_ROWS` (20) rows, the rows a +/// UnivMon state encodes, 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 { - asap_sketchlib::UnivMon::check_shape(1, rows as usize, cols as usize, usize::from(layers)) + rows as usize <= asap_sketchlib::MATRIX_MAX_ROWS + && asap_sketchlib::UnivMon::check_shape( + 1, + rows as usize, + cols as usize, + usize::from(layers), + ) .is_ok() } @@ -485,14 +496,15 @@ 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). + /// The executor builds at most 20 rows, the rows a UnivMon state + /// encodes, so a 21-row UnivMon certifies nothing even though its hashes + /// fit the budget (21 · (5 + 1) = 126 ≤ 128 at 32 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()); + assert!(hash_independent(20, 32, 1)); + assert!(!hash_independent(21, 32, 1)); + assert!(l2(&univmon(1, 20, 32, 1, None)).is_some()); + assert!(l2(&univmon(1, 21, 32, 1, None)).is_none()); } /// Each distinct-count heap meets its target once the source's bound diff --git a/crates/logical-asap-optimization/src/workload_candidates.rs b/crates/logical-asap-optimization/src/workload_candidates.rs index 090f6ac98..7aba36cfc 100644 --- a/crates/logical-asap-optimization/src/workload_candidates.rs +++ b/crates/logical-asap-optimization/src/workload_candidates.rs @@ -459,7 +459,7 @@ mod tests { /// length 0 skips every choice. #[test] fn skipping_enumeration_drops_only_the_invalid_prefixes() { - let queries = vec![ + let queries = [ "sum by (job) (sum_over_time(m[5m]))".to_string(), "quantile_over_time(0.99, n[5m])".to_string(), ]; diff --git a/crates/plan-selection/src/accuracy.rs b/crates/plan-selection/src/accuracy.rs index 192a05d7c..3bed89786 100644 --- a/crates/plan-selection/src/accuracy.rs +++ b/crates/plan-selection/src/accuracy.rs @@ -44,7 +44,8 @@ pub trait AccuracyModel { /// weights not proven non-negative is rejected /// (`RejectionCause::CountMinWeightsNotProvenNonNegative`) whatever this /// returns, because the executor refuses CmsWithHeap and HydraCms when it - /// builds the plan and plain Count-Min fails at the first negative weight. The default accepts + /// builds the plan and plain Count-Min adds a negative weight, which voids + /// its never-underestimate bound. The default accepts /// none, so Stage 3 checks the Stage 1 guarantee, then the evidence /// ([`AccuracyEvidenceProvider::output_guarantee`]) with /// [`Self::satisfies`]. diff --git a/crates/plan-selection/src/lib.rs b/crates/plan-selection/src/lib.rs index e252da672..981cf7c6b 100644 --- a/crates/plan-selection/src/lib.rs +++ b/crates/plan-selection/src/lib.rs @@ -1911,7 +1911,8 @@ fn accuracy_check( // Count-Min (plain, CmsWithHeap or Hydra Count-Min) over // weights not proven non-negative does not run: the executor // refuses CmsWithHeap and HydraCms when it builds the plan, - // and plain Count-Min fails at the first negative weight. So + // and plain Count-Min adds a negative weight, which voids its + // never-underestimate bound. So // this rejects the plan whatever the deployment's accuracy // model accepts. return Err(CheckFailure { @@ -3856,7 +3857,8 @@ fn keeps_every_value(measure: &AggIntent) -> bool { /// parameters as `asap_sketchlib` allocates it: /// - Count-Min / Count-Sketch: `depth × width` 8-byte counters, plus a /// 24-byte entry per heap slot for the top-k variants; -/// - KLL: the 8-byte item array it allocates up front ([`kll_items`]); +/// - KLL: the 8-byte item array at its full capacity ([`kll_items`]), an +/// upper bound: `asap_sketchlib`'s KLL grows the array as items arrive; /// - DDSketch: one 8-byte count per bin it touches over an assumed nine /// decades of values ([`ddsketch_bins`]); /// - HLL: one 1-byte register per bucket, `2^precision`; @@ -3916,8 +3918,8 @@ fn sketch_shape(params: &SketchParams) -> (f64, u64) { } } -/// Items `asap_sketchlib`'s KLL allocates up front for `k` -/// (`KLL::init_internal` → `compute_max_capacity(k, 8)`): over its 61 levels, +/// The most items `asap_sketchlib`'s KLL holds for `k`, the capacity its item +/// array grows to (`Kll::init_internal` → `compute_max_capacity(k, 8)`): over its 61 levels, /// `max(ceil(k · (2/3)^i), 8)` each, with `k` capped at [`KLL_MAX_K`] as the /// library caps it. 1,206 items at `k` = 269. fn kll_items(k: u32) -> u64 { @@ -5682,8 +5684,9 @@ mod tests { ); } - /// KLL state is the item array `asap_sketchlib` allocates up front: - /// `compute_max_capacity(k, 8)` 8-byte items, with `k` capped at 26 602. + /// KLL state is charged the item array at the capacity `asap_sketchlib` + /// grows it to: `compute_max_capacity(k, 8)` 8-byte items, with `k` + /// capped at 26 602. #[test] fn kll_state_is_sized_as_asap_sketchlib_allocates_it() { let bytes = |k| { diff --git a/crates/plan-selection/src/pass/query_candidates.rs b/crates/plan-selection/src/pass/query_candidates.rs index b9a6c3219..e65f1ccc8 100644 --- a/crates/plan-selection/src/pass/query_candidates.rs +++ b/crates/plan-selection/src/pass/query_candidates.rs @@ -153,7 +153,7 @@ fn query_digits( }; let reachable: HashSet<*const OperatorNode> = starts .into_iter() - .flat_map(|node| OperatorNode::reachable(node)) + .flat_map(OperatorNode::reachable) .map(|node| Rc::as_ptr(&node)) .collect(); let read: Vec = per_computation diff --git a/crates/plan-selection/src/pass/stage1_to_stage3.rs b/crates/plan-selection/src/pass/stage1_to_stage3.rs index 1033449d4..659eee8b6 100644 --- a/crates/plan-selection/src/pass/stage1_to_stage3.rs +++ b/crates/plan-selection/src/pass/stage1_to_stage3.rs @@ -371,6 +371,7 @@ impl Searched { /// Keeping every candidate lists them all; otherwise only the roots of /// the Stage 1 candidates whose physical candidates Stage 3 still keeps are /// held. `queries` is each query's own-candidate report, for the selection. +#[allow(clippy::too_many_arguments)] fn search_stage2_and_stage3( per_computation: &PerComputationCandidates, stage1: impl Iterator, diff --git a/crates/planner/tests/runtime_install.rs b/crates/planner/tests/runtime_install.rs index 298881db8..981d435b9 100644 --- a/crates/planner/tests/runtime_install.rs +++ b/crates/planner/tests/runtime_install.rs @@ -1263,7 +1263,7 @@ async fn plan_p99_avg(values_non_negative: bool) -> Result, Bound); + pub(super) fn serialize( - window: &Option<(Bound, Bound)>, + window: &Option, serializer: S, ) -> Result { window @@ -136,7 +139,7 @@ mod relative_time_bounds { pub(super) fn deserialize<'de, D: Deserializer<'de>>( deserializer: D, - ) -> Result, Bound)>, D::Error> { + ) -> Result, D::Error> { let window = Option::<(WireBound, WireBound)>::deserialize(deserializer)?; Ok(window.map(|(lower, upper)| (lower.into(), upper.into()))) }