From 2dff6b98a2c294b9b8b33c106784604449b38376 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Wed, 21 Oct 2026 00:20:25 +0000 Subject: [PATCH] chore: pin asap_sketchlib to main 0.4.0 and port the executor to its APIs Pin eb04e7f (release 0.4.0) instead of planner-pin/sketch-library-fixes. WeightedFrequency wraps main's f64 CountMinSketchWithHeap / CountSketchWithHeap keyed by TupleKey bytes, with executor-side weight checks and per-update heap rescoring until sketchlib #245 is released. KLL, KMV, CMS and heap memory charges use the library's memory_bytes. Remove the never-built HydraKLL accumulator. Closes #785 --- Cargo.lock | 86 +---- Cargo.toml | 2 +- crates/executor/src/capability.rs | 7 +- .../src/operators/row_reference/mod.rs | 8 +- .../src/summary_kernels/count_min_sketch.rs | 55 ++- .../src/summary_kernels/datasketches_kll.rs | 43 ++- .../executor/src/summary_kernels/dd_sketch.rs | 84 ++++- .../executor/src/summary_kernels/factory.rs | 77 +--- .../src/summary_kernels/hll_sketch.rs | 159 +++++++- .../executor/src/summary_kernels/hydra_cms.rs | 12 +- .../executor/src/summary_kernels/hydra_kll.rs | 58 --- crates/executor/src/summary_kernels/kmv.rs | 23 +- crates/executor/src/summary_kernels/mod.rs | 2 - crates/executor/src/summary_kernels/traits.rs | 2 +- .../executor/src/summary_kernels/univmon.rs | 91 +++-- .../src/summary_kernels/weighted_frequency.rs | 346 +++++++++++++++--- crates/executor/src/values/mod.rs | 6 +- crates/executor/tests/deployment.rs | 26 +- .../executor/tests/deployment_computation.rs | 4 +- crates/executor/tests/state_column_updater.rs | 2 +- .../tests/stored_pane_memory_limit.rs | 29 +- .../tests/promql_numeric_regressions.rs | 23 +- .../tests/promql_to_post_asap.rs | 4 +- .../src/accuracy/estimators/hll.rs | 6 +- .../src/accuracy/estimators/univmon.rs | 25 +- 25 files changed, 738 insertions(+), 442 deletions(-) delete mode 100644 crates/executor/src/summary_kernels/hydra_kll.rs 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..ae03edb2a 100644 --- a/crates/executor/src/capability.rs +++ b/crates/executor/src/capability.rs @@ -8,6 +8,7 @@ //! //! Stored-state encodings belong to deployments. Full plan acceptance is //! owned by `binding`, which also validates schemas, expressions and inputs. +use crate::summary_kernels::weighted_frequency::{FrequencyAlgorithm, WeightedFrequency}; use crate::Error; use planner_types::ir::schema::{ ExactKind, ExactParams, FieldDataType as SummaryFamilyType, GroupingStrategy, SketchAlgorithm, @@ -65,8 +66,8 @@ pub fn validate_summary_kernel( // 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, + && WeightedFrequency::check_shape( + FrequencyAlgorithm::CountSketch, *width as usize, *depth as usize, ) @@ -99,7 +100,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, diff --git a/crates/executor/src/operators/row_reference/mod.rs b/crates/executor/src/operators/row_reference/mod.rs index 80c547970..505c5a132 100644 --- a/crates/executor/src/operators/row_reference/mod.rs +++ b/crates/executor/src/operators/row_reference/mod.rs @@ -108,7 +108,7 @@ pub(crate) mod harness { } /// A summary state's content: an exact state serialized, a KLL's size - /// and quantiles, a keyed frequency state serialized. + /// and quantiles, a keyed frequency state's counters and heap. pub(crate) fn state_fingerprint(state: &dyn crate::AggregateCore) -> String { use crate::summary_kernels::{ exact::ExactAccumulator, weighted_frequency::WeightedFrequency, @@ -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:?} {:?} {:?}", + 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..519382aa9 100644 --- a/crates/executor/src/summary_kernels/count_min_sketch.rs +++ b/crates/executor/src/summary_kernels/count_min_sketch.rs @@ -1,23 +1,40 @@ //! Count-Min Sketch frequency summary over `asap_sketchlib::CountMinSketch`. use crate::{AggregateCore, KernelError, KeyByLabelValues}; -use asap_sketchlib::CountMinSketch; +use asap_sketchlib::{DataInput, FastPath, Vector2D}; use planner_types::ir::schema::SketchStatistic; -#[derive(Debug, Clone)] +/// Float64 Count-Min counters, an item keyed by its text. +pub type CountMinSketchF64 = asap_sketchlib::CountMinSketch, FastPath>; + +#[derive(Clone)] pub struct CountMinSketchAccumulator { - pub inner: CountMinSketch, + pub inner: CountMinSketchF64, +} + +impl std::fmt::Debug for CountMinSketchAccumulator { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("CountMinSketchAccumulator") + .field("rows", &self.inner.rows()) + .field("cols", &self.inner.cols()) + .finish() + } } impl CountMinSketchAccumulator { pub fn new(row_num: usize, col_num: usize) -> Self { Self { - inner: CountMinSketch::new(row_num, col_num), + inner: CountMinSketchF64::with_dimensions(row_num, col_num), } } + /// Adds `weight` to `item`. + 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,27 +55,33 @@ 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(row_mass(&self.inner)), _ => Err(format!("{query:?} is not supported by Count-Min Sketch").into()), } } + /// sketchlib's count of the struct and its counters. 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()) +/// The sum of row 0's counters (rows are stored one after another). +pub(crate) fn row_mass(sketch: &CountMinSketchF64) -> f64 { + sketch.as_storage().as_slice()[..sketch.cols()].iter().sum() } #[cfg(test)] @@ -73,8 +96,8 @@ mod tests { CountMinSketchAccumulator::new(3, 128), ); 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() @@ -91,8 +114,8 @@ mod tests { value: None, }; let mut state = CountMinSketchAccumulator::new(2, 1); - state.inner.update("a", 3.0); - state.inner.update("b", 7.0); + 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); diff --git a/crates/executor/src/summary_kernels/datasketches_kll.rs b/crates/executor/src/summary_kernels/datasketches_kll.rs index f1288e070..e37f6cd05 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.inner.count() == 0 { + 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,18 @@ 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 = Kll::with_k(self.inner.k() as i32); + merged.merge(&self.inner); + merged.merge(&other.inner); + Ok(Box::new(Self { inner: merged })) } fn estimate(&self, query: &SketchStatistic) -> Result { @@ -78,8 +83,10 @@ impl AggregateCore for DatasketchesKLLAccumulator { self.inner.count() == 0 } + /// sketchlib's count of what the sketch holds now; its item buffer + /// grows with the stream. fn approx_memory_bytes(&self) -> usize { - kll_memory_bytes(self.inner.k()) + self.inner.memory_bytes() } } diff --git a/crates/executor/src/summary_kernels/dd_sketch.rs b/crates/executor/src/summary_kernels/dd_sketch.rs index 735effce7..5a93d27b3 100644 --- a/crates/executor/src/summary_kernels/dd_sketch.rs +++ b/crates/executor/src/summary_kernels/dd_sketch.rs @@ -14,6 +14,75 @@ impl DDSketchAccumulator { inner: DdSketch::new(alpha), } } + + /// Adds one sample; a non-finite or out-of-range value is ignored. + pub fn update(&mut self, value: f64) { + self.inner.add(&value); + } + + /// The lower order statistic at `q`: the bucket representative + /// `γ^i (1 + α)` of the sample at 0-based rank `⌊q (n − 1)⌋`; `None` when + /// empty. + pub fn quantile(&self, q: f64) -> Option { + let count = self.inner.get_count(); + if count == 0 { + return None; + } + self.value_at_rank((q * count.saturating_sub(1) as f64).floor() as u64) + } + + /// The linear interpolation at rank `q (n − 1)` between the two + /// adjacent order statistics' representatives, as PromQL's `quantile` + /// interpolates; `None` for `q` outside `[0, 1]`, an empty sketch or a + /// count past exact Float64 integers. Across opposite signs it need not + /// keep α relative error. + pub fn quantile_interpolated(&self, q: f64) -> Option { + let count = self.inner.get_count(); + if !(0.0..=1.0).contains(&q) || count == 0 || count > (1u64 << 53) { + return None; + } + let rank = q * (count - 1) as f64; + let lower = self.value_at_rank(rank.floor() as u64)?; + let upper = self.value_at_rank(rank.ceil() as u64)?; + let fraction = rank - rank.floor(); + let result = lower * (1.0 - fraction) + upper * fraction; + result.is_finite().then_some(result) + } + + /// The bucket representative of the sample at 0-based rank `rank`, + /// signed, from the stores sketchlib exposes. sketchlib's own + /// `get_value_at_quantile` takes the nearest rank `⌈q n⌉` clamped to the + /// sample range, a different answer this read keeps the executor from + /// changing. + pub fn value_at_rank(&self, mut rank: u64) -> Option { + let sketch = &self.inner; + let gamma = (1.0 + sketch.alpha()) / (1.0 - sketch.alpha()); + let representative = |offset: i32, i: usize| { + gamma.powf((i64::from(offset) + i as i64) as f64) * (1.0 + sketch.alpha()) + }; + let negative = sketch.negative_store_counts(); + for (i, &count) in negative.iter().enumerate().rev() { + if rank < count { + return Some(-representative(sketch.negative_store_offset(), i)); + } + rank -= count; + } + if rank < sketch.zero_count() { + return Some(0.0); + } + rank -= sketch.zero_count(); + let positive = sketch.positive_store_counts(); + let mut seen = 0u64; + for (i, &count) in positive.iter().enumerate() { + seen = seen.saturating_add(count); + if seen > rank { + return Some(representative(sketch.positive_store_offset(), i)); + } + } + // Past the last sample: the highest non-empty bucket's representative. + let last = positive.iter().rposition(|&count| count > 0)?; + Some(representative(sketch.positive_store_offset(), last)) + } } impl AggregateCore for DDSketchAccumulator { @@ -33,24 +102,23 @@ 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`. fn estimate(&self, query: &SketchStatistic) -> Result { match query { SketchStatistic::Quantile { q } if (0.0..=1.0).contains(q) => self - .inner .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 +144,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.update(f64::from(v)); + b.update(f64::from(v + 50)); } let merged = a.merge_with(&b).unwrap(); assert_eq!(merged.estimate(&bare_count()).unwrap(), 100.0); diff --git a/crates/executor/src/summary_kernels/factory.rs b/crates/executor/src/summary_kernels/factory.rs index 93324d057..87db68a77 100644 --- a/crates/executor/src/summary_kernels/factory.rs +++ b/crates/executor/src/summary_kernels/factory.rs @@ -3,7 +3,6 @@ use crate::summary_kernels::univmon::UnivMonAccumulator; use crate::summary_kernels::weighted_frequency::{FrequencyAlgorithm, WeightedFrequency}; use crate::summary_kernels::{ CountMinSketchAccumulator, DDSketchAccumulator, DatasketchesKLLAccumulator, HydraCms, - HydraKllSketchAccumulator, }; use crate::{AggregateCore, KeyByLabelValues}; use planner_types::ir::schema::{ @@ -172,8 +171,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 +204,7 @@ impl AccumulatorUpdater for DDSketchAccumulatorUpdater { } fn update_single(&mut self, value: f64, _timestamp_ms: i64) -> Result<(), String> { - self.acc.inner.update(value); + self.acc.update(value); Ok(()) } @@ -272,7 +270,7 @@ 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(()) } @@ -287,8 +285,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() } } @@ -366,59 +363,6 @@ impl AccumulatorUpdater for WeightedFrequencyUpdater { } } -// --------------------------------------------------------------------------- -// HydraKllAccumulatorUpdater -// --------------------------------------------------------------------------- - -pub struct HydraKllAccumulatorUpdater { - acc: HydraKllSketchAccumulator, - row_num: usize, - col_num: usize, - k: u16, -} - -impl HydraKllAccumulatorUpdater { - pub fn new(row_num: usize, col_num: usize, k: u16) -> Self { - Self { - acc: HydraKllSketchAccumulator::new(row_num, col_num, k), - row_num, - col_num, - k, - } - } -} - -impl AccumulatorUpdater for HydraKllAccumulatorUpdater { - fn update_single(&mut self, _value: f64, _timestamp_ms: i64) -> Result<(), String> { - Err("update_single called on a keyed updater; use update_keyed".into()) - } - - fn update_keyed( - &mut self, - key: &KeyByLabelValues, - value: f64, - _timestamp_ms: i64, - ) -> Result<(), String> { - self.acc.update(key, value); - Ok(()) - } - - impl_clone_accumulator_methods!(acc); - - fn reset(&mut self) { - self.acc = HydraKllSketchAccumulator::new(self.row_num, self.col_num, self.k); - } - - fn is_keyed(&self) -> bool { - true - } - - 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 - } -} - // --------------------------------------------------------------------------- // Config helpers // --------------------------------------------------------------------------- @@ -506,10 +450,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)) @@ -603,10 +544,7 @@ impl AccumulatorUpdater for HllUpdater { self.acc.approx_memory_bytes() } 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.insert_sample(value); Ok(()) } fn update_keyed( @@ -619,8 +557,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..9318ed3c7 100644 --- a/crates/executor/src/summary_kernels/hll_sketch.rs +++ b/crates/executor/src/summary_kernels/hll_sketch.rs @@ -1,19 +1,124 @@ -//! HyperLogLog distinct-count summary over `asap_sketchlib::HllSketch`. +//! HyperLogLog distinct-count summary over `asap_sketchlib::HyperLogLogImpl` +//! with the classic estimator. use crate::{AggregateCore, KernelError}; -use asap_sketchlib::{HllSketch, HllVariant}; +use asap_sketchlib::sketches::hll::HyperLogLogImpl; +use asap_sketchlib::{Classic, DataInput}; use planner_types::ir::schema::SketchStatistic; +/// The precisions the executor builds. +pub const HLL_PRECISIONS: std::ops::RangeInclusive = 4..=18; + +// sketchlib fixes the precision in the register type and defines P12, P14 +// and P16; the executor builds every precision in `HLL_PRECISIONS`. +asap_sketchlib::impl_hll_registers!(HllRegistersP4, 4, 1_usize << 4); +asap_sketchlib::impl_hll_registers!(HllRegistersP5, 5, 1_usize << 5); +asap_sketchlib::impl_hll_registers!(HllRegistersP6, 6, 1_usize << 6); +asap_sketchlib::impl_hll_registers!(HllRegistersP7, 7, 1_usize << 7); +asap_sketchlib::impl_hll_registers!(HllRegistersP8, 8, 1_usize << 8); +asap_sketchlib::impl_hll_registers!(HllRegistersP9, 9, 1_usize << 9); +asap_sketchlib::impl_hll_registers!(HllRegistersP10, 10, 1_usize << 10); +asap_sketchlib::impl_hll_registers!(HllRegistersP11, 11, 1_usize << 11); +asap_sketchlib::impl_hll_registers!(HllRegistersP13, 13, 1_usize << 13); +asap_sketchlib::impl_hll_registers!(HllRegistersP15, 15, 1_usize << 15); +asap_sketchlib::impl_hll_registers!(HllRegistersP17, 17, 1_usize << 17); +asap_sketchlib::impl_hll_registers!(HllRegistersP18, 18, 1_usize << 18); + +macro_rules! hll_by_precision { + ($($variant:ident = $precision:literal => $registers:ty),* $(,)?) => { + /// A classic-estimator HyperLogLog at one precision. + #[derive(Debug, Clone)] + enum Hll { + $($variant(HyperLogLogImpl),)* + } + + impl Hll { + fn new(precision: u8) -> Option { + match precision { + $($precision => Some(Self::$variant(HyperLogLogImpl::::new())),)* + _ => None, + } + } + fn precision(&self) -> u8 { + match self { + $(Self::$variant(_) => $precision,)* + } + } + fn update(&mut self, item: &DataInput) { + match self { + $(Self::$variant(hll) => hll.update(item),)* + } + } + fn estimate(&self) -> usize { + match self { + $(Self::$variant(hll) => hll.estimate(),)* + } + } + /// `None` when the precisions differ. + fn merge(&mut self, other: &Self) -> Option<()> { + match (self, other) { + $((Self::$variant(hll), Self::$variant(other)) => hll.merge(other),)* + _ => return None, + } + Some(()) + } + } + }; +} + +hll_by_precision! { + P4 = 4 => HllRegistersP4, + P5 = 5 => HllRegistersP5, + P6 = 6 => HllRegistersP6, + P7 = 7 => HllRegistersP7, + P8 = 8 => HllRegistersP8, + P9 = 9 => HllRegistersP9, + P10 = 10 => HllRegistersP10, + P11 = 11 => HllRegistersP11, + P12 = 12 => asap_sketchlib::HllRegistersP12, + P13 = 13 => HllRegistersP13, + P14 = 14 => asap_sketchlib::HllRegistersP14, + P15 = 15 => HllRegistersP15, + P16 = 16 => asap_sketchlib::HllRegistersP16, + P17 = 17 => HllRegistersP17, + P18 = 18 => HllRegistersP18, +} + #[derive(Debug, Clone)] pub struct HllSketchAccumulator { - pub inner: HllSketch, + inner: Hll, } 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 { + Ok(Self { + inner: Hll::new(precision) + .ok_or_else(|| format!("HLL precision {precision} is outside 4..=18"))?, + }) + } + + pub fn precision(&self) -> u8 { + self.inner.precision() + } + + /// Observes one sample value, hashed by its bits with `-0.0` folded into + /// `0.0`; NaN is not an item. + 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.update(&DataInput::Bytes(&bits.to_le_bytes())); } } + + /// Observes one item given as bytes. + pub fn insert_bytes(&mut self, item: &[u8]) { + self.inner.update(&DataInput::Bytes(item)); + } + + /// Empties the sketch, keeping its precision. + pub fn clear(&mut self) { + self.inner = Hll::new(self.precision()).expect("a built precision"); + } } impl AggregateCore for HllSketchAccumulator { @@ -33,23 +138,28 @@ 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(); + merged + .inner + .merge(&other.inner) + .ok_or("HLL merges only with the same precision")?; + Ok(Box::new(merged)) } - /// Distinct count. A bare `PointCount` over an HLL also reads the distinct count. + /// Distinct count, rounded down to an integer by sketchlib. 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()), } } + /// The struct and its `2^precision` one-byte registers. fn approx_memory_bytes(&self) -> usize { - std::mem::size_of::().saturating_add(self.inner.registers.capacity()) + std::mem::size_of::().saturating_add(1 << self.precision()) } } @@ -61,22 +171,37 @@ 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.insert_bytes(&v.to_le_bytes()); + b.insert_bytes(&(v + 500).to_le_bytes()); } let merged = a.merge_with(&b).unwrap(); let estimate = merged.estimate(&SketchStatistic::Cardinality).unwrap(); assert!((estimate - 1500.0).abs() / 1500.0 < 0.05, "{estimate}"); + assert!(a + .merge_with(&HllSketchAccumulator::new(13).unwrap()) + .is_err()); + } + + // Every precision in 4..=18 builds, and no other. + #[test] + fn builds_exactly_the_supported_precisions() { + for precision in 0..=20u8 { + let built = HllSketchAccumulator::new(precision); + assert_eq!(built.is_ok(), HLL_PRECISIONS.contains(&precision)); + if let Ok(hll) = built { + assert_eq!(hll.precision(), precision); + } + } } // 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()); } } diff --git a/crates/executor/src/summary_kernels/hydra_cms.rs b/crates/executor/src/summary_kernels/hydra_cms.rs index 811920ce8..b45690a29 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, @@ -52,8 +52,8 @@ 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.inner.rows, + self.inner.cols, self.width, self.depth, ) diff --git a/crates/executor/src/summary_kernels/hydra_kll.rs b/crates/executor/src/summary_kernels/hydra_kll.rs deleted file mode 100644 index 8d6f6a795..000000000 --- a/crates/executor/src/summary_kernels/hydra_kll.rs +++ /dev/null @@ -1,58 +0,0 @@ -use crate::{AggregateCore, KeyByLabelValues}; -use asap_sketchlib::HydraKllSketch; - -/// HydraKLL (shared-grouping quantiles) over `asap_sketchlib::HydraKllSketch`. -#[derive(Debug, Clone)] -pub struct HydraKllSketchAccumulator { - pub inner: HydraKllSketch, -} - -impl HydraKllSketchAccumulator { - pub fn new(row_num: usize, col_num: usize, k: u16) -> Self { - Self { - inner: HydraKllSketch::new(row_num, col_num, k), - } - } - - pub fn update(&mut self, key: &KeyByLabelValues, value: f64) { - self.inner.update(&key.to_semicolon_str(), value); - } - - pub fn query_key(&self, key: &KeyByLabelValues, quantile: f64) -> f64 { - self.inner.quantile(&key.to_semicolon_str(), quantile) - } -} - -impl AggregateCore for HydraKllSketchAccumulator { - 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 hk = other - .as_any() - .downcast_ref::() - .ok_or("Failed to downcast to HydraKllSketchAccumulator")?; - - let mut merged = self.clone(); - merged.inner.merge(&hk.inner)?; - Ok(Box::new(merged)) - } - - 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 - } -} diff --git a/crates/executor/src/summary_kernels/kmv.rs b/crates/executor/src/summary_kernels/kmv.rs index 82e9adbc1..95d330985 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 with the library's shared `DEFAULT_CANONICAL_SEED_INDEX` seed, 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())); } } } @@ -72,8 +72,9 @@ impl AggregateCore for KmvAccumulator { } } + /// sketchlib's count of the struct and the hash slots it holds. fn approx_memory_bytes(&self) -> usize { - self.inner.approx_memory_bytes() + self.inner.memory_bytes() } } @@ -116,7 +117,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 +173,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 +190,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 +229,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/mod.rs b/crates/executor/src/summary_kernels/mod.rs index 4d7787cc5..6b8b1a5c6 100644 --- a/crates/executor/src/summary_kernels/mod.rs +++ b/crates/executor/src/summary_kernels/mod.rs @@ -5,7 +5,6 @@ pub mod dd_sketch; pub mod exact; pub mod hll_sketch; pub mod hydra_cms; -pub mod hydra_kll; pub mod increase; pub mod kmv; pub mod univmon; @@ -15,7 +14,6 @@ pub use datasketches_kll::*; pub use dd_sketch::*; pub use hll_sketch::*; pub use hydra_cms::*; -pub use hydra_kll::*; pub use increase::*; pub use kmv::*; diff --git a/crates/executor/src/summary_kernels/traits.rs b/crates/executor/src/summary_kernels/traits.rs index 2bbf707ea..6694e36e7 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.update(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..839aa2fb5 100644 --- a/crates/executor/src/summary_kernels/univmon.rs +++ b/crates/executor/src/summary_kernels/univmon.rs @@ -1,12 +1,21 @@ //! One frequency state shared by count, distinct, L2 and entropy evaluations. use crate::AggregateCore; -use asap_sketchlib::{DataInput, UnivMon}; +use asap_sketchlib::{DataInput, OwnedDataInput, UnivMon, MATRIX_MAX_ROWS}; use planner_types::ir::scalar::ColumnRef; 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 per layer so the state encodes. +pub fn check_shape(heap_size: usize, rows: usize, cols: usize, layers: usize) -> Result<(), Error> { + if rows > MATRIX_MAX_ROWS { + return Err(format!("UnivMon rows {rows} exceeds MATRIX_MAX_ROWS {MATRIX_MAX_ROWS}").into()); + } + Ok(UnivMon::check_shape(heap_size, rows, cols, layers)?) +} + #[derive(Debug, Clone)] pub struct UnivMonAccumulator { inner: UnivMon, @@ -24,15 +33,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 +51,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(), + OwnedDataInput::String(key) => key.capacity(), + OwnedDataInput::Bytes(key) => key.capacity(), _ => 0, }) .collect(); @@ -79,11 +83,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,35 +105,25 @@ 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(()) } fn compatible(&self, other: &Self) -> bool { - ( - self.inner.heap_size, - self.inner.sketch_row, - self.inner.sketch_col, - self.inner.layer_size, - ) == ( - other.inner.heap_size, - other.inner.sketch_row, - other.inner.sketch_col, - other.inner.layer_size, - ) + self.dimensions() == other.dimensions() } 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 +132,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 +152,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 +160,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 +196,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,14 +258,15 @@ 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; + // `new` refuses 21 rows; a state claiming them is refused too. + assert!(UnivMonAccumulator::new(4, 21, 2, 2).is_err()); + let mut misshapen = UnivMon::new(4, 3, 16, 2); + misshapen.rows = 21; assert!(UnivMonAccumulator::from_sketch(misshapen).is_err()); } diff --git a/crates/executor/src/summary_kernels/weighted_frequency.rs b/crates/executor/src/summary_kernels/weighted_frequency.rs index b3a5e4b8e..52c155a08 100644 --- a/crates/executor/src/summary_kernels/weighted_frequency.rs +++ b/crates/executor/src/summary_kernels/weighted_frequency.rs @@ -1,27 +1,64 @@ -//! ASAP type and trait adapter for sketchlib's Float64 weighted frequency kernel. +//! Keyed Float64 frequency state over sketchlib's f64 Count-Min and Count +//! Sketch with a candidate heap (`CountMinSketchWithHeap`, +//! `CountSketchWithHeap`), keyed by the item tuple's canonical `TupleKey` +//! bytes. A heap of capacity 0 makes a plain Count Sketch. //! -//! 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. +//! The sketches hash each row with its own seed (`RegularPath`). Until +//! sketchlib releases its checked update, this adapter enforces the weight +//! rule (finite, and nonnegative for Count-Min) and keeps every counter +//! finite, and it rescores the heap's residents after each update, so the +//! heap holds the current top-k by the current estimates. 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::{ + CountMinSketchWithHeap, CountSketchWithHeap, DataInput, KeyColumn, OwnedDataInput, + RegularPath, TopKHeap, TupleKey, Vector2D, +}; -fn adapt_error(error: WeightedFrequencyError) -> Error { - match error { - WeightedFrequencyError::Invalid(message) => Error::Invalid(message), - WeightedFrequencyError::Update(message) => Error::Operator(message), +/// The counter matrix of a [`WeightedFrequency`] state. +#[derive(Copy, Clone, Debug, PartialEq, Eq)] +pub enum FrequencyAlgorithm { + /// Count-Min: nonnegative weights, the minimum over rows. + Cms, + /// Count Sketch: signed weights, the median over an odd number of rows. + CountSketch, +} + +#[derive(Clone)] +enum Sketch { + Cms(CountMinSketchWithHeap, RegularPath>), + CountSketch(CountSketchWithHeap, RegularPath>), +} + +#[derive(Clone)] +pub struct WeightedFrequency { + sketch: Sketch, + /// The sum of every added weight's magnitude, merges included. No + /// counter's magnitude exceeds it, so while it is finite every counter is. + weight_mass: 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()) + .finish() } } -fn identity(value: &Value) -> Result { + +fn key_column(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()), + Value::Null => KeyColumn::Null, + Value::Bool(v) => KeyColumn::Bool(*v), + Value::Int64(v) => KeyColumn::I64(*v), + Value::Float64(v) if v.is_finite() => KeyColumn::F64(*v), + Value::Float64(_) => { + return Err(Error::Invalid( + "non-finite weighted frequency identity".into(), + )) + } + Value::Utf8(v) => KeyColumn::String(v.to_string()), _ => { return Err(Error::Invalid( "unsupported weighted frequency identity".into(), @@ -29,20 +66,41 @@ fn identity(value: &Value) -> Result { } }) } -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()), + +fn value(column: &KeyColumn) -> Value { + 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()), } } -#[derive(Clone, Debug, Serialize, Deserialize)] -#[serde(transparent)] -pub struct WeightedFrequency { - inner: Kernel, + +fn tuple_key(columns: Vec) -> Result, Error> { + TupleKey::new(columns).to_bytes().map_err(Error::Invalid) +} + +/// Writes each resident's current estimate back to the heap: +/// `update_owned` re-sifts a resident and never evicts one. +fn rescore(heap: &mut TopKHeap, scores: Vec<(OwnedDataInput, f64)>) { + for (key, count) in &scores { + heap.update_owned(key, *count); + } } + +/// The current estimate of every resident of `$sketch`'s heap. +macro_rules! resident_scores { + ($sketch:expr) => { + $sketch + .heap() + .heap() + .iter() + .map(|entry| (entry.key.clone(), $sketch.estimate(&entry.key.as_data_input()))) + .collect::>() + }; +} + impl WeightedFrequency { pub(crate) fn configuration( kind: &planner_types::ir::schema::SketchKind, @@ -71,24 +129,50 @@ 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)?; + Self::check_shape(algorithm, width, depth)?; Ok((algorithm, width, depth, capacity as usize)) } + /// The shapes this state builds: sketchlib's (both dimensions non-zero, + /// the matrix within `usize`), and an odd depth for Count Sketch, whose + /// median the guarantee assumes. Any depth: past `MATRIX_MAX_ROWS` rows + /// sketchlib derives each row's seed from its list, so rows stay + /// independent (the state then has no ASAPv1 encoding, which the + /// executor does not use). + pub fn check_shape( + algorithm: FrequencyAlgorithm, + width: usize, + depth: usize, + ) -> Result<(), Error> { + asap_sketchlib::CountMinSketch::check_shape(depth, width).map_err(Error::Invalid)?; + if algorithm == FrequencyAlgorithm::CountSketch && depth % 2 == 0 { + return Err(Error::Invalid( + "Count Sketch depth must be odd for its median".into(), + )); + } + Ok(()) + } + /// 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(_) => FrequencyAlgorithm::Cms, + Sketch::CountSketch(_) => FrequencyAlgorithm::CountSketch, + } } pub fn shape(&self) -> (usize, usize, usize) { - self.inner.shape() + match &self.sketch { + Sketch::Cms(s) => (s.cols(), s.rows(), s.heap().capacity()), + Sketch::CountSketch(s) => (s.cols(), s.rows(), s.heap().capacity()), + } } pub fn new( algorithm: FrequencyAlgorithm, @@ -96,48 +180,137 @@ impl WeightedFrequency { depth: usize, capacity: usize, ) -> Result { - Kernel::new(algorithm, width, depth, capacity) - .map(|inner| Self { inner }) - .map_err(adapt_error) + Self::check_shape(algorithm, width, depth)?; + let sketch = match algorithm { + FrequencyAlgorithm::Cms => Sketch::Cms(CountMinSketchWithHeap::new(depth, width, capacity)), + FrequencyAlgorithm::CountSketch => { + Sketch::CountSketch(CountSketchWithHeap::new(depth, width, capacity)) + } + }; + Ok(Self { + sketch, + weight_mass: 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 - .iter() - .map(|label| FrequencyIdentity::Utf8(label.clone())) - .collect::>(); - self.inner.update(&identity, weight).map_err(adapt_error) + let key = tuple_key(labels.iter().cloned().map(KeyColumn::String).collect())?; + 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<(), Error> { + if !weight.is_finite() || (self.algorithm() == FrequencyAlgorithm::Cms && weight < 0.0) { + return Err(Error::Operator( + "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(), + } + self.weight_mass = 0.0; } 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 = tuple_key(values.iter().map(key_column).collect::>()?)?; + self.update_key(&key, weight) + } + /// Adds `weight` to `key`, rescores the heap's residents and offers `key` + /// to the heap. A rejected weight or a sum past `f64::MAX` leaves the + /// state unchanged. + fn update_key(&mut self, key: &[u8], weight: f64) -> Result<(), Error> { + self.check_weight(weight)?; + let weight_mass = self.weight_mass + weight.abs(); + if !weight_mass.is_finite() { + return Err(Error::Operator("weighted frequency sum overflow".into())); + } + self.weight_mass = weight_mass; + let key = DataInput::Bytes(key); + match &mut self.sketch { + Sketch::Cms(s) => { + s.count_min_sketch_mut().update_weighted(&key, weight); + let scores = resident_scores!(s); + rescore(s.heap_mut(), scores); + let estimate = s.estimate(&key); + s.heap_mut().update(&key, estimate); + } + Sketch::CountSketch(s) => { + s.count_sketch_mut().update_weighted(&key, weight); + let scores = resident_scores!(s); + rescore(s.heap_mut(), scores); + let estimate = s.estimate(&key); + s.heap_mut().update(&key, estimate); + } + } + Ok(()) } /// 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) + let key = tuple_key(values.iter().map(key_column).collect::>()?)?; + let key = DataInput::Bytes(&key); + Ok(match &self.sketch { + Sketch::Cms(s) => s.estimate(&key), + Sketch::CountSketch(s) => s.estimate(&key), + }) } + /// The heap's `n` highest-scoring candidates, highest first (ties by + /// key bytes), each as its identity values followed by its score. pub fn rows(&self, n: usize) -> Vec> { - self.inner - .topk(n) + let heap = match &self.sketch { + Sketch::Cms(s) => s.heap().heap(), + Sketch::CountSketch(s) => s.heap().heap(), + }; + let mut entries: Vec<(&[u8], f64)> = heap + .iter() + .map(|entry| match &entry.key { + OwnedDataInput::Bytes(key) => (key.as_slice(), entry.count), + _ => unreachable!("weighted frequency heap keys are tuple-key bytes"), + }) + .collect(); + entries.sort_by(|a, b| b.1.total_cmp(&a.1).then_with(|| a.0.cmp(b.0))); + entries .into_iter() - .map(|(items, score)| { - let mut row = items.into_iter().map(value).collect::>(); + .take(n) + .map(|(key, score)| { + let tuple = TupleKey::from_bytes(key).expect("heap keys are encoded tuple keys"); + let mut row = tuple.columns().iter().map(value).collect::>(); row.push(Value::Float64(score)); row }) .collect() } + /// The counters, `depth` rows of `width` one after another, so a test + /// can check per-row hashing or compare two states cell by cell. + pub fn counters(&self) -> &[f64] { + match &self.sketch { + Sketch::Cms(s) => s.count_min_sketch().as_storage().as_slice(), + Sketch::CountSketch(s) => s.count_sketch().as_storage().as_slice(), + } + } + fn merge(&self, other: &Self) -> Result { + // sketchlib's merge panics on a shape mismatch. + if self.algorithm() != other.algorithm() || self.shape() != other.shape() { + return Err(Error::Invalid("weighted frequency shape mismatch".into())); + } + let weight_mass = self.weight_mass + other.weight_mass; + if !weight_mass.is_finite() { + return Err(Error::Operator("weighted frequency merge overflow".into())); + } + let mut merged = self.clone(); + merged.weight_mass = weight_mass; + // Both merges re-estimate every candidate of both heaps. + match (&mut merged.sketch, &other.sketch) { + (Sketch::Cms(s), Sketch::Cms(o)) => s.merge(o), + (Sketch::CountSketch(s), Sketch::CountSketch(o)) => s.merge(o), + _ => unreachable!("the algorithms match"), + } + Ok(merged) + } } impl AggregateCore for WeightedFrequency { fn clone_boxed_core(&self) -> Box { @@ -157,12 +330,15 @@ 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)?)) } + /// sketchlib's count of the struct, the counters and the heap. 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(), + } } } @@ -170,7 +346,7 @@ impl AggregateCore for WeightedFrequency { mod tests { use super::*; - // Merge uses the same Float64 state representation and rejects other shapes. + // Merge adds fractional weights exactly and rejects other shapes. #[test] fn compatible_merge_preserves_fractional_weights() { let mut left = WeightedFrequency::new(FrequencyAlgorithm::Cms, 4096, 5, 8).unwrap(); @@ -184,4 +360,60 @@ mod tests { .merge_with(&WeightedFrequency::new(FrequencyAlgorithm::Cms, 32, 5, 8).unwrap()) .is_err()); } + + // Count Sketch keeps signed weights per typed identity, and a rejected + // weight leaves the state unchanged. + #[test] + fn count_sketch_signed_updates_and_rejected_weights() { + let mut state = + WeightedFrequency::new(FrequencyAlgorithm::CountSketch, 4096, 5, 8).unwrap(); + state.update(&[Value::Int64(1)], -10.5).unwrap(); + state.update(&[Value::Null], 0.125).unwrap(); + state.update(&[Value::Null], 0.1875).unwrap(); + let rows = state.rows(2); + assert!(matches!(rows[0][0], Value::Null)); + assert!(matches!(rows[0][1], Value::Float64(0.3125))); + assert!(matches!(rows[1][1], Value::Float64(-10.5))); + let before = state.counters().to_vec(); + for weight in [f64::NAN, f64::INFINITY, f64::NEG_INFINITY] { + assert!(state.update(&[Value::Null], weight).is_err()); + assert_eq!(state.counters(), before.as_slice()); + } + assert!(state.update(&[Value::Float64(f64::NAN)], 1.0).is_err()); + } + + // Count-Min rejects negative weights, and a sum past f64::MAX is refused + // before it reaches a counter. + #[test] + fn cms_rejects_negative_weights_and_overflow() { + let mut state = WeightedFrequency::new(FrequencyAlgorithm::Cms, 64, 3, 4).unwrap(); + assert!(state.update(&[Value::Int64(1)], -1.0).is_err()); + state.update(&[Value::Int64(1)], f64::MAX).unwrap(); + assert!(state.update(&[Value::Int64(2)], f64::MAX).is_err()); + assert!(state.counters().iter().all(|c| c.is_finite())); + assert!(state.merge_with(&state).is_err()); + } + + // A heap resident's score follows later updates that share its cells. + #[test] + fn residents_are_rescored_after_every_update() { + // One column: every key shares every cell. + let mut state = WeightedFrequency::new(FrequencyAlgorithm::Cms, 1, 1, 2).unwrap(); + state.update(&[Value::Int64(1)], 1.0).unwrap(); + state.update(&[Value::Int64(2)], 2.0).unwrap(); + let scores: Vec<_> = state.rows(2).into_iter().map(|row| row[1].clone()).collect(); + assert!(matches!(scores[..], [Value::Float64(3.0), Value::Float64(3.0)])); + } + + // Shapes: odd Count Sketch depth, any depth past MATRIX_MAX_ROWS (the + // planner sizes Count Sketch heaps at depth 85), no zero dimension. + #[test] + fn shape_rules() { + use FrequencyAlgorithm::*; + assert!(WeightedFrequency::check_shape(CountSketch, 8, 2).is_err()); + assert!(WeightedFrequency::check_shape(Cms, 8, 2).is_ok()); + assert!(WeightedFrequency::check_shape(CountSketch, 8, 85).is_ok()); + assert!(WeightedFrequency::check_shape(Cms, 0, 3).is_err()); + assert!(WeightedFrequency::check_shape(Cms, 8, 0).is_err()); + } } 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..b047dbf57 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); @@ -237,8 +236,9 @@ fn heap_sketch( FieldDataType::Sketch(SketchKind::new(algorithm, params), Default::default()) } -// The factory's top-k sketches use the same per-row-seeded kernel: past 20 -// rows (where a fixed seed list would repeat) every row stays independent. +// The factory's top-k sketches hash each row with its own seed: past 20 rows +// sketchlib derives further seeds from its list, and every row stays +// independent. #[test] fn deep_heap_sketch_rows_hash_independently() { for algorithm in [ @@ -296,7 +296,7 @@ fn heap_sketch_states_load_as_dag_summary_values() { #[test] fn factory_and_dag_heap_states_are_interchangeable() { use asap_executor::{summary_kernels::weighted_frequency::WeightedFrequency, values::Value}; - use asap_sketchlib::FrequencyAlgorithm; + use asap_executor::summary_kernels::weighted_frequency::FrequencyAlgorithm; let estimate = |state: &dyn AggregateCore, i: usize| { state .as_any() @@ -377,14 +377,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..c7fa877fb 100644 --- a/crates/executor/tests/deployment_computation.rs +++ b/crates/executor/tests/deployment_computation.rs @@ -668,8 +668,8 @@ fn stored_count_min_bare_count_compiles_to_a_evaluation() { 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); + 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..6423567cf 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:?} {:?} {:?}", 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. diff --git a/crates/executor/tests/stored_pane_memory_limit.rs b/crates/executor/tests/stored_pane_memory_limit.rs index c7fd9e18f..f1af245f5 100644 --- a/crates/executor/tests/stored_pane_memory_limit.rs +++ b/crates/executor/tests/stored_pane_memory_limit.rs @@ -16,14 +16,17 @@ use std::sync::Arc; const SERIES: u32 = 1_000; const PANES: usize = 12; +/// Samples per series in a pane. +const SAMPLES: u32 = 1_000; -/// One pane of `SERIES` KLL states (k = 200, about 11 kB charged each, so -/// about 11 MB per pane), keyed by series. +/// One pane of `SERIES` KLL states (k = 200, about 5.6 kB charged each, so +/// about 5.6 MB per pane), keyed by series. fn pane() -> (Arc, Batch) { pane_of(SERIES) } -/// One pane of `series` KLL states (k = 200), keyed by series. +/// One pane of `series` KLL states (k = 200, `SAMPLES` samples each), keyed +/// by series. fn pane_of(series_count: u32) -> (Arc, Batch) { let family = FieldDataType::Sketch( SketchKind::new( @@ -56,8 +59,14 @@ fn pane_of(series_count: u32) -> (Arc, Batch) { }); let mut updater = StateColumnUpdater::new(&family).unwrap(); updater.ensure_groups(series_count as usize).unwrap(); + // Enough samples that each KLL's item buffer, which grows with the + // stream, reaches its full size. 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 * SAMPLES + sample), 0) + .unwrap(); + } } let names = (0..series_count) .map(|series| format!("s{series}")) @@ -114,15 +123,15 @@ fn merge_stored_panes(max_bytes: usize) -> Result { }) } -// Twelve stored panes hold about 130 MB of states, which charging them per +// Twelve stored panes hold about 67 MB of states, which charging them per // read would put far past a 32 MiB limit. The merge allocates one state per -// series (about 11 MB), so it fits. +// series (about 6 MB), so it fits. #[test] fn reading_stored_panes_does_not_charge_their_states() { assert_eq!(merge_stored_panes(32 << 20).unwrap(), SERIES as usize); } -// The merged states themselves (about 11 MB) do not fit in 4 MiB. +// The merged states themselves (about 6 MB) do not fit in 4 MiB. #[test] fn a_merge_that_allocates_past_the_limit_still_fails() { assert!(matches!( @@ -131,7 +140,7 @@ fn a_merge_that_allocates_past_the_limit_still_fails() { )); } -// A Limit keeping 49% of a stored pane of 2,000 KLLs (about 21 MB of +// A Limit keeping 49% of a stored pane of 2,000 KLLs (about 11 MB of // states) compacts its state column, and the compacted column shares the // stored sketches by pointer: under an 8 MiB limit it is charged for the // pointers, not the sketches. Keeping 51% (no compaction) fits too. @@ -176,7 +185,7 @@ fn compacting_a_stored_pane_does_not_charge_its_shared_sketches() { } // topk over several panes: a Sort concatenates two stored panes of 1,000 -// KLLs (about 21 MB of states), then a Limit keeps 49% and compacts the +// KLLs (about 11 MB of states), then a Limit keeps 49% and compacts the // concatenated column. Both derivations share the stored sketches by // pointer, so the run fits in 8 MiB. #[test] @@ -225,7 +234,7 @@ fn sorting_then_limiting_stored_panes_does_not_charge_their_sketches() { assert_eq!(rows.unwrap(), 980); } -// A merge over one stored pane of 2,000 KLLs (about 21 MB of states), one +// A merge over one stored pane of 2,000 KLLs (about 11 MB of states), one // state per series: every state passes through unchanged, shared by // pointer, so the merge fits in 8 MiB. #[test] diff --git a/crates/integration-tests/tests/promql_numeric_regressions.rs b/crates/integration-tests/tests/promql_numeric_regressions.rs index 6d93b1a8a..0efdb8ae7 100644 --- a/crates/integration-tests/tests/promql_numeric_regressions.rs +++ b/crates/integration-tests/tests/promql_numeric_regressions.rs @@ -123,9 +123,9 @@ fn checked_ratio_must_not_certify_cross_zero_interpolation() { // even though the quantile interpolation does not preserve relative error. let alpha = (0.01 - 8.0 * f64::EPSILON) / 2.01; let estimate = |q| { - let mut sketch = asap_sketchlib::DdSketch::new(alpha); + let mut sketch = asap_executor::summary_kernels::DDSketchAccumulator::new(alpha); for value in [-1., 1.011] { - sketch.try_update(value).unwrap(); + sketch.update(value); } sketch.quantile_interpolated(q).unwrap() }; @@ -238,26 +238,21 @@ 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 mut cs = - asap_sketchlib::message_pack_format::portable::countsketch::CountSketch::new( - 5, 128, - ); + use asap_sketchlib::{CountMinSketch, CountSketch, DataInput, RegularPath, Vector2D}; + let mut cms = CountMinSketch::, RegularPath>::with_dimensions(5, 128); + let mut cs = CountSketch::, RegularPath>::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(&DataInput::Str("a"), weight); } else { - cs.update("a", weight); + cs.update_weighted(&DataInput::Str("a"), weight); } } let got = if is_count { - cms.estimate(&asap_sketchlib::DataInput::String("a".into())) + cms.estimate(&DataInput::Str("a")) } else { - cs.estimate("a") + cs.estimate(&DataInput::Str("a")) }; 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 f8666f782..c9054f37d 100644 --- a/crates/integration-tests/tests/promql_to_post_asap.rs +++ b/crates/integration-tests/tests/promql_to_post_asap.rs @@ -833,9 +833,9 @@ fn ddsketch_ratio_bound_holds_for_signed_pinned_sketch_evaluations() { sorted[r.floor() as usize] * (1. - r.fract()) + sorted[r.ceil() as usize] * r.fract() }; - let mut sketch = asap_sketchlib::DdSketch::new(alpha); + let mut sketch = asap_executor::summary_kernels::DDSketchAccumulator::new(alpha); for v in values { - sketch.try_update(v).unwrap(); + sketch.update(v); } let want = exact(0.9) / exact(0.5); let got = sketch.quantile_interpolated(0.9).unwrap() 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..63c5c5ce5 100644 --- a/crates/logical-asap-optimization/src/accuracy/estimators/univmon.rs +++ b/crates/logical-asap-optimization/src/accuracy/estimators/univmon.rs @@ -1,6 +1,6 @@ //! 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`, for `estimate_l2` and `estimate_cardinality`): //! - 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,7 +20,7 @@ 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; +pub(crate) const MAX_LAYERS: u8 = asap_sketchlib::sketch_framework::univmon::UNIVMON_MAX_LAYERS 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; @@ -30,7 +30,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 +91,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 its +/// 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() }