This is an automated email from the ASF dual-hosted git repository.
tisonkun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datasketches-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 4abc2da fix: preserve quantile query boundaries and simplify REQ
(#284)
4abc2da is described below
commit 4abc2dac5c2e4519de177b0f545f9b5d9298d274
Author: tison <[email protected]>
AuthorDate: Sat Oct 3 15:02:58 2026 +0800
fix: preserve quantile query boundaries and simplify REQ (#284)
---
CHANGELOG.md | 3 +
README.md | 1 +
benchmarks/kll/query.rs | 6 +-
benchmarks/req/query.rs | 6 +-
datasketches/src/kll/sketch.rs | 17 +++-
datasketches/src/kll/sorted_view.rs | 22 ++++-
datasketches/src/req/compactor.rs | 111 +++++--------------------
datasketches/src/req/sketch.rs | 30 ++++---
datasketches/src/req/sorted_view.rs | 131 +++++++++---------------------
tests-integration/tests/kll_test/query.rs | 47 +++++++++++
tests-integration/tests/req_test/query.rs | 79 ++++++++++++++++++
11 files changed, 252 insertions(+), 201 deletions(-)
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5137769..c9739f2 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -17,6 +17,7 @@ All significant changes to this project will be documented in
this file.
### Improvements
* Improve the readability of `Debug` output for HLL and CPC sketches and
unions.
+* `KllSketch::quantile` and `ReqSketch::quantile` are faster at ranks `0.0`
and `1.0`, with query work independent of the number of retained samples.
* `BloomFilter::insert` is faster for integer and raw-byte inputs.
`BloomFilter::contains_and_insert` is also faster when checking already-present
integer values.
* `CountMinSketch` updates are faster for integer and raw-byte inputs.
* `CpcSketch` updates are faster for integer and raw-byte inputs.
@@ -29,6 +30,7 @@ All significant changes to this project will be documented in
this file.
### Bug fixes
+* `KllSketch` and `ReqSketch` quantile queries, including their sorted views,
now return the exact stream minimum and maximum at ranks `0.0` and `1.0`, even
after compaction or when the stream weight exceeds `2^53`.
* Empty compact Theta and Tuple sketches now serialize with a zero seed hash,
matching the canonical cross-language encoding. Deserialization associates
empty sketches with the supplied seed, including legacy Theta v2 images with
zero or mismatched seed hashes.
* `CountMinSketch` updates now panic and merges return `InvalidArgument` if
the total absolute weight would exceed the counter type's maximum. Both leave
the sketch unchanged, including in release builds.
* `CountMinSketch::upper_bound` now clamps to the counter type's maximum
instead of overflowing.
@@ -36,6 +38,7 @@ All significant changes to this project will be documented in
this file.
* `FrequentItemsSketch` updates and merges now panic without changing the
sketch if the total stream weight would overflow, including in release builds.
* `FrequentItemsSketch` deserialization now returns `InvalidData` if a
non-empty image declares zero stream weight or the item weights sum to more
than the declared stream weight.
* `ReqSketch` updates now panic and merges return `InvalidArgument` if the
stream weight would exceed `u64::MAX`. Both leave the sketch unchanged,
including in release builds.
+* `ReqSketch::cdf` and its sorted view now preserve the rank at each split
point and end at exactly `1.0`, avoiding extra rounding from accumulating PMF
values.
* `TDigestMut` updates and merges now panic without changing the digest if the
total weight would exceed `u64::MAX`, including in release builds.
* `TDigestMut::merge` now preserves the true minimum and maximum from both
inputs, including compressed digests.
* `TDigest` and `TDigestMut` deserialization now returns `InvalidData` for
invalid flags or extrema, out-of-range values, unsorted centroids, or non-empty
images with no stored values.
diff --git a/README.md b/README.md
index b7b04ba..a4a0373 100644
--- a/README.md
+++ b/README.md
@@ -70,6 +70,7 @@ Enable multiple algorithms by listing their features
together, such as `features
| `cpc` | `CpcSketch`, `CpcUnion`, `CpcWrapper` | Highly compact
distinct-count estimation and unions.
|
| `frequencies` | `FrequentItemsSketch` | Heavy-hitter
discovery with upper and lower frequency bounds.
|
| `hll` | `HllSketch`, `HllUnion` | Fast distinct-count
estimation and unions. |
+| `kll` | `KllSketch` | General-purpose
quantile, rank, PMF, and CDF queries over totally ordered items.
|
| `req` | `ReqSketch` | Relative-error
quantile, rank, PMF, and CDF queries with configurable high- or low-rank
accuracy. |
| `tdigest` | `TDigestMut`, `TDigest` | Quantile and rank
estimation, with high accuracy near distribution tails.
|
| `theta` | `ThetaSketch` and set operations | Distinct counts, set
expressions, and Jaccard similarity. |
diff --git a/benchmarks/kll/query.rs b/benchmarks/kll/query.rs
index 1c61aa0..508a54b 100644
--- a/benchmarks/kll/query.rs
+++ b/benchmarks/kll/query.rs
@@ -29,10 +29,10 @@ fn rank(bencher: Bencher) {
bencher.bench_local(|| black_box(&sketch).rank(black_box(&item),
SearchCriteria::Inclusive));
}
-#[divan::bench]
-fn quantile(bencher: Bencher) {
+#[divan::bench(args = [0.0, 0.5, 1.0])]
+fn quantile(bencher: Bencher, rank: f64) {
let sketch = prepared_sketch();
- bencher.bench_local(|| black_box(&sketch).quantile(black_box(0.5),
SearchCriteria::Inclusive));
+ bencher.bench_local(|| black_box(&sketch).quantile(black_box(rank),
SearchCriteria::Inclusive));
}
#[divan::bench]
diff --git a/benchmarks/req/query.rs b/benchmarks/req/query.rs
index b52dfc5..57f4eae 100644
--- a/benchmarks/req/query.rs
+++ b/benchmarks/req/query.rs
@@ -30,11 +30,11 @@ fn rank(bencher: Bencher) {
bencher.bench_local(|| black_box(&sketch).rank(black_box(&item),
SearchCriteria::Inclusive));
}
-#[divan::bench]
-fn quantile(bencher: Bencher) {
+#[divan::bench(args = [0.0, 0.5, 1.0])]
+fn quantile(bencher: Bencher, rank: f64) {
let sketch = prepared_sketch();
- bencher.bench_local(|| black_box(&sketch).quantile(black_box(0.5),
SearchCriteria::Inclusive));
+ bencher.bench_local(|| black_box(&sketch).quantile(black_box(rank),
SearchCriteria::Inclusive));
}
// Intended pattern for repeated queries: build the view once, query it many
diff --git a/datasketches/src/kll/sketch.rs b/datasketches/src/kll/sketch.rs
index 76e76fb..9fb3a62 100644
--- a/datasketches/src/kll/sketch.rs
+++ b/datasketches/src/kll/sketch.rs
@@ -227,6 +227,8 @@ impl<T: Clone + Ord> KllSketch<T> {
/// Returns the quantile for the given normalized rank.
///
+ /// Ranks `0.0` and `1.0` return the exact minimum and maximum items of
the stream.
+ ///
/// # Errors
///
/// Returns an error if the sketch is empty or `rank` is outside `[0.0,
1.0]`.
@@ -239,11 +241,19 @@ impl<T: Clone + Ord> KllSketch<T> {
"rank must be in [0.0, 1.0], got {rank}"
)));
}
+ if rank == 0.0 {
+ return Ok(self.min_item.as_ref().unwrap().clone());
+ }
+ if rank == 1.0 {
+ return Ok(self.max_item.as_ref().unwrap().clone());
+ }
self.sorted_view().quantile(rank, criteria)
}
/// Returns approximate quantiles for the given normalized ranks.
///
+ /// Ranks `0.0` and `1.0` return the exact minimum and maximum items of
the stream.
+ ///
/// The sorted view is built once for the whole batch.
///
/// # Errors
@@ -283,7 +293,12 @@ impl<T: Clone + Ord> KllSketch<T> {
///
/// The view can be reused for repeated queries while this sketch
continues to receive updates.
pub fn sorted_view(&self) -> SortedView<T> {
- build_sorted_view(&self.levels, self.is_level_zero_sorted)
+ build_sorted_view(
+ &self.levels,
+ self.is_level_zero_sorted,
+ self.min_item.clone(),
+ self.max_item.clone(),
+ )
}
/// Returns the normalized single-sided rank error for the configured k.
diff --git a/datasketches/src/kll/sorted_view.rs
b/datasketches/src/kll/sorted_view.rs
index b58a69a..d11f45b 100644
--- a/datasketches/src/kll/sorted_view.rs
+++ b/datasketches/src/kll/sorted_view.rs
@@ -28,6 +28,8 @@ use crate::error::Error;
pub struct SortedView<T: Clone> {
entries: Vec<Entry<T>>,
total_weight: u64,
+ min_item: Option<T>,
+ max_item: Option<T>,
}
#[derive(Debug, Clone)]
@@ -37,7 +39,7 @@ struct Entry<T> {
}
impl<T: Clone + Ord> SortedView<T> {
- fn from_sorted(mut entries: Vec<Entry<T>>) -> Self {
+ fn from_sorted(mut entries: Vec<Entry<T>>, min_item: Option<T>, max_item:
Option<T>) -> Self {
let mut total_weight = 0u64;
for entry in &mut entries {
total_weight += entry.cumulative_weight;
@@ -46,6 +48,8 @@ impl<T: Clone + Ord> SortedView<T> {
Self {
entries,
total_weight,
+ min_item,
+ max_item,
}
}
@@ -87,6 +91,8 @@ impl<T: Clone + Ord> SortedView<T> {
/// Returns the approximate quantile for `rank`.
///
+ /// Ranks `0.0` and `1.0` return the exact minimum and maximum items of
the stream.
+ ///
/// # Errors
///
/// Returns an error if the view is empty or `rank` is outside `[0.0,
1.0]`.
@@ -100,6 +106,14 @@ impl<T: Clone + Ord> SortedView<T> {
)));
}
+ // Compaction may discard extrema, and large weights can lose
precision in f64.
+ if rank == 0.0 {
+ return Ok(self.min_item.as_ref().unwrap().clone());
+ }
+ if rank == 1.0 {
+ return Ok(self.max_item.as_ref().unwrap().clone());
+ }
+
let weight = if criteria == SearchCriteria::Inclusive {
(rank * self.total_weight as f64).ceil() as u64
} else {
@@ -116,6 +130,8 @@ impl<T: Clone + Ord> SortedView<T> {
/// Returns approximate quantiles for all `ranks`.
///
+ /// Ranks `0.0` and `1.0` return the exact minimum and maximum items of
the stream.
+ ///
/// # Errors
///
/// Returns an error if the view is empty or any rank is outside `[0.0,
1.0]`.
@@ -164,6 +180,8 @@ impl<T: Clone + Ord> SortedView<T> {
pub fn build_sorted_view<T: Clone + Ord>(
levels: &[Vec<T>],
is_level_zero_sorted: bool,
+ min_item: Option<T>,
+ max_item: Option<T>,
) -> SortedView<T> {
let mut runs = Vec::with_capacity(levels.len());
for (level_index, level) in levels.iter().enumerate() {
@@ -197,7 +215,7 @@ pub fn build_sorted_view<T: Clone + Ord>(
runs = merged_runs;
}
- SortedView::from_sorted(runs.pop().unwrap_or_default())
+ SortedView::from_sorted(runs.pop().unwrap_or_default(), min_item, max_item)
}
fn merge_sorted_entries<T: Ord>(left: Vec<Entry<T>>, right: Vec<Entry<T>>) ->
Vec<Entry<T>> {
diff --git a/datasketches/src/req/compactor.rs
b/datasketches/src/req/compactor.rs
index a56c0fa..1a4aa64 100644
--- a/datasketches/src/req/compactor.rs
+++ b/datasketches/src/req/compactor.rs
@@ -17,8 +17,7 @@
//! Compactor implementation for REQ sketch levels.
//!
-//! Each level in the REQ sketch uses a compactor to maintain a bounded set of
items
-//! with deterministic compaction when capacity is exceeded.
+//! Each level compacts a selected range, retaining one item from each pair at
twice the weight.
use crate::common::random::random_bit;
use crate::error::Error;
@@ -31,31 +30,20 @@ use crate::req::value::ReqValue;
/// A compactor maintains items at a specific level of the REQ sketch.
///
-/// When the compactor reaches its nominal capacity, it performs compaction
-/// by keeping approximately half the items and promoting the rest to the next
level.
+/// Compaction preserves the configured accuracy tail and promotes samples
from the other tail.
#[derive(Debug, Clone)]
pub struct Compactor<T> {
- /// Current items in the compactor
items: Vec<T>,
- /// Whether items are currently sorted
is_sorted: bool,
- /// State for deterministic compaction
+ /// Determines which sections are compacted and when their count doubles.
state: u64,
- /// Reusable scratch buffer for compaction operations
scratch_buffer: Vec<T>,
-
- /// Actual section size (rounded to integer)
section_size: u32,
- /// Number of sections in this compactor
num_sections: u8,
- /// The level of this compactor (0 = base level)
lg_weight: u8,
-
- /// Whether this compactor is configured for high rank accuracy
rank_accuracy: RankAccuracy,
- /// Raw section size (maybe fractional)
+ /// Preserves fractional section sizes across successive divisions by
sqrt(2).
section_size_raw: f32,
- /// Random bit for compaction
coin: bool,
}
@@ -64,11 +52,6 @@ where
T: Clone + Ord,
{
/// Creates a new compactor for the given level.
- ///
- /// # Arguments
- /// * `lg_weight` - The level (log weight) of this compactor
- /// * `k` - The k parameter from the parent sketch
- /// * `rank_accuracy` - Rank accuracy configuration
pub fn new(lg_weight: u8, k: u16, rank_accuracy: RankAccuracy) -> Self {
let section_size_raw = k as f32;
let section_size = nearest_even_section_size(section_size_raw);
@@ -156,8 +139,8 @@ where
}
}
- /// Merges pre-sorted items into this compactor.
- /// Merges sorted items into this compactor using scratch buffer to avoid
allocation.
+ /// Merges sorted items using the reusable scratch buffer.
+ ///
/// Both this compactor's items and the input must be sorted.
#[inline(always)]
pub fn merge_sorted(&mut self, items: &[T]) {
@@ -171,18 +154,13 @@ where
return;
}
- // Ensure sorted on both inputs by contract
let total = self.items.len() + items.len();
self.scratch_buffer.clear();
- if self.scratch_buffer.capacity() < total {
- self.scratch_buffer
- .reserve(total - self.scratch_buffer.capacity());
- }
+ self.scratch_buffer.reserve(total);
let (mut i, mut j) = (0usize, 0usize);
let (a, b) = (&self.items, items);
- // Two-pointer merge into scratch buffer
while i < a.len() && j < b.len() {
if a[i] <= b[j] {
self.scratch_buffer.push(a[i].clone());
@@ -193,15 +171,9 @@ where
}
}
- // Add remaining elements
- if i < a.len() {
- self.scratch_buffer.extend_from_slice(&a[i..]);
- }
- if j < b.len() {
- self.scratch_buffer.extend_from_slice(&b[j..]);
- }
+ self.scratch_buffer.extend_from_slice(&a[i..]);
+ self.scratch_buffer.extend_from_slice(&b[j..]);
- // Swap scratch buffer with items (zero-copy)
self.items.clear();
std::mem::swap(&mut self.items, &mut self.scratch_buffer);
self.is_sorted = true;
@@ -211,62 +183,43 @@ where
#[inline(always)]
pub fn sort(&mut self) {
if !self.is_sorted {
- // Use unstable sort for better performance (stable not needed for
REQ sketch)
self.items.sort_unstable();
self.is_sorted = true;
}
}
- /// Compacts into the provided output buffer without allocating.
- /// Writes promoted items into `out` and removes the compacted range
in-place via `copy_within +
- /// truncate`.
+ /// Writes promoted samples into `out` and removes their source range from
this level.
#[inline(always)]
- pub fn compact_into(&mut self, _rank_accuracy: RankAccuracy, out: &mut
Vec<T>) {
- if self.items.is_empty() {
- out.clear();
- return;
- }
-
- // Sort entire buffer (C++ sorts full buffer before compaction)
+ pub fn compact_into(&mut self, out: &mut Vec<T>) {
+ out.clear();
self.sort();
- // Calculate sections to compact based on state
let secs_to_compact =
((!self.state).trailing_zeros() + 1).min(self.num_sections as u32)
as u8;
- let compaction_range = self.compute_compaction_range(secs_to_compact);
-
- // Must have at least 2 items to compact
- if compaction_range.1 <= compaction_range.0 || (compaction_range.1 -
compaction_range.0) < 2
- {
- out.clear();
+ let (start, end) = self.compute_compaction_range(secs_to_compact);
+ if end - start < 2 {
return;
}
+ // Complement consecutive choices so each pair of compactions promotes
both parities.
if (self.state & 1) == 1 {
- self.coin = !self.coin; // flip coin for odd states
+ self.coin = !self.coin;
} else {
- self.coin = random_bit(); // random coin flip for even states
+ self.coin = random_bit();
}
- let odds = self.coin;
- // Build promoted items directly into output buffer (no alloc)
- out.clear();
- let (start, end) = compaction_range;
- let mut i = start + if odds { 1 } else { 0 };
+ let mut i = start + usize::from(self.coin);
while i < end {
- out.push(self.items[i].clone()); // TODO: use Copy fast-path for
numeric types
+ out.push(self.items[i].clone());
i += 2;
}
- // Remove the compacted range in-place by rotating elements left
let removed = end - start;
if end < self.items.len() {
- // Use rotate_left to move tail elements to fill the gap
self.items[start..].rotate_left(removed);
}
self.items.truncate(self.items.len() - removed);
- // Update state, then ensure enough sections (C++ order)
self.state = self.state.wrapping_add(1);
self.ensure_enough_sections();
}
@@ -286,8 +239,6 @@ where
1u64 << self.lg_weight
}
- // Private helper methods
-
fn ensure_enough_sections(&mut self) -> bool {
let Some(threshold) = self
.num_sections
@@ -317,36 +268,18 @@ where
let mut non_compact = nom_capacity / 2
+ (self.num_sections - secs_to_compact) as usize *
self.section_size as usize;
- // if (((num_items_ - non_compact) & 1) == 1) ++non_compact;
+ // Preserve an extra item when needed to make the compacted range even.
if self.items.len() >= non_compact && ((self.items.len() -
non_compact) & 1) == 1 {
non_compact += 1;
}
- let (low, high) = match self.rank_accuracy {
- RankAccuracy::HighRank => {
- // HRA: Protect high ranks by compacting LOW sections (low
values)
- // This means we compact from [0, num_items - non_compact]
(bottom end)
- let high = if self.items.len() >= non_compact {
- self.items.len() - non_compact
- } else {
- 0
- };
- (0, high)
- }
+ match self.rank_accuracy {
+ RankAccuracy::HighRank => (0,
self.items.len().saturating_sub(non_compact)),
RankAccuracy::LowRank => {
- // LRA: Protect low ranks by compacting HIGH sections (high
values)
- // This means we compact from [non_compact, num_items] (top
end)
let low = non_compact.min(self.items.len());
(low, self.items.len())
}
- };
-
- // Empty window safety: ensure we have at least 2 items to compact
- if high <= low || (high - low) < 2 {
- return (0, 0); // Signal no compaction needed
}
-
- (low, high)
}
/// Serialize this compactor (preamble + items) into the byte buffer.
diff --git a/datasketches/src/req/sketch.rs b/datasketches/src/req/sketch.rs
index 2373b1f..77a3284 100644
--- a/datasketches/src/req/sketch.rs
+++ b/datasketches/src/req/sketch.rs
@@ -198,8 +198,10 @@ where
/// 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.
+ /// Ranks `0.0` and `1.0` return the exact minimum and maximum items of
the stream.
+ ///
+ /// For ranks strictly between `0.0` and `1.0`, builds a transient
[`SortedView`].
+ /// For repeated queries at those ranks, take one snapshot with
[`Self::sorted_view`].
///
/// # Errors
///
@@ -213,11 +215,19 @@ where
"rank {rank} must be in [0, 1]"
)));
}
+ if rank == 0.0 {
+ return Ok(self.min_item.as_ref().unwrap().clone());
+ }
+ if rank == 1.0 {
+ return Ok(self.max_item.as_ref().unwrap().clone());
+ }
self.sorted_view().quantile(rank, criteria)
}
/// Returns approximate quantiles for the given normalized ranks.
///
+ /// Ranks `0.0` and `1.0` return the exact minimum and maximum items of
the stream.
+ ///
/// The sorted view is built once and shared across all ranks.
///
/// # Errors
@@ -270,10 +280,10 @@ where
/// 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.
+ /// Building the view costs `O(retained · log retained)`; rank and
quantile queries
+ /// then take `O(log retained)`. Prefer reusing one view for repeated
quantile
+ /// queries at ranks strictly between `0.0` and `1.0`, or repeated
+ /// [`Self::pmf`]/[`Self::cdf`] queries.
pub fn sorted_view(&self) -> SortedView<T> {
let mut weighted_items = Vec::with_capacity(self.num_retained as
usize);
for compactor in &self.compactors {
@@ -282,7 +292,7 @@ where
weighted_items.push((item.clone(), weight));
}
}
- SortedView::new(weighted_items)
+ SortedView::new(weighted_items, self.min_item.clone(),
self.max_item.clone())
}
/// Merges another sketch into this one.
@@ -785,14 +795,10 @@ where
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);
+ self.compactors[h].compact_into(&mut self.promotion_buf);
if !self.promotion_buf.is_empty() {
self.compactors[h + 1].sort();
self.compactors[h + 1].merge_sorted(&self.promotion_buf);
diff --git a/datasketches/src/req/sorted_view.rs
b/datasketches/src/req/sorted_view.rs
index 8affbbb..ec62f87 100644
--- a/datasketches/src/req/sorted_view.rs
+++ b/datasketches/src/req/sorted_view.rs
@@ -30,34 +30,23 @@ use crate::error::Error;
/// repeated quantile/rank queries.
#[derive(Debug, Clone)]
pub struct SortedView<T> {
- /// Items in sorted order
items: Vec<T>,
- /// Cumulative weights for each item
cumulative_weights: Vec<u64>,
- /// Total weight of all items
total_weight: u64,
+ min_item: Option<T>,
+ max_item: Option<T>,
}
impl<T> SortedView<T>
where
T: Clone + Ord,
{
- /// Creates a new sorted view from weighted items.
- ///
- /// # Arguments
- /// * `weighted_items` - Vector of (item, weight) pairs
- ///
- /// The items will be sorted and cumulative weights computed.
- pub(super) fn new(mut weighted_items: Vec<(T, u64)>) -> Self {
- if weighted_items.is_empty() {
- return Self {
- items: vec![],
- cumulative_weights: vec![],
- total_weight: 0,
- };
- }
-
- // Sort by item value - use unstable sort for better performance
+ /// Creates a sorted view, combining the weights of equal items.
+ pub(super) fn new(
+ mut weighted_items: Vec<(T, u64)>,
+ min_item: Option<T>,
+ max_item: Option<T>,
+ ) -> Self {
weighted_items.sort_unstable_by(|a, b| a.0.cmp(&b.0));
let mut items: Vec<T> = Vec::with_capacity(weighted_items.len());
@@ -82,6 +71,8 @@ where
items,
cumulative_weights,
total_weight: cumulative_weight,
+ min_item,
+ max_item,
}
}
@@ -90,7 +81,7 @@ where
self.items.is_empty()
}
- /// Returns the number of distinct items in the sorted view.
+ /// Returns the number of distinct retained items in the view.
pub fn len(&self) -> usize {
self.items.len()
}
@@ -109,31 +100,21 @@ where
if self.is_empty() {
return Err(Error::invalid_argument("sketch is empty"));
}
- match criteria {
- SearchCriteria::Inclusive => {
- // Find the last position where items[i] <= item
- // partition_point finds first index where predicate is false
- let pos = self.items.partition_point(|x| x <= item);
- if pos == 0 {
- Ok(0.0)
- } else {
- Ok(self.cumulative_weights[pos - 1] as f64 /
self.total_weight as f64)
- }
- }
- SearchCriteria::Exclusive => {
- // Find the last position where items[i] < item
- let pos = self.items.partition_point(|x| x < item);
- if pos == 0 {
- Ok(0.0)
- } else {
- Ok(self.cumulative_weights[pos - 1] as f64 /
self.total_weight as f64)
- }
- }
+ let pos = match criteria {
+ SearchCriteria::Inclusive => self.items.partition_point(|x| x <=
item),
+ SearchCriteria::Exclusive => self.items.partition_point(|x| x <
item),
+ };
+ if pos == 0 {
+ Ok(0.0)
+ } else {
+ Ok(self.cumulative_weights[pos - 1] as f64 / self.total_weight as
f64)
}
}
/// Returns the approximate quantile at the given normalized rank.
///
+ /// Ranks `0.0` and `1.0` return the exact minimum and maximum items of
the stream.
+ ///
/// # Errors
///
/// Returns an error if the view is empty or `rank` is outside `[0.0,
1.0]`.
@@ -148,43 +129,29 @@ where
)));
}
- // Handle edge cases
+ // Compaction may discard extrema, and large weights can lose
precision in f64.
if rank == 0.0 {
- match criteria {
- SearchCriteria::Inclusive => return Ok(self.items[0].clone()),
- SearchCriteria::Exclusive => return Ok(self.items[0].clone()),
- }
+ return Ok(self.min_item.as_ref().unwrap().clone());
}
if rank == 1.0 {
- return Ok(self.items[self.items.len() - 1].clone());
+ return Ok(self.max_item.as_ref().unwrap().clone());
}
- // Convert rank to target cumulative weight
- // uint64_t weight = static_cast<uint64_t>(inclusive ? std::ceil(rank
* total_weight_) :
- // rank * total_weight_);
let target_weight = match criteria {
SearchCriteria::Inclusive => (rank * self.total_weight as
f64).ceil() as u64,
SearchCriteria::Exclusive => (rank * self.total_weight as f64) as
u64,
};
let index = match criteria {
- SearchCriteria::Inclusive => {
- // Equivalent to C++ lower_bound: first index where
cumulative_weight >= target
- self.cumulative_weights
- .partition_point(|&w| w < target_weight)
- }
- SearchCriteria::Exclusive => {
- // Equivalent to C++ upper_bound: first index where
cumulative_weight > target
- self.cumulative_weights
- .partition_point(|&w| w <= target_weight)
- }
+ SearchCriteria::Inclusive => self
+ .cumulative_weights
+ .partition_point(|&w| w < target_weight),
+ SearchCriteria::Exclusive => self
+ .cumulative_weights
+ .partition_point(|&w| w <= target_weight),
};
- if index >= self.items.len() {
- return Ok(self.items[self.items.len() - 1].clone());
- }
-
- Ok(self.items[index].clone())
+ Ok(self.items[index.min(self.items.len() - 1)].clone())
}
/// Returns the probability mass function (PMF) over the given split
points.
@@ -195,24 +162,10 @@ where
///
/// Returns an error if the view is empty or the split points are not
strictly increasing.
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.validate_split_points(split_points)?;
-
- let mut result = Vec::with_capacity(split_points.len() + 1);
- let mut prev_rank = 0.0;
-
- for split_point in split_points {
- let rank = self.rank(split_point, criteria)?;
- result.push(rank - prev_rank);
- prev_rank = rank;
+ let mut result = self.cdf(split_points, criteria)?;
+ for index in (1..result.len()).rev() {
+ result[index] -= result[index - 1];
}
-
- // Add the final interval
- result.push(1.0 - prev_rank);
-
Ok(result)
}
@@ -231,21 +184,17 @@ where
self.validate_split_points(split_points)?;
let mut result = Vec::with_capacity(split_points.len() + 1);
- let mut cumulative = 0.0;
-
- let pmf = self.pmf(split_points, criteria)?;
- for mass in pmf {
- cumulative += mass;
- result.push(cumulative);
+ for split_point in split_points {
+ result.push(self.rank(split_point, criteria)?);
}
-
+ result.push(1.0);
Ok(result)
}
fn validate_split_points(&self, split_points: &[T]) -> Result<(), Error> {
if split_points.windows(2).any(|pair| pair[0] >= pair[1]) {
return Err(Error::invalid_argument(
- "Split points must be unique and monotonically
increasing".to_string(),
+ "Split points must be unique and monotonically increasing",
));
}
Ok(())
@@ -266,7 +215,7 @@ mod tests {
fn create_test_view() -> SortedView<i32> {
let weighted_items = vec![(1, 1), (3, 1), (5, 1), (7, 1), (9, 1)];
- SortedView::new(weighted_items)
+ SortedView::new(weighted_items, Some(1), Some(9))
}
#[test]
@@ -349,7 +298,7 @@ mod tests {
#[test]
fn test_empty_view() {
- let view: SortedView<i32> = SortedView::new(vec![]);
+ let view: SortedView<i32> = SortedView::new(vec![], None, None);
assert!(view.is_empty());
assert_eq!(view.len(), 0);
assert_eq!(view.total_weight(), 0);
diff --git a/tests-integration/tests/kll_test/query.rs
b/tests-integration/tests/kll_test/query.rs
index 5474df9..3730224 100644
--- a/tests-integration/tests/kll_test/query.rs
+++ b/tests-integration/tests/kll_test/query.rs
@@ -22,6 +22,53 @@ use datasketches::kll::KllSketch;
const DEFAULT_K: u16 = 200;
const NUMERIC_NOISE_TOLERANCE: f64 = 1e-6;
+#[test]
+fn quantile_endpoints_preserve_extrema_after_compaction() {
+ let mut sketch = KllSketch::<i64>::new(8).unwrap();
+ for item in 0..8 {
+ sketch.update(item);
+ }
+ // Compact an even run with distinct extrema before inserting an interior
item.
+ // Either sampling parity discards one of the two endpoints.
+ sketch.update(3);
+ assert!(sketch.is_estimation_mode());
+ let view = sketch.sorted_view();
+
+ for criteria in [SearchCriteria::Inclusive, SearchCriteria::Exclusive] {
+ assert_eq!(sketch.quantile(0.0, criteria).unwrap(), 0);
+ assert_eq!(sketch.quantile(1.0, criteria).unwrap(), 7);
+ assert_eq!(sketch.quantiles(&[0.0, 1.0], criteria).unwrap(), [0, 7]);
+ }
+
+ sketch.reset();
+ for criteria in [SearchCriteria::Inclusive, SearchCriteria::Exclusive] {
+ assert!(sketch.quantile(0.0, criteria).is_err());
+ assert!(sketch.quantile(1.0, criteria).is_err());
+ assert_eq!(view.quantile(0.0, criteria).unwrap(), 0);
+ assert_eq!(view.quantile(1.0, criteria).unwrap(), 7);
+ assert_eq!(view.quantiles(&[0.0, 1.0], criteria).unwrap(), [0, 7]);
+ }
+}
+
+#[test]
+fn quantile_endpoints_with_large_stream_weight() {
+ let mut sketch = KllSketch::<i64>::new(8).unwrap();
+ sketch.update(0);
+ for _ in 0..53 {
+ sketch.merge(&sketch.clone()).unwrap();
+ }
+ sketch.update(1);
+ assert_eq!(sketch.n(), (1 << 53) + 1);
+ let view = sketch.sorted_view();
+
+ for criteria in [SearchCriteria::Inclusive, SearchCriteria::Exclusive] {
+ assert_eq!(sketch.quantile(0.0, criteria).unwrap(), 0);
+ assert_eq!(sketch.quantile(1.0, criteria).unwrap(), 1);
+ assert_eq!(view.quantile(0.0, criteria).unwrap(), 0);
+ assert_eq!(view.quantile(1.0, criteria).unwrap(), 1);
+ }
+}
+
#[test]
fn empty_and_invalid_queries_return_errors() {
let mut sketch = KllSketch::<i64>::new(DEFAULT_K).unwrap();
diff --git a/tests-integration/tests/req_test/query.rs
b/tests-integration/tests/req_test/query.rs
index 4fb2e62..bc03743 100644
--- a/tests-integration/tests/req_test/query.rs
+++ b/tests-integration/tests/req_test/query.rs
@@ -19,6 +19,7 @@
use datasketches::common::SearchCriteria;
use datasketches::error::Error;
+use datasketches::req::RankAccuracy;
use datasketches::req::ReqSketch;
use googletest::assert_that;
use googletest::prelude::all;
@@ -120,6 +121,84 @@ fn pmf_and_cdf_are_consistent() {
assert_that!(cdf[cdf.len() - 1], near(1.0, 1e-10));
}
+#[test]
+fn cdf_preserves_ranks_and_exact_endpoint() {
+ let mut sketch = ReqSketch::<i64>::default();
+ for item in 1..=70 {
+ sketch.update(item);
+ }
+ assert!(!sketch.is_estimation_mode());
+ let view = sketch.sorted_view();
+
+ for criteria in [SearchCriteria::Inclusive, SearchCriteria::Exclusive] {
+ let offset = i64::from(criteria == SearchCriteria::Exclusive);
+ let splits = [8, 17, 58].map(|item| item + offset);
+ let expected: Vec<_> = splits
+ .iter()
+ .map(|item| sketch.rank(item, criteria).unwrap())
+ .chain(std::iter::once(1.0))
+ .collect();
+
+ assert_eq!(sketch.cdf(&splits, criteria).unwrap(), expected);
+ assert_eq!(view.cdf(&splits, criteria).unwrap(), expected);
+ assert_eq!(view.cdf(&[], criteria).unwrap(), [1.0]);
+ }
+}
+
+#[test]
+fn quantile_endpoints_preserve_extrema_after_compaction() {
+ for accuracy in [RankAccuracy::HighRank, RankAccuracy::LowRank] {
+ let mut sketch = ReqSketch::<i64>::new(4, accuracy).unwrap();
+ // Introduce new extrema across consecutive compactions, whose sampling
+ // parities alternate, so both choices must preserve the exact
endpoints.
+ for item in 1..=64 {
+ sketch.update(-item);
+ sketch.update(item);
+ let view = sketch.sorted_view();
+
+ for criteria in [SearchCriteria::Inclusive,
SearchCriteria::Exclusive] {
+ assert_eq!(sketch.quantile(0.0, criteria).unwrap(), -item);
+ assert_eq!(sketch.quantile(1.0, criteria).unwrap(), item);
+ assert_eq!(
+ sketch.quantiles(&[0.0, 1.0], criteria).unwrap(),
+ [-item, item]
+ );
+ assert_eq!(view.quantile(0.0, criteria).unwrap(), -item);
+ assert_eq!(view.quantile(1.0, criteria).unwrap(), item);
+ }
+ }
+ assert!(sketch.is_estimation_mode());
+
+ let view = sketch.sorted_view();
+ sketch.reset();
+ for criteria in [SearchCriteria::Inclusive, SearchCriteria::Exclusive]
{
+ assert!(sketch.quantile(0.0, criteria).is_err());
+ assert!(sketch.quantile(1.0, criteria).is_err());
+ assert_eq!(view.quantile(0.0, criteria).unwrap(), -64);
+ assert_eq!(view.quantile(1.0, criteria).unwrap(), 64);
+ }
+ }
+}
+
+#[test]
+fn quantile_endpoints_with_large_stream_weight() {
+ let mut sketch = ReqSketch::<i64>::default();
+ sketch.update(0);
+ for _ in 0..53 {
+ sketch.merge(&sketch.clone()).unwrap();
+ }
+ sketch.update(1);
+ assert_eq!(sketch.n(), (1 << 53) + 1);
+ let view = sketch.sorted_view();
+
+ for criteria in [SearchCriteria::Inclusive, SearchCriteria::Exclusive] {
+ assert_eq!(sketch.quantile(0.0, criteria).unwrap(), 0);
+ assert_eq!(sketch.quantile(1.0, criteria).unwrap(), 1);
+ assert_eq!(view.quantile(0.0, criteria).unwrap(), 0);
+ assert_eq!(view.quantile(1.0, criteria).unwrap(), 1);
+ }
+}
+
#[test]
fn rank_is_monotonic_and_bounded() {
let mut sketch: ReqSketch<ReqF64> = ReqSketch::default();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
