tisonkun commented on code in PR #204: URL: https://github.com/apache/datasketches-rust/pull/204#discussion_r3814835517
########## datasketches/src/req/sketch.rs: ########## @@ -0,0 +1,833 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! REQ sketch — generic over `T: ReqValue`. + +use std::fmt; + +use super::DEFAULT_K; +use super::MAX_K; +use super::MIN_K; +use super::RankAccuracy; +use super::SearchCriteria; +use super::compactor::Compactor; +use super::iter::ReqSketchIterator; +use super::sorted_view::SortedView; +use super::value::ReqValue; +use crate::error::Error; + +/// A Relative Error Quantiles sketch for approximate quantile estimation. +/// +/// See the [module-level documentation](super) for background. +#[derive(Debug, Clone)] +pub struct ReqSketch<T: ReqValue> { + pub(super) k: u16, + pub(super) rank_accuracy: RankAccuracy, + pub(super) n: u64, + pub(super) max_nom_size: u32, + pub(super) num_retained: u32, + pub(super) compactors: Vec<Compactor<T>>, + pub(super) promotion_buf: Vec<T>, + pub(super) min_item: Option<T>, + pub(super) max_item: Option<T>, +} + +impl<T: ReqValue> ReqSketch<T> { + /// Creates a new sketch with default parameters (`k = 12`, `RankAccuracy::HighRank`). + pub fn new() -> Self { + let mut s = Self { + k: DEFAULT_K, + rank_accuracy: RankAccuracy::HighRank, + n: 0, + max_nom_size: 0, + num_retained: 0, + compactors: Vec::new(), + promotion_buf: Vec::with_capacity(DEFAULT_K as usize), + min_item: None, + max_item: None, + }; + // C++ parity: an empty sketch has a level-0 compactor present from the start. + // This makes is_raw_items() and flags_byte() byte-compatible with the C++/Java + // wire format for the empty case. + s.grow(); + s + } + + /// Creates a new sketch with the given `k` and rank accuracy. + /// + /// # Errors + /// + /// Returns an error if `k` is odd or outside `[MIN_K, MAX_K]`. + pub fn try_new(k: u16, rank_accuracy: RankAccuracy) -> Result<Self, Error> { + if !(MIN_K..=MAX_K).contains(&k) { + return Err(Error::invalid_argument(format!( + "k must be in [{}, {}], got {k}", + MIN_K, MAX_K + ))); + } + if k % 2 != 0 { + return Err(Error::invalid_argument(format!("k must be even, got {k}"))); + } + let mut s = Self { + k, + rank_accuracy, + n: 0, + max_nom_size: 0, + num_retained: 0, + compactors: Vec::new(), + promotion_buf: Vec::with_capacity(k as usize), + min_item: None, + max_item: None, + }; + s.grow(); + Ok(s) + } + + /// Returns a builder for chained configuration. + pub fn builder() -> ReqSketchBuilder<T> { + ReqSketchBuilder::new() + } + + /// Returns the configured `k` parameter. + pub fn k(&self) -> u16 { + self.k + } + + /// Returns the configured rank accuracy. + pub fn rank_accuracy(&self) -> RankAccuracy { + self.rank_accuracy + } + + /// Returns the total number of items observed (matches C++ `get_n`). + pub fn n(&self) -> u64 { + self.n + } + + /// Returns true if the sketch has observed no items. + pub fn is_empty(&self) -> bool { + self.n == 0 + } + + /// Returns true if compaction has occurred. + pub fn is_estimation_mode(&self) -> bool { + self.compactors.len() > 1 + } + + /// Returns the number of items currently stored across all compactors. + pub fn num_retained(&self) -> u32 { + self.num_retained + } + + /// Returns the smallest item ever observed, or `None` if empty. + pub fn min_item(&self) -> Option<&T> { + self.min_item.as_ref() + } + + /// Returns the largest item ever observed, or `None` if empty. + pub fn max_item(&self) -> Option<&T> { + self.max_item.as_ref() + } + + /// Updates the sketch with a new item. + /// + /// NaN inputs are silently ignored for floating-point types, matching the behavior + /// of the Java reference implementation (`checkNaNUpdate`). This is intentional and + /// documented in the cross-language differences doc. + pub fn update(&mut self, item: T) { + if item.is_nan() { + return; + } + match &mut self.min_item { + None => self.min_item = Some(item.clone()), + Some(cur) if item.total_cmp(cur).is_lt() => *cur = item.clone(), + _ => {} + } + match &mut self.max_item { + None => self.max_item = Some(item.clone()), + Some(cur) if item.total_cmp(cur).is_gt() => *cur = item.clone(), + _ => {} + } + + self.compactors[0].append(item); + self.n += 1; + self.num_retained += 1; + + if self.num_retained == self.max_nom_size { + self.compress(); + } + } + + /// Resets the sketch to the empty state. + pub fn reset(&mut self) { + self.n = 0; + self.num_retained = 0; + self.max_nom_size = 0; + self.min_item = None; + self.max_item = None; + self.compactors.clear(); + self.grow(); + } + + /// Returns an iterator over `(item, weight)` pairs. + pub fn iter(&self) -> ReqSketchIterator<'_, T> { + ReqSketchIterator::new(&self.compactors) + } + + /// Returns the approximate rank of `item` in `[0.0, 1.0]`. + /// + /// Computed directly from the retained items in a single `O(retained)` pass, + /// without building a sorted view. The result is identical to + /// [`SortedView::rank`] on [`Self::sorted_view`]. + /// + /// # Errors + /// Returns an error if the sketch is empty or `item` is NaN. + pub fn rank(&self, item: &T, criteria: SearchCriteria) -> Result<f64, Error> { + if self.is_empty() { + return Err(Error::invalid_argument("sketch is empty")); + } + if item.is_nan() { + return Err(Error::invalid_argument("query item is NaN")); + } + let inclusive = matches!(criteria, SearchCriteria::Inclusive); + let weight: u64 = self + .compactors + .iter() + .map(|c| c.count_below(item, inclusive) as u64 * c.weight()) + .sum(); + Ok(weight as f64 / self.n as f64) + } + + /// Returns the approximate quantile at the given normalized rank. + /// + /// Builds a transient [`SortedView`] internally. For repeated quantile + /// queries, take one snapshot with [`Self::sorted_view`] and query it. + pub fn quantile(&self, rank: f64, criteria: SearchCriteria) -> Result<T, Error> { + if self.is_empty() { + return Err(Error::invalid_argument("sketch is empty")); + } + if !(0.0..=1.0).contains(&rank) { + return Err(Error::invalid_argument(format!( + "rank {rank} must be in [0, 1]" + ))); + } + self.sorted_view().quantile(rank, criteria) + } + + /// Returns approximate quantiles for the given normalized ranks. + /// + /// The sorted view is built once and shared across all ranks. + pub fn quantiles(&self, ranks: &[f64], criteria: SearchCriteria) -> Result<Vec<T>, Error> { + if self.is_empty() { + return Err(Error::invalid_argument("sketch is empty")); + } + // Reject invalid ranks before paying for the view build. + for &r in ranks { + if !(0.0..=1.0).contains(&r) { + return Err(Error::invalid_argument(format!( + "rank {r} must be in [0, 1]" + ))); + } + } + let view = self.sorted_view(); + ranks.iter().map(|&r| view.quantile(r, criteria)).collect() + } + + /// Returns the Probability Mass Function over the given split points. + pub fn pmf(&self, split_points: &[T], criteria: SearchCriteria) -> Result<Vec<f64>, Error> { + if self.is_empty() { + return Err(Error::invalid_argument("sketch is empty")); + } + self.sorted_view().pmf(split_points, criteria) + } + + /// Returns the Cumulative Distribution Function over the given split points. + pub fn cdf(&self, split_points: &[T], criteria: SearchCriteria) -> Result<Vec<f64>, Error> { + if self.is_empty() { + return Err(Error::invalid_argument("sketch is empty")); + } + self.sorted_view().cdf(split_points, criteria) + } + + /// Returns an owned, sorted snapshot of the sketch's current state. + /// + /// An empty sketch yields an empty view; queries on it return an error. The + /// view is independent of the sketch — it can be queried (and sent to other + /// threads) while the sketch keeps receiving updates, and it keeps answering + /// from the state it was taken at. + /// + /// Building the view costs `O(retained · log retained)`; each query on it is + /// then `O(log retained)`. Prefer taking one view for repeated queries over + /// calling [`Self::quantile`]/[`Self::pmf`]/[`Self::cdf`], which each build a + /// transient view. + pub fn sorted_view(&self) -> SortedView<T> { + let mut weighted_items = Vec::with_capacity(self.num_retained as usize); + for compactor in &self.compactors { + let weight = compactor.weight(); + for item in compactor.iter() { + weighted_items.push((item.clone(), weight)); + } + } + SortedView::new(weighted_items) + } + + /// Merges another sketch into this one. + /// + /// # Errors + /// + /// Returns an error if the two sketches have different `rank_accuracy`. + pub fn merge(&mut self, other: &Self) -> Result<(), Error> { + if self.rank_accuracy != other.rank_accuracy { + return Err(Error::invalid_argument( + "sketches must have the same rank_accuracy", + )); + } + + if other.is_empty() { + return Ok(()); + } + + self.n += other.n; + + if let Some(m) = &other.min_item { + match &self.min_item { + None => self.min_item = Some(m.clone()), + Some(cur) if m.total_cmp(cur).is_lt() => self.min_item = Some(m.clone()), + _ => {} + } + } + if let Some(m) = &other.max_item { + match &self.max_item { + None => self.max_item = Some(m.clone()), + Some(cur) if m.total_cmp(cur).is_gt() => self.max_item = Some(m.clone()), + _ => {} + } + } + + while self.compactors.len() < other.compactors.len() { + self.grow(); + } + + for (i, other_c) in other.compactors.iter().enumerate() { + self.compactors[i].merge(other_c); + } + + self.update_max_nom_size(); + self.update_num_retained(); + + if self.num_retained >= self.max_nom_size { + self.compress(); + } + + Ok(()) + } + + /// Returns the lower bound for the rank of a quantile at `num_std_dev` confidence. + pub fn rank_lower_bound(&self, rank: f64, num_std_dev: u8) -> f64 { + self.compute_rank_lower_bound( + self.k, + self.compactors.len() as u8, + rank, + num_std_dev, + self.n, + matches!(self.rank_accuracy, RankAccuracy::HighRank), + ) + } + + /// Returns the upper bound for the rank of a quantile at `num_std_dev` confidence. + pub fn rank_upper_bound(&self, rank: f64, num_std_dev: u8) -> f64 { + self.compute_rank_upper_bound( + self.k, + self.compactors.len() as u8, + rank, + num_std_dev, + self.n, + matches!(self.rank_accuracy, RankAccuracy::HighRank), + ) + } + + const FIXED_RSE_FACTOR: f64 = 0.084; + const INIT_NUM_SECTIONS: u8 = 3; + + fn relative_rse_factor() -> f64 { + (0.0512 / Self::INIT_NUM_SECTIONS as f64).sqrt() + } + + fn compute_rank_lower_bound( + &self, + k: u16, + num_levels: u8, + rank: f64, + num_std_dev: u8, + n: u64, + hra: bool, + ) -> f64 { + if self.is_exact_rank_threshold(k, num_levels, rank, n, hra) { + return rank; + } + let relative = Self::relative_rse_factor() / k as f64 * if hra { 1.0 - rank } else { rank }; + let fixed = Self::FIXED_RSE_FACTOR / k as f64; + let lb_rel = rank - num_std_dev as f64 * relative; + let lb_fix = rank - num_std_dev as f64 * fixed; + lb_rel.max(lb_fix).max(0.0) + } + + fn compute_rank_upper_bound( + &self, + k: u16, + num_levels: u8, + rank: f64, + num_std_dev: u8, + n: u64, + hra: bool, + ) -> f64 { + if self.is_exact_rank_threshold(k, num_levels, rank, n, hra) { + return rank; + } + let relative = Self::relative_rse_factor() / k as f64 * if hra { 1.0 - rank } else { rank }; + let fixed = Self::FIXED_RSE_FACTOR / k as f64; + let ub_rel = rank + num_std_dev as f64 * relative; + let ub_fix = rank + num_std_dev as f64 * fixed; + ub_rel.min(ub_fix).min(1.0) + } + + fn is_exact_rank_threshold( + &self, + k: u16, + num_levels: u8, + rank: f64, + n: u64, + hra: bool, + ) -> bool { + let base_cap = k as u64 * Self::INIT_NUM_SECTIONS as u64; + if num_levels == 1 || n <= base_cap { + return true; + } + let exact_rank_thresh = base_cap as f64 / n as f64; + if hra { + rank >= 1.0 - exact_rank_thresh + } else { + rank <= exact_rank_thresh + } + } + + /// Returns per-level info: `(level_index, num_items, capacity, weight)`. + /// Internal/test API; subject to change. + #[doc(hidden)] + pub fn level_info(&self) -> Vec<(usize, u32, u32, u64)> { + self.compactors + .iter() + .enumerate() + .map(|(i, c)| (i, c.num_items(), c.nominal_capacity(), c.weight())) + .collect() + } + + /// Total nominal capacity across all levels. Internal/test API. + #[doc(hidden)] + pub fn total_nominal_capacity(&self) -> u32 { + self.compactors.iter().map(|c| c.nominal_capacity()).sum() + } + + /// Total retained items across all levels. Internal/test API. + #[doc(hidden)] + pub fn total_retained_items(&self) -> u32 { + self.compactors.iter().map(|c| c.num_items()).sum() + } + + /// Sum of `level_items × level_weight` across compactors. Internal/test API. + #[doc(hidden)] + pub fn computed_total_weight(&self) -> u64 { + self.compactors + .iter() + .map(|c| c.num_items() as u64 * c.weight()) + .sum() + } + + pub(super) fn flags_byte(&self) -> u8 { + use super::serialization::FLAG_IS_EMPTY; + use super::serialization::FLAG_IS_HIGH_RANK; + use super::serialization::FLAG_IS_LEVEL_ZERO_SORTED; + use super::serialization::FLAG_RAW_ITEMS; + let mut flags = 0u8; + if self.is_empty() { + flags |= FLAG_IS_EMPTY; + } + if matches!(self.rank_accuracy, RankAccuracy::HighRank) { + flags |= FLAG_IS_HIGH_RANK; + } + if self.is_raw_items() { + flags |= FLAG_RAW_ITEMS; + } + if self.compactors[0].is_sorted() { + flags |= FLAG_IS_LEVEL_ZERO_SORTED; + } + flags + } + + pub(super) fn is_raw_items(&self) -> bool { + use super::serialization::RAW_ITEMS_THRESHOLD; + self.n <= RAW_ITEMS_THRESHOLD && self.compactors.len() == 1 + } + + /// Number of bytes required to serialize the sketch. + pub fn serialized_size_bytes(&self) -> usize { + // Fixed sketch preamble: 8 bytes (preamble_ints, serial_version, family, + // flags, k(2), num_levels, num_raw_items). + let mut size = 8usize; + if self.is_empty() { + return size; + } + if self.is_estimation_mode() { + size += 8; // n + size += T::serialize_size(self.min_item.as_ref().unwrap()); + size += T::serialize_size(self.max_item.as_ref().unwrap()); + } + if self.is_raw_items() { + for item in self.compactors[0].iter() { + size += T::serialize_size(item); + } + } else { + for c in &self.compactors { + // 20-byte compactor preamble + items + size += 20; + for item in c.iter() { + size += T::serialize_size(item); + } + } + } + size + } + + /// Serialize the sketch into a `Vec<u8>` matching the C++/Java REQ wire format. + pub fn serialize(&self) -> Vec<u8> { + use super::serialization::PREAMBLE_INTS_ESTIMATION; + use super::serialization::PREAMBLE_INTS_EXACT; + use super::serialization::SERIAL_VERSION; + use crate::codec::SketchBytes; + use crate::codec::family::Family; + + let mut out = SketchBytes::with_capacity(self.serialized_size_bytes()); + let preamble_ints = if self.is_estimation_mode() { + PREAMBLE_INTS_ESTIMATION + } else { + PREAMBLE_INTS_EXACT + }; + out.write_u8(preamble_ints); + out.write_u8(SERIAL_VERSION); + out.write_u8(Family::REQ.id); + out.write_u8(self.flags_byte()); + out.write_u16_le(self.k); + let num_levels = if self.is_empty() { + 0 + } else { + self.compactors.len() as u8 + }; + out.write_u8(num_levels); + let num_raw_items = if self.is_raw_items() { self.n as u8 } else { 0 }; + out.write_u8(num_raw_items); + + if self.is_empty() { + return out.into_bytes(); + } + + if self.is_estimation_mode() { + out.write_u64_le(self.n); + self.min_item.as_ref().unwrap().serialize_value(&mut out); + self.max_item.as_ref().unwrap().serialize_value(&mut out); + } + + if self.is_raw_items() { + for item in self.compactors[0].iter() { + item.serialize_value(&mut out); + } + } else { + for c in &self.compactors { + c.serialize_into(&mut out); + } + } + + out.into_bytes() + } + + /// Deserialize a sketch from bytes produced by [`Self::serialize`] or by the + /// C++/Java reference implementations. + pub fn deserialize(bytes: &[u8]) -> Result<Self, Error> { + use super::compactor::Compactor; + use super::serialization::FLAG_IS_EMPTY; + use super::serialization::FLAG_IS_HIGH_RANK; + use super::serialization::FLAG_IS_LEVEL_ZERO_SORTED; + use super::serialization::FLAG_RAW_ITEMS; + use super::serialization::RAW_ITEMS_THRESHOLD; + use super::serialization::check_preamble_ints; + use super::serialization::check_serial_version; + use crate::codec::SketchSlice; + use crate::codec::assert::insufficient_data; + use crate::codec::family::Family; + + let mut cursor = SketchSlice::new(bytes); + let preamble_ints = cursor + .read_u8() + .map_err(insufficient_data("preamble_ints"))?; + let serial_version = cursor + .read_u8() + .map_err(insufficient_data("serial_version"))?; + let family_id = cursor.read_u8().map_err(insufficient_data("family_id"))?; + let flags = cursor.read_u8().map_err(insufficient_data("flags"))?; + let k = cursor.read_u16_le().map_err(insufficient_data("k"))?; + let num_levels = cursor.read_u8().map_err(insufficient_data("num_levels"))?; + let num_raw_items = cursor + .read_u8() + .map_err(insufficient_data("num_raw_items"))?; + + check_preamble_ints(preamble_ints, num_levels)?; + check_serial_version(serial_version)?; + Family::REQ.validate_id(family_id)?; + + let is_empty = flags & FLAG_IS_EMPTY != 0; + let hra = flags & FLAG_IS_HIGH_RANK != 0; + let raw_items = flags & FLAG_RAW_ITEMS != 0; + let is_level_zero_sorted = flags & FLAG_IS_LEVEL_ZERO_SORTED != 0; + + let rank_accuracy = if hra { + RankAccuracy::HighRank + } else { + RankAccuracy::LowRank + }; + if !(MIN_K..=MAX_K).contains(&k) || k % 2 != 0 { + return Err(Error::invalid_argument(format!( + "k {k} is not a valid REQ k value" + ))); + } + + if is_empty { + if num_levels != 0 { + return Err(Error::invalid_argument(format!( + "empty REQ sketch must have 0 levels, got {num_levels}" + ))); + } + if num_raw_items != 0 { + return Err(Error::invalid_argument(format!( + "empty REQ sketch must have 0 raw items, got {num_raw_items}" + ))); + } + return ReqSketch::try_new(k, rank_accuracy); + } + + if num_levels == 0 { + return Err(Error::invalid_argument( + "non-empty REQ sketch must have at least one level", + )); + } + + if raw_items { + if num_levels != 1 { + return Err(Error::invalid_argument(format!( + "raw-items REQ sketch must have exactly 1 level, got {num_levels}" + ))); + } + if num_raw_items == 0 || num_raw_items as u64 > RAW_ITEMS_THRESHOLD { + return Err(Error::invalid_argument(format!( + "raw-items REQ sketch must contain 1..={RAW_ITEMS_THRESHOLD} items, got {num_raw_items}" + ))); + } + } else if num_raw_items != 0 { + return Err(Error::invalid_argument(format!( + "non-raw REQ sketch must have 0 raw items, got {num_raw_items}" + ))); + } + + let mut min_item: Option<T> = None; + let mut max_item: Option<T> = None; + let mut n: u64 = 1; + + if num_levels > 1 { + n = cursor.read_u64_le().map_err(insufficient_data("n"))?; + min_item = Some(T::deserialize_value(&mut cursor)?); + max_item = Some(T::deserialize_value(&mut cursor)?); + } + + let mut compactors: Vec<Compactor<T>> = Vec::with_capacity(num_levels as usize); + + if raw_items { + // Single compactor at level 0; items follow directly. + let mut items = Vec::with_capacity(num_raw_items as usize); + for _ in 0..num_raw_items { + items.push(T::deserialize_value(&mut cursor)?); + } + let c = + Compactor::<T>::raw_items_compactor(k, rank_accuracy, items, is_level_zero_sorted); + compactors.push(c); + } else { + for i in 0..num_levels { + let level_sorted = if i == 0 { is_level_zero_sorted } else { true }; + let c = Compactor::<T>::deserialize(&mut cursor, rank_accuracy, level_sorted)?; + compactors.push(c); + } + } + + if num_levels == 1 { + // Recover n / min / max from level 0 (these aren't in the preamble for exact mode). + let level0 = &compactors[0]; + n = level0.num_items() as u64; + let mut iter = level0.iter(); + if let Some(first) = iter.next() { + let mut mn = first.clone(); + let mut mx = first.clone(); + for x in iter { + if x.total_cmp(&mn).is_lt() { + mn = x.clone(); + } + if x.total_cmp(&mx).is_gt() { + mx = x.clone(); + } + } + min_item = Some(mn); + max_item = Some(mx); + } + } + + if n == 0 || min_item.is_none() || max_item.is_none() { + return Err(Error::invalid_argument( + "non-empty REQ sketch contains no items", + )); + } + + let mut sketch = ReqSketch::try_new(k, rank_accuracy)?; + sketch.n = n; + sketch.min_item = min_item; + sketch.max_item = max_item; + sketch.compactors = compactors; + sketch.update_max_nom_size(); + sketch.update_num_retained(); + Ok(sketch) + } + + // --- Internal --- + + pub(super) fn grow(&mut self) { + let level = self.compactors.len() as u8; + let compactor = Compactor::new(level, self.k, self.rank_accuracy); + self.compactors.push(compactor); + self.update_max_nom_size(); + } + + pub(super) fn compress(&mut self) { + for h in 0..self.compactors.len() { + if self.compactors[h].num_items() >= self.compactors[h].nominal_capacity() { + if h == 0 { + self.compactors[0].sort(); + } + if h + 1 >= self.compactors.len() { + self.grow(); + } + self.promotion_buf.clear(); + self.compactors[h].compact_into(self.rank_accuracy, &mut self.promotion_buf); + if !self.promotion_buf.is_empty() { + self.compactors[h + 1].sort(); + self.compactors[h + 1].merge_sorted(&self.promotion_buf); + } + self.update_max_nom_size(); + self.update_num_retained(); + } + } + } + + pub(super) fn update_max_nom_size(&mut self) { + self.max_nom_size = self.compactors.iter().map(|c| c.nominal_capacity()).sum(); + } + + pub(super) fn update_num_retained(&mut self) { + self.num_retained = self.compactors.iter().map(|c| c.num_items()).sum(); + } +} + +impl<T: ReqValue> Default for ReqSketch<T> { + fn default() -> Self { + Self::new() + } +} + +/// Builder for [`ReqSketch`]. +#[derive(Debug, Clone)] +pub struct ReqSketchBuilder<T: ReqValue> { + k: u16, + rank_accuracy: RankAccuracy, + _marker: std::marker::PhantomData<T>, +} + +impl<T: ReqValue> Default for ReqSketchBuilder<T> { + fn default() -> Self { + Self::new() + } +} + +impl<T: ReqValue> ReqSketchBuilder<T> { + /// Creates a new builder with default parameters. + pub fn new() -> Self { + Self { + k: DEFAULT_K, + rank_accuracy: RankAccuracy::HighRank, + _marker: std::marker::PhantomData, + } + } + + /// Sets the `k` parameter. + /// + /// # Errors + /// + /// Returns an error if `k` is odd or outside `[MIN_K, MAX_K]`. + pub fn k(mut self, k: u16) -> Result<Self, Error> { + if !(MIN_K..=MAX_K).contains(&k) { + return Err(Error::invalid_argument(format!( + "k must be in [{}, {}], got {k}", + MIN_K, MAX_K + ))); + } + if k % 2 != 0 { + return Err(Error::invalid_argument(format!("k must be even, got {k}"))); + } + self.k = k; + Ok(self) + } + + /// Sets the rank accuracy. + pub fn rank_accuracy(mut self, ra: RankAccuracy) -> Self { + self.rank_accuracy = ra; + self + } + + /// Builds the sketch. + pub fn build(self) -> Result<ReqSketch<T>, Error> { + ReqSketch::try_new(self.k, self.rank_accuracy) + } +} + +impl<T: ReqValue + fmt::Display> fmt::Display for ReqSketch<T> { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + writeln!(f, "REQ Sketch Summary:")?; + writeln!(f, " k : {}", self.k)?; + writeln!(f, " rank accuracy : {:?}", self.rank_accuracy)?; + writeln!(f, " n : {}", self.n)?; + writeln!(f, " num retained : {}", self.num_retained)?; + writeln!(f, " num levels : {}", self.compactors.len())?; + writeln!(f, " estimation mode : {}", self.is_estimation_mode())?; + if let (Some(min), Some(max)) = (&self.min_item, &self.max_item) { + writeln!(f, " min item : {min}")?; + writeln!(f, " max item : {max}")?; + } + Ok(()) + } +} Review Comment: Good point. I wonder if we'd have a tracking issue to `impl fmt::Display` for all sketches when applicible. This is what datasketches-java does IMO, but the format can be considered once more. cc @notfilippo @ariesdevil @ZENOTME -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
