diff --git a/CHANGELOG.md b/CHANGELOG.md index 92c76dda..404f8276 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,10 @@ All significant changes to this project will be documented in this file. ## Unreleased +### New features + +* Add `DdSketch` behind the `ddsketch` feature for relative-value quantiles with configurable accuracy, bounded buckets, lowest- or highest-value collapse, weighted updates, merging, and Rust-specific serialization. Queries can identify collapsed buckets that no longer retain the configured accuracy. + ### Improvements * `KllSketch::rank` is faster for repeated queries on compacted sketches. diff --git a/README.md b/README.md index 7ec87caf..9d6e73a8 100644 --- a/README.md +++ b/README.md @@ -35,7 +35,7 @@ [actions-badge]: https://github.com/apache/datasketches-rust/actions/workflows/ci.yml/badge.svg [actions-url]: https://github.com/apache/datasketches-rust/actions/workflows/ci.yml -Apache DataSketches Rust provides stochastic streaming algorithms for answering queries over large data sets with compact, mergeable summaries. It is the core Rust component of Apache DataSketches and currently implements a subset of the algorithms available in the other language components. +Apache DataSketches Rust provides stochastic streaming algorithms for answering queries over large data sets with compact, mergeable summaries. It is the core Rust component of Apache DataSketches. ## Getting started @@ -68,6 +68,7 @@ Enable multiple algorithms by listing their features together, such as `features | `bloom` | `BloomFilter` | Space-efficient probabilistic set membership with a configurable false-positive rate. | | `countmin` | `CountMinSketch` | Approximate point-frequency queries over a stream. | | `cpc` | `CpcSketch`, `CpcUnion`, `CpcWrapper` | Highly compact distinct-count estimation and unions. | +| `ddsketch` | `DdSketch` | Numeric quantiles with relative value accuracy and a fixed bucket budget. | | `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. | @@ -82,7 +83,7 @@ See the [API documentation](https://docs.rs/datasketches) for configuration, acc The minimum supported Rust version is 1.86.0. The crate currently supports little-endian targets only. -Supported serialization formats are tested with fixtures produced by Apache DataSketches Java, C++, and Go through the [DataSketches TCK](https://github.com/apache/datasketches-tck). +Each module documents its serialization format and compatibility guarantees. Cross-language formats are tested with fixtures produced by Apache DataSketches Java, C++, and Go through the [DataSketches TCK](https://github.com/apache/datasketches-tck). Serialization compatibility does not imply that an ordinary Rust `Hash` implementation produces the same update bytes across languages. When sketches must represent the same inputs across implementations, use `hash::value::{raw_bytes, canonical_float, sign_extend, natural_extend}` (and the constructors within those modules) to select matching input encodings. When interoperating with an implementation that skips empty strings, skip them before updating here as well. diff --git a/benchmarks/Cargo.toml b/benchmarks/Cargo.toml index 25af32a6..82580de2 100644 --- a/benchmarks/Cargo.toml +++ b/benchmarks/Cargo.toml @@ -27,6 +27,7 @@ datasketches = { workspace = true, features = [ "bloom", "countmin", "cpc", + "ddsketch", "frequencies", "hll", "kll", diff --git a/benchmarks/ddsketch/merge.rs b/benchmarks/ddsketch/merge.rs new file mode 100644 index 00000000..a9162277 --- /dev/null +++ b/benchmarks/ddsketch/merge.rs @@ -0,0 +1,41 @@ +// 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. + +use datasketches::ddsketch::DdSketch; +use divan::Bencher; +use divan::black_box; +use divan::counter::ItemsCount; + +use super::support; + +#[divan::bench(args = [8, 128, 4096])] +fn host_partials(bencher: Bencher, rows: usize) { + let values = support::values(64 * rows); + let partials: Vec<_> = values + .chunks(rows) + .map(|group| support::build(group, 2048)) + .collect(); + bencher + .counter(ItemsCount::new(partials.len())) + .bench_local(|| { + let mut sketch = DdSketch::default(); + for partial in black_box(&partials) { + sketch.merge(partial).unwrap(); + } + sketch + }); +} diff --git a/benchmarks/ddsketch/mod.rs b/benchmarks/ddsketch/mod.rs new file mode 100644 index 00000000..4ff11ff9 --- /dev/null +++ b/benchmarks/ddsketch/mod.rs @@ -0,0 +1,22 @@ +// 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. + +mod merge; +mod query; +mod serde; +mod support; +mod update; diff --git a/benchmarks/ddsketch/query.rs b/benchmarks/ddsketch/query.rs new file mode 100644 index 00000000..d6580362 --- /dev/null +++ b/benchmarks/ddsketch/query.rs @@ -0,0 +1,37 @@ +// 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. + +use divan::Bencher; +use divan::black_box; + +use super::support; + +#[divan::bench(args = [0.5, 0.99])] +fn quantile(bencher: Bencher, rank: f64) { + let sketch = support::build(&support::values(100_000), 2048); + bencher.bench_local(|| sketch.quantile(black_box(rank)).unwrap()); +} + +#[divan::bench] +fn percentiles(bencher: Bencher) { + let sketch = support::build(&support::values(100_000), 2048); + bencher.bench_local(|| { + sketch + .quantiles(black_box(&[0.5, 0.75, 0.9, 0.95, 0.99])) + .unwrap() + }); +} diff --git a/benchmarks/ddsketch/serde.rs b/benchmarks/ddsketch/serde.rs new file mode 100644 index 00000000..e292336a --- /dev/null +++ b/benchmarks/ddsketch/serde.rs @@ -0,0 +1,49 @@ +// 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. + +use datasketches::ddsketch::DdSketch; +use divan::Bencher; +use divan::black_box; +use divan::counter::BytesCount; + +use super::support; + +#[divan::bench] +fn serialize(bencher: Bencher) { + let sketch = support::build(&support::values(100_000), 2048); + bencher + .counter(BytesCount::new(sketch.serialize().len())) + .bench_local(|| sketch.serialize()); +} + +#[divan::bench] +fn serialize_reused(bencher: Bencher) { + let sketch = support::build(&support::values(100_000), 2048); + let mut out = sketch.serialize(); + bencher.counter(BytesCount::new(out.len())).bench_local(|| { + sketch.serialize_into(&mut out); + black_box(&out); + }); +} + +#[divan::bench] +fn deserialize(bencher: Bencher) { + let bytes = support::build(&support::values(100_000), 2048).serialize(); + bencher + .counter(BytesCount::new(bytes.len())) + .bench_local(|| DdSketch::deserialize(black_box(&bytes)).unwrap()); +} diff --git a/benchmarks/ddsketch/support.rs b/benchmarks/ddsketch/support.rs new file mode 100644 index 00000000..50b5d2f2 --- /dev/null +++ b/benchmarks/ddsketch/support.rs @@ -0,0 +1,37 @@ +// 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. + +use datasketches::ddsketch::DdSketch; + +pub fn values(len: usize) -> Vec { + let mut state = 42u64; + (0..len) + .map(|_| { + state = state.wrapping_mul(6364136223846793005).wrapping_add(1); + let u = ((state >> 11) as f64 + 1.0) / ((1u64 << 53) as f64 + 1.0); + 0.001 / u.sqrt() + }) + .collect() +} + +pub fn build(values: &[f64], limit: u32) -> DdSketch { + let mut sketch = DdSketch::new(limit, 0.01).unwrap(); + for &value in values { + sketch.update(value).unwrap(); + } + sketch +} diff --git a/benchmarks/ddsketch/update.rs b/benchmarks/ddsketch/update.rs new file mode 100644 index 00000000..9b07a6b0 --- /dev/null +++ b/benchmarks/ddsketch/update.rs @@ -0,0 +1,56 @@ +// 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. + +use datasketches::ddsketch::DdSketch; +use divan::Bencher; +use divan::black_box; +use divan::counter::ItemsCount; + +use super::support; + +#[divan::bench(args = [1_000, 100_000])] +fn update(bencher: Bencher, len: usize) { + let values = support::values(len); + bencher + .counter(ItemsCount::new(len)) + .bench_local(|| support::build(black_box(&values), 2048)); +} + +#[divan::bench] +fn collapsing(bencher: Bencher) { + let values = support::values(100_000); + bencher + .counter(ItemsCount::new(values.len())) + .bench_local(|| support::build(black_box(&values), 32)); +} + +#[divan::bench(args = [8, 128])] +fn reset_groups(bencher: Bencher, rows: usize) { + let values = support::values(64 * rows); + let mut sketch = DdSketch::default(); + bencher + .counter(ItemsCount::new(values.len())) + .bench_local(|| { + for group in black_box(&values).chunks(rows) { + sketch.reset(); + for &value in group { + sketch.update(value).unwrap(); + } + black_box(&sketch); + } + }); +} diff --git a/benchmarks/main.rs b/benchmarks/main.rs index ee288034..719c6580 100644 --- a/benchmarks/main.rs +++ b/benchmarks/main.rs @@ -23,6 +23,7 @@ static ALLOC: AllocProfiler = AllocProfiler::system(); mod bloom; mod countmin; mod cpc; +mod ddsketch; mod frequencies; mod hash_inputs; mod hll; diff --git a/datasketches/Cargo.toml b/datasketches/Cargo.toml index 2c931fc7..b9f0f0eb 100644 --- a/datasketches/Cargo.toml +++ b/datasketches/Cargo.toml @@ -41,6 +41,7 @@ default = [] bloom = [] countmin = [] cpc = [] +ddsketch = [] frequencies = [] hll = [] kll = [] @@ -53,6 +54,10 @@ tuple = [] name = "bloom_difference" required-features = ["bloom"] +[[example]] +name = "ddsketch_latency" +required-features = ["ddsketch"] + [dev-dependencies] googletest = { workspace = true } insta = { workspace = true } diff --git a/datasketches/examples/ddsketch_latency.rs b/datasketches/examples/ddsketch_latency.rs new file mode 100644 index 00000000..7d831548 --- /dev/null +++ b/datasketches/examples/ddsketch_latency.rs @@ -0,0 +1,57 @@ +// 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. + +//! Aggregates request latency distributions across hosts before querying service percentiles. +//! +//! Each host sends a sketch of latency observations in milliseconds. Integer weights represent +//! repeated observations with the same latency. Merging preserves those counts; averaging the +//! hosts' individual percentiles would not produce a service percentile. +//! The default collapse direction favors high quantiles. A collapsed query is reported explicitly +//! because it no longer has the configured relative accuracy. + +use datasketches::ddsketch::DdSketch; +use datasketches::error::Error; + +fn main() -> Result<(), Error> { + let hosts = [ + [(2.0, 900), (10.0, 90), (100.0, 10)], + [(5.0, 500), (20.0, 450), (500.0, 50)], + [(1.0, 950), (50.0, 40), (2000.0, 10)], + ]; + let mut service = DdSketch::new(2048, 0.01)?; + let mut host = service.clone(); + let mut message = Vec::new(); + for observations in hosts { + host.reset(); + for (milliseconds, requests) in observations { + host.update_weighted(milliseconds, requests)?; + } + host.serialize_into(&mut message); + service.merge(&DdSketch::deserialize(&message)?)?; + } + println!("Requests: {}", service.total_weight()); + for rank in [0.5, 0.95, 0.99] { + if let Some(value) = service.quantile(rank)? { + let collapsed = service.quantile_is_collapsed(rank)?.unwrap(); + println!( + "p{:.0}: {value:.3} ms (collapsed: {collapsed})", + rank * 100.0 + ); + } + } + Ok(()) +} diff --git a/datasketches/src/ddsketch/mapping.rs b/datasketches/src/ddsketch/mapping.rs new file mode 100644 index 00000000..6d2041d5 --- /dev/null +++ b/datasketches/src/ddsketch/mapping.rs @@ -0,0 +1,56 @@ +// 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. + +#[derive(Clone, Debug)] +pub struct Mapping { + log_gamma: f64, + log_scale: f64, +} + +impl Mapping { + pub fn new(accuracy: f64) -> Self { + Self { + // ln(1 + 2 alpha / (1 - alpha)) avoids cancellation for small alpha. + log_gamma: (2.0 * accuracy / (1.0 - accuracy)).ln_1p(), + log_scale: (-accuracy).ln_1p(), + } + } + + pub fn index(&self, magnitude: f64) -> i64 { + (magnitude.ln() / self.log_gamma).ceil() as i64 + } + + pub fn index_bounds(&self, magnitude: f64) -> (i64, i64) { + let coordinate = magnitude.ln() / self.log_gamma; + // Deserialization must accept a boundary value assigned to the adjacent + // bin by another platform's logarithm. Leave room for rounding in ln, + // the logarithmic base, and division without admitting arbitrary bins. + let tolerance = 8.0 * f64::EPSILON * coordinate.abs().max(1.0); + ( + (coordinate - tolerance).ceil() as i64, + (coordinate + tolerance).ceil() as i64, + ) + } + + pub fn value(&self, index: i64) -> f64 { + // Evaluate (1 - alpha) * gamma^index in log space: gamma^index itself + // may overflow even when the representative is finite. + (index as f64 * self.log_gamma + self.log_scale) + .exp() + .min(f64::MAX) + } +} diff --git a/datasketches/src/ddsketch/mod.rs b/datasketches/src/ddsketch/mod.rs new file mode 100644 index 00000000..0b8a6ad0 --- /dev/null +++ b/datasketches/src/ddsketch/mod.rs @@ -0,0 +1,83 @@ +// 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. + +//! Logarithmic histograms for quantiles with relative value accuracy. +//! +//! [`DdSketch`] implements [DDSketch][paper] for distributed aggregation of numeric data, +//! such as request latencies with long tails. For relative accuracy `alpha`, it maps each positive +//! value to `i = ceil(ln(x) / ln(gamma))`, where `gamma = (1 + alpha) / (1 - alpha)`, and +//! represents that bucket by `2 * gamma^i / (gamma + 1)`. Negative values use a mirrored store, +//! and zero has its own counter. Compatible sketches merge by adding bucket counts. +//! +//! # Accuracy and memory +//! +//! Queries approximate the lower discrete quantile, sorted item `floor(q * (n - 1))`, with +//! zero-based indexing and no interpolation between ranks. Endpoints return exact extrema. +//! An uncollapsed bucket's estimate `y` of a nonzero item `x` satisfies +//! `abs(y - x) / abs(x) <= alpha` in exact arithmetic. This bounds value error, unlike the +//! rank error controlled by KLL and REQ. Floating-point logarithms and exponentials introduce +//! rounding error; `alpha` is not an outward-rounded numerical certificate. Inputs with magnitude +//! below [`DdSketch::zero_threshold`] share the zero bucket, for which only absolute error below +//! that threshold is guaranteed. NaN and infinities are rejected. +//! +//! Each sign has a dense store with at most `max_buckets` counters, including empty gaps. +//! Storage grows lazily and is independent of stream length. Increasing the value range beyond +//! this budget collapses one end of each store. The default [`CollapseDirection::Lowest`] +//! sacrifices smaller values to retain high-quantile accuracy; [`CollapseDirection::Highest`] +//! sacrifices larger values instead. Directions refer to numerical value, including for negative +//! inputs. Zero is never collapsed. [`DdSketch::quantile_is_collapsed`] identifies queries whose +//! selected bucket has lost the relative accuracy guarantee. Uniformly collapsing sketches +//! instead coarsen the entire mapping, trading accuracy at every quantile for their memory bound. +//! +//! Updates inside the retained range take constant time. Extending a store, merging, and scalar +//! quantile queries take at most linear time in the bucket limit. The same mapping and limit are +//! required for merging, so a collapsed input cannot hide lost information in a finer store. +//! Insertion order and merge trees preserve bucket counts under the same mapping; platform math +//! functions can disagree at bucket boundaries, so cross-platform bitwise reproducibility is not +//! promised. [`DdSketch::reset`] and [`DdSketch::serialize_into`] allow allocation reuse. +//! +//! Serialization uses a versioned Rust-specific format. It is not a Datadog wire format or an +//! Apache DataSketches TCK format. Datadog's [sketches-go][reference] informs the operational +//! design, but its mapping conventions and treatment of negative collapsing stores differ. +//! +//! [paper]: https://www.vldb.org/pvldb/vol12/p2195-masson.pdf +//! [reference]: https://github.com/DataDog/sketches-go/tree/228b76a8e69df016726ce47579da54cffb0d0779/ddsketch +//! +//! # Examples +//! +//! ``` +//! use datasketches::ddsketch::DdSketch; +//! +//! let mut sketch = DdSketch::new(2048, 0.01)?; +//! for latency in [1.0, 2.0, 100.0] { +//! sketch.update(latency)?; +//! } +//! let median = sketch.quantile(0.5)?.unwrap(); +//! assert!((median - 2.0).abs() <= 0.01 * 2.0); +//! assert_eq!(sketch.quantile_is_collapsed(0.5)?, Some(false)); +//! let restored = DdSketch::deserialize(&sketch.serialize())?; +//! sketch.merge(&restored)?; +//! assert_eq!(sketch.total_weight(), 6); +//! # Ok::<(), datasketches::error::Error>(()) +//! ``` + +mod mapping; +mod sketch; +mod store; + +pub use self::sketch::CollapseDirection; +pub use self::sketch::DdSketch; diff --git a/datasketches/src/ddsketch/serialization.rs b/datasketches/src/ddsketch/serialization.rs new file mode 100644 index 00000000..03107d29 --- /dev/null +++ b/datasketches/src/ddsketch/serialization.rs @@ -0,0 +1,239 @@ +// 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. + +use super::CollapseDirection; +use super::DdSketch; +use super::Store; +use crate::codec::SketchSlice; +use crate::error::Error; + +const HEADER_SIZE: usize = 76; + +impl DdSketch { + /// Serializes the sketch using the Rust-specific version 1 format. + /// + /// The little-endian header contains `DDS1`, the bucket limit (`u32`), accuracy (`f64`), + /// direction (`u8`, lowest = `0`, highest = `1`), positive and negative collapse flags + /// (`u8` each), a reserved zero byte, total and zero weights (`u64` each), and extrema + /// (`f64` each). Negative and positive stores each have an offset (`i64`) and length (`u32`). + /// The payload contains their dense counts (`u64`), negative store first, in index order. + /// Empty extrema are positive and negative infinity. There is no cross-language format + /// compatibility contract. + pub fn serialize(&self) -> Vec { + let mut out = Vec::new(); + self.serialize_into(&mut out); + out + } + + /// Replaces `out` with the version 1 image, reusing its allocated capacity. + /// + /// Produces the same bytes as [`Self::serialize`]. + pub fn serialize_into(&self, out: &mut Vec) { + out.clear(); + out.reserve(HEADER_SIZE + 8 * (self.negative.counts.len() + self.positive.counts.len())); + out.extend_from_slice(b"DDS1"); + out.extend_from_slice(&self.max_buckets.to_le_bytes()); + out.extend_from_slice(&self.accuracy.to_le_bytes()); + out.push(match self.direction { + CollapseDirection::Lowest => 0, + CollapseDirection::Highest => 1, + }); + out.push(u8::from(self.positive.collapsed)); + out.push(u8::from(self.negative.collapsed)); + out.push(0); + out.extend_from_slice(&self.count.to_le_bytes()); + out.extend_from_slice(&self.zero_count.to_le_bytes()); + out.extend_from_slice(&self.min.to_le_bytes()); + out.extend_from_slice(&self.max.to_le_bytes()); + for store in [&self.negative, &self.positive] { + out.extend_from_slice(&store.offset.to_le_bytes()); + out.extend_from_slice(&(store.counts.len() as u32).to_le_bytes()); + } + for store in [&self.negative, &self.positive] { + for count in &store.counts { + out.extend_from_slice(&count.to_le_bytes()); + } + } + } + + /// Deserializes a Rust-specific version 1 image. + /// + /// Validates histogram structure, counts, extrema, and collapse boundaries, allowing for + /// floating-point rounding at logarithmic bucket boundaries. The image does + /// not authenticate the original observations or prove that collapse flags reflect their + /// history. Unmodified roundtrips preserve collapse state and subsequent update behavior. + /// + /// # Errors + /// + /// Returns `InvalidData` for unsupported versions, truncation, trailing data, invalid + /// parameters, invalid stores or extrema, and inconsistent or overflowing counts. + pub fn deserialize(bytes: &[u8]) -> Result { + let bad = || Error::deserial("invalid DDSketch image"); + if bytes.len() < HEADER_SIZE || &bytes[..4] != b"DDS1" { + return Err(bad()); + } + let mut input = SketchSlice::new(&bytes[4..]); + let limit = input.read_u32_le().map_err(|_| bad())?; + let accuracy = input.read_f64_le().map_err(|_| bad())?; + let direction = match input.read_u8().map_err(|_| bad())? { + 0 => CollapseDirection::Lowest, + 1 => CollapseDirection::Highest, + _ => return Err(bad()), + }; + let mut sketch = + Self::with_collapse_direction(limit, accuracy, direction).map_err(|_| bad())?; + let positive_collapsed = input.read_u8().map_err(|_| bad())?; + let negative_collapsed = input.read_u8().map_err(|_| bad())?; + if positive_collapsed > 1 + || negative_collapsed > 1 + || input.read_u8().map_err(|_| bad())? != 0 + { + return Err(bad()); + } + sketch.positive.collapsed = positive_collapsed != 0; + sketch.negative.collapsed = negative_collapsed != 0; + sketch.count = input.read_u64_le().map_err(|_| bad())?; + sketch.zero_count = input.read_u64_le().map_err(|_| bad())?; + sketch.min = input.read_f64_le().map_err(|_| bad())?; + sketch.max = input.read_f64_le().map_err(|_| bad())?; + sketch.negative.offset = input.read_i64_le().map_err(|_| bad())?; + let negative_len = input.read_u32_le().map_err(|_| bad())?; + sketch.positive.offset = input.read_i64_le().map_err(|_| bad())?; + let positive_len = input.read_u32_le().map_err(|_| bad())?; + // Check payload length before allocating from untrusted bucket counts. + if negative_len > limit + || positive_len > limit + || input.remaining().len() as u64 + != 8 * (u64::from(negative_len) + u64::from(positive_len)) + { + return Err(bad()); + } + let mut count = sketch.zero_count; + for (store, len) in [ + (&mut sketch.negative, negative_len), + (&mut sketch.positive, positive_len), + ] { + for _ in 0..len { + let weight = input.read_u64_le().map_err(|_| bad())?; + count = count.checked_add(weight).ok_or_else(bad)?; + store.counts.push_back(weight); + } + } + if count != sketch.count { + return Err(bad()); + } + sketch.validate_image()?; + Ok(sketch) + } + + fn validate_image(&self) -> Result<(), Error> { + let bad = || Error::deserial("inconsistent DDSketch histogram"); + let low = self.mapping.index_bounds(self.zero_threshold()).0; + let high = self.mapping.index_bounds(f64::MAX).1; + for (store, negative) in [(&self.negative, true), (&self.positive, false)] { + if store.counts.is_empty() { + if store.offset != 0 || store.collapsed { + return Err(bad()); + } + continue; + } + let (low, high) = if negative { (-high, -low) } else { (low, high) }; + let last = store.offset.checked_add(store.counts.len() as i64 - 1); + if store.offset < low + || last.is_none_or(|last| last > high) + || store.counts.front() == Some(&0) + || store.counts.back() == Some(&0) + || (store.collapsed && store.counts.len() != self.max_buckets as usize) + { + return Err(bad()); + } + } + if self.is_empty() { + if self.num_buckets() != 0 || self.min != f64::INFINITY || self.max != f64::NEG_INFINITY + { + return Err(bad()); + } + return Ok(()); + } + if !self.min.is_finite() + || !self.max.is_finite() + || self.min > self.max + || (self.count == 1 && self.min != self.max) + || self.min.to_bits() == (-0.0f64).to_bits() + || self.max.to_bits() == (-0.0f64).to_bits() + { + return Err(bad()); + } + + let category = |value: f64| { + if value <= -self.zero_threshold() { + -1 + } else if value >= self.zero_threshold() { + 1 + } else { + 0 + } + }; + let first = if !self.negative.counts.is_empty() { + -1 + } else if self.zero_count != 0 { + 0 + } else { + 1 + }; + let last = if !self.positive.counts.is_empty() { + 1 + } else if self.zero_count != 0 { + 0 + } else { + -1 + }; + if category(self.min) != first || category(self.max) != last { + return Err(bad()); + } + for (value, minimum) in [(self.min, true), (self.max, false)] { + if category(value) == 0 { + continue; + } + let (low, high) = self.mapping.index_bounds(value.abs()); + let (store, low, high): (&Store, i64, i64) = if value < 0.0 { + (&self.negative, -high, -low) + } else { + (&self.positive, low, high) + }; + let expected = if minimum { + store.offset + } else { + store.last_index() + }; + let retained = |index: i64| { + if store.collapsed { + match self.direction { + CollapseDirection::Lowest => index.max(store.offset), + CollapseDirection::Highest => index.min(store.last_index()), + } + } else { + index + } + }; + if !(retained(low)..=retained(high)).contains(&expected) { + return Err(bad()); + } + } + Ok(()) + } +} diff --git a/datasketches/src/ddsketch/sketch.rs b/datasketches/src/ddsketch/sketch.rs new file mode 100644 index 00000000..845eb999 --- /dev/null +++ b/datasketches/src/ddsketch/sketch.rs @@ -0,0 +1,372 @@ +// 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. + +use super::mapping::Mapping; +use super::store::Store; +use crate::error::Error; + +#[path = "serialization.rs"] +mod serialization; + +/// The numerical end of each signed store to collapse when its bucket limit is exceeded. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum CollapseDirection { + /// Collapses the smallest values, prioritizing high quantiles. + Lowest, + /// Collapses the largest values, prioritizing low quantiles. + Highest, +} + +/// A bounded, mergeable logarithmic histogram for relative-value quantile estimates. +/// +/// See the [module documentation](crate::ddsketch) for accuracy and memory tradeoffs. +#[derive(Clone, Debug)] +pub struct DdSketch { + max_buckets: u32, + accuracy: f64, + direction: CollapseDirection, + mapping: Mapping, + negative: Store, + positive: Store, + zero_count: u64, + count: u64, + min: f64, + max: f64, +} + +impl Default for DdSketch { + /// Creates a sketch with `2048` buckets per sign, `0.01` accuracy, and lowest-value collapse. + fn default() -> Self { + Self::new(2048, 0.01).unwrap() + } +} + +impl DdSketch { + /// Creates a sketch that collapses the smallest values when either store exceeds its limit. + /// + /// `max_buckets` limits each sign separately and includes empty bins between retained values. + /// The zero counter is additional. With `2048` buckets and `0.01` relative accuracy, a positive + /// range from approximately `80` microseconds to one year fits without collapse. + /// + /// # Errors + /// + /// Returns `InvalidArgument` if `max_buckets` is zero or `relative_accuracy` is outside + /// `[1e-9, 1)`, including NaN. The lower accuracy limit keeps the full normal `f64` domain + /// indexable and limits the effect of floating-point rounding on the requested precision. + pub fn new(max_buckets: u32, relative_accuracy: f64) -> Result { + Self::with_collapse_direction(max_buckets, relative_accuracy, CollapseDirection::Lowest) + } + + /// Creates a sketch with the specified collapse direction. + /// + /// Both the budget and direction apply to each sign separately. Directions are in numerical + /// order, so `Lowest` collapses large negative magnitudes and small positive magnitudes. + /// + /// # Errors + /// + /// Returns `InvalidArgument` for the same invalid parameters as [`Self::new`]. + pub fn with_collapse_direction( + max_buckets: u32, + relative_accuracy: f64, + direction: CollapseDirection, + ) -> Result { + if max_buckets == 0 || !(1e-9..1.0).contains(&relative_accuracy) { + return Err(Error::invalid_argument( + "max_buckets must be positive and relative_accuracy must be in [1e-9, 1)", + )); + } + Ok(Self { + max_buckets, + accuracy: relative_accuracy, + direction, + mapping: Mapping::new(relative_accuracy), + negative: Store::default(), + positive: Store::default(), + zero_count: 0, + count: 0, + min: f64::INFINITY, + max: f64::NEG_INFINITY, + }) + } + + /// Removes all values and collapse state, retaining configuration and allocated capacity. + pub fn reset(&mut self) { + self.negative.reset(); + self.positive.reset(); + self.zero_count = 0; + self.count = 0; + self.min = f64::INFINITY; + self.max = f64::NEG_INFINITY; + } + + /// Inserts one finite value, normalizing signed zeros to positive zero. + /// + /// # Errors + /// + /// Returns `InvalidArgument` for a non-finite value or total weight overflow, leaving the + /// sketch unchanged. + pub fn update(&mut self, value: f64) -> Result<(), Error> { + self.update_weighted(value, 1) + } + + /// Inserts `weight` copies of a finite value. + /// + /// A zero weight has no effect, but the value is still validated. Subnormal values share + /// the zero counter; exact extrema still include them. + /// + /// # Errors + /// + /// Returns `InvalidArgument` for a non-finite value or total weight overflow, leaving the + /// sketch unchanged. + pub fn update_weighted(&mut self, value: f64, weight: u64) -> Result<(), Error> { + if !value.is_finite() { + return Err(Error::invalid_argument("value must be finite")); + } + let count = self + .count + .checked_add(weight) + .ok_or_else(|| Error::invalid_argument("total weight overflow"))?; + if weight == 0 { + return Ok(()); + } + let value = if value == 0.0 { 0.0 } else { value }; + if value.abs() < self.zero_threshold() { + self.zero_count += weight; + } else { + let index = self.mapping.index(value.abs()); + if value < 0.0 { + self.negative + .add(-index, weight, self.max_buckets, self.direction); + } else { + self.positive + .add(index, weight, self.max_buckets, self.direction); + } + } + self.count = count; + self.min = self.min.min(value); + self.max = self.max.max(value); + Ok(()) + } + + /// Merges a compatible sketch without modifying the input. + /// + /// Accuracy, bucket limit, and collapse direction must match exactly, even for empty inputs. + /// Previously collapsed buckets retain their loss of accuracy after merging. + /// + /// # Errors + /// + /// Returns `InvalidArgument` for incompatible parameters or total weight overflow, leaving + /// the receiver unchanged. + pub fn merge(&mut self, other: &Self) -> Result<(), Error> { + if self.max_buckets != other.max_buckets + || self.accuracy.to_bits() != other.accuracy.to_bits() + || self.direction != other.direction + { + return Err(Error::invalid_argument("incompatible DDSketch parameters")); + } + let count = self + .count + .checked_add(other.count) + .ok_or_else(|| Error::invalid_argument("total weight overflow"))?; + self.negative + .merge(&other.negative, self.max_buckets, self.direction); + self.positive + .merge(&other.positive, self.max_buckets, self.direction); + self.zero_count += other.zero_count; + self.count = count; + self.min = self.min.min(other.min); + self.max = self.max.max(other.max); + Ok(()) + } + + /// Returns the total number of inserted values, including weights. + pub fn total_weight(&self) -> u64 { + self.count + } + + /// Returns whether the sketch contains no values. + pub fn is_empty(&self) -> bool { + self.count == 0 + } + + /// Returns the configured dense bucket limit per sign, excluding the zero counter. + pub fn max_buckets(&self) -> u32 { + self.max_buckets + } + + /// Returns the configured relative accuracy for uncollapsed, nonzero buckets. + /// + /// Floating-point rounding and the zero threshold qualify this bound; see the + /// [module documentation](crate::ddsketch). + pub fn relative_accuracy(&self) -> f64 { + self.accuracy + } + + /// Returns the configured numerical collapse direction. + pub fn collapse_direction(&self) -> CollapseDirection { + self.direction + } + + /// Returns the magnitude below which inputs share the zero counter (`f64::MIN_POSITIVE`). + /// + /// Values at the threshold are indexed normally. Exact extrema include smaller inputs. + pub fn zero_threshold(&self) -> f64 { + f64::MIN_POSITIVE + } + + /// Returns the weight of zeros and subnormal values. + pub fn zero_count(&self) -> u64 { + self.zero_count + } + + /// Returns the retained dense bucket count, including empty gaps and a nonempty zero bucket. + pub fn num_buckets(&self) -> usize { + self.negative.counts.len() + self.positive.counts.len() + usize::from(self.zero_count != 0) + } + + /// Returns whether either signed store has collapsed any buckets since the last reset. + pub fn is_collapsed(&self) -> bool { + self.negative.collapsed || self.positive.collapsed + } + + /// Returns the exact minimum, or `None` when empty. + pub fn min_value(&self) -> Option { + (!self.is_empty()).then_some(self.min) + } + + /// Returns the exact maximum, or `None` when empty. + pub fn max_value(&self) -> Option { + (!self.is_empty()).then_some(self.max) + } + + /// Returns an estimate of the lower discrete quantile at `rank`, or `None` when empty. + /// + /// The target is sorted item `floor(rank * (n - 1))` (zero-based). The multiplication is + /// evaluated exactly for the supplied binary floating-point rank, even above `2^53` weight. + /// Ranks `0` and `1` return the exact extrema. Other estimates are clamped to those extrema. + /// Use [`Self::quantile_is_collapsed`] to detect queries affected by bucket collapse. + /// + /// # Errors + /// + /// Returns `InvalidArgument` if `rank` is outside `[0, 1]`, including NaN, even when empty. + pub fn quantile(&self, rank: f64) -> Result, Error> { + Ok(self.query(rank)?.map(|(value, _)| value)) + } + + /// Returns quantile estimates in input order, or `None` when empty. + /// + /// Uses the same semantics as [`Self::quantile`]. An empty rank slice returns an empty vector + /// for a nonempty sketch. + /// + /// # Errors + /// + /// Returns `InvalidArgument` if any rank is outside `[0, 1]`, including NaN, even when empty. + pub fn quantiles(&self, ranks: &[f64]) -> Result>, Error> { + for &rank in ranks { + validate_rank(rank)?; + } + if self.is_empty() { + return Ok(None); + } + ranks + .iter() + .map(|&rank| self.quantile(rank).map(Option::unwrap)) + .collect::, _>>() + .map(Some) + } + + /// Returns whether the quantile estimate uses a collapsed bucket, or `None` when empty. + /// + /// `true` means that the configured relative accuracy no longer applies to this query. + /// Exact extrema and constant streams return `false`. A zero-bucket query also returns + /// `false`, but is subject to the absolute error described by [`Self::zero_threshold`]. + /// + /// # Errors + /// + /// Returns `InvalidArgument` if `rank` is outside `[0, 1]`, including NaN, even when empty. + pub fn quantile_is_collapsed(&self, rank: f64) -> Result, Error> { + Ok(self.query(rank)?.map(|(_, collapsed)| collapsed)) + } + + /// Returns an implementation-dependent heap memory estimate including reserved capacity. + pub fn estimated_size(&self) -> usize { + (self.negative.counts.capacity() + self.positive.counts.capacity()) * size_of::() + } + + fn query(&self, rank: f64) -> Result, Error> { + validate_rank(rank)?; + if self.is_empty() { + return Ok(None); + } + if rank == 0.0 || self.min == self.max { + return Ok(Some((self.min, false))); + } + if rank == 1.0 { + return Ok(Some((self.max, false))); + } + let mut target = quantile_index(rank, self.count); + for (store, negative) in [(&self.negative, true), (&self.positive, false)] { + if !negative { + if target < self.zero_count { + return Ok(Some((0.0f64.clamp(self.min, self.max), false))); + } + target -= self.zero_count; + } + for (position, &weight) in store.counts.iter().enumerate() { + if target < weight { + let index = store.offset + position as i64; + let value = if negative { + -self.mapping.value(-index) + } else { + self.mapping.value(index) + }; + return Ok(Some(( + value.clamp(self.min, self.max), + store.is_collapsed_index(index, self.direction), + ))); + } + target -= weight; + } + } + unreachable!("validated counts cover the target rank") + } +} + +fn validate_rank(rank: f64) -> Result<(), Error> { + if !(0.0..=1.0).contains(&rank) { + return Err(Error::invalid_argument("rank must be in [0, 1]")); + } + Ok(()) +} + +fn quantile_index(rank: f64, count: u64) -> u64 { + // Multiply the exact binary significand by the integer weight before shifting. + // Converting count to f64 first can select a different bucket above 2^53. + let bits = rank.to_bits(); + let exponent = ((bits >> 52) & 0x7ff) as u32; + if exponent == 0 { + return 0; + } + let significand = (bits & ((1u64 << 52) - 1)) | (1u64 << 52); + let product = u128::from(significand) * u128::from(count - 1); + let shift = 1075 - exponent; + if shift >= 128 { + 0 + } else { + (product >> shift) as u64 + } +} diff --git a/datasketches/src/ddsketch/store.rs b/datasketches/src/ddsketch/store.rs new file mode 100644 index 00000000..c2be1549 --- /dev/null +++ b/datasketches/src/ddsketch/store.rs @@ -0,0 +1,118 @@ +// 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. + +use std::collections::VecDeque; + +use super::CollapseDirection; + +// Indices increase in numerical value order, including in the negative store. +// A collapsed store retains a full window; its boundary absorbs all omitted values. +#[derive(Clone, Debug, Default)] +pub struct Store { + pub offset: i64, + pub counts: VecDeque, + pub collapsed: bool, +} + +impl Store { + pub fn reset(&mut self) { + self.offset = 0; + self.counts.clear(); + self.collapsed = false; + } + + pub fn last_index(&self) -> i64 { + self.offset + self.counts.len() as i64 - 1 + } + + pub fn add(&mut self, index: i64, weight: u64, limit: u32, direction: CollapseDirection) { + if self.counts.is_empty() { + self.offset = index; + self.counts.push_back(weight); + return; + } + + let index = if self.collapsed { + match direction { + CollapseDirection::Lowest => index.max(self.offset), + CollapseDirection::Highest => index.min(self.last_index()), + } + } else { + index + }; + if index < self.offset || index > self.last_index() { + self.extend(index, limit, direction); + } + let index = index.clamp(self.offset, self.last_index()); + // The sketch checks the combined total before any store mutation. + self.counts[(index - self.offset) as usize] += weight; + } + + pub fn merge(&mut self, other: &Self, limit: u32, direction: CollapseDirection) { + for (position, &weight) in other.counts.iter().enumerate() { + if weight != 0 { + self.add(other.offset + position as i64, weight, limit, direction); + } + } + // Equal limits imply that inserting the other store's endpoints aligns + // the collapsed boundary before its loss of information is propagated. + self.collapsed |= other.collapsed; + } + + pub fn is_collapsed_index(&self, index: i64, direction: CollapseDirection) -> bool { + self.collapsed + && index + == match direction { + CollapseDirection::Lowest => self.offset, + CollapseDirection::Highest => self.last_index(), + } + } + + fn extend(&mut self, index: i64, limit: u32, direction: CollapseDirection) { + let mut low = self.offset.min(index); + let mut high = self.last_index().max(index); + if high - low + 1 > i64::from(limit) { + match direction { + CollapseDirection::Lowest => low = high - i64::from(limit) + 1, + CollapseDirection::Highest => high = low + i64::from(limit) - 1, + } + self.collapsed = true; + } + + let mut lower_weight = 0; + while !self.counts.is_empty() && self.offset < low { + lower_weight += self.counts.pop_front().unwrap(); + self.offset += 1; + } + let mut upper_weight = 0; + while !self.counts.is_empty() && self.last_index() > high { + upper_weight += self.counts.pop_back().unwrap(); + } + if self.counts.is_empty() { + self.offset = low; + } + while self.offset > low { + self.counts.push_front(0); + self.offset -= 1; + } + // Work and allocation depend on the retained window, never on the size + // of an index gap (which can span hundreds of billions of empty bins). + self.counts.resize((high - low + 1) as usize, 0); + *self.counts.front_mut().unwrap() += lower_weight; + *self.counts.back_mut().unwrap() += upper_weight; + } +} diff --git a/datasketches/src/lib.rs b/datasketches/src/lib.rs index 813a841b..abecc7ee 100644 --- a/datasketches/src/lib.rs +++ b/datasketches/src/lib.rs @@ -40,6 +40,7 @@ //! * Use `kll`, `req`, or `tdigest` for ranks and quantiles. KLL provides strong general-purpose //! rank accuracy, REQ targets configurable high- or low-rank accuracy, and T-Digest emphasizes //! distribution tails. +//! * Use `ddsketch` for numeric quantiles with relative value accuracy and a fixed bucket budget. //! * Use `tuple` when retained Theta keys need application-defined summaries. //! //! See each module's documentation for accuracy, memory, serialization, and update examples. @@ -73,6 +74,8 @@ pub mod bloom; pub mod countmin; #[cfg(feature = "cpc")] pub mod cpc; +#[cfg(feature = "ddsketch")] +pub mod ddsketch; #[cfg(feature = "frequencies")] pub mod frequencies; #[cfg(feature = "hll")] diff --git a/tests-integration/Cargo.toml b/tests-integration/Cargo.toml index 79021d6c..eb11db58 100644 --- a/tests-integration/Cargo.toml +++ b/tests-integration/Cargo.toml @@ -27,6 +27,7 @@ datasketches = { workspace = true, features = [ "bloom", "countmin", "cpc", + "ddsketch", "frequencies", "hll", "kll", diff --git a/tests-integration/tests/ddsketch_test/accuracy.rs b/tests-integration/tests/ddsketch_test/accuracy.rs new file mode 100644 index 00000000..e7aed059 --- /dev/null +++ b/tests-integration/tests/ddsketch_test/accuracy.rs @@ -0,0 +1,179 @@ +// 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. + +use datasketches::ddsketch::CollapseDirection; +use datasketches::ddsketch::DdSketch; + +use super::assert_relative_close; +use super::next_random; + +fn check_quantiles( + mut values: Vec, + limit: u32, + accuracy: f64, + direction: CollapseDirection, +) -> DdSketch { + let mut sketch = DdSketch::with_collapse_direction(limit, accuracy, direction).unwrap(); + for &value in &values { + sketch.update(value).unwrap(); + assert!(sketch.num_buckets() <= 2 * limit as usize + 1); + } + values.sort_by(f64::total_cmp); + let mut previous = f64::NEG_INFINITY; + for i in 0..=1024 { + let q = i as f64 / 1024.0; + let exact = values[i * (values.len() - 1) / 1024]; + let estimate = sketch.quantile(q).unwrap().unwrap(); + assert!( + estimate >= previous && estimate >= values[0] && estimate <= *values.last().unwrap() + ); + previous = estimate; + if !sketch.quantile_is_collapsed(q).unwrap().unwrap() { + if exact.abs() < sketch.zero_threshold() { + assert!((estimate - exact).abs() < sketch.zero_threshold()); + } else { + assert_relative_close(estimate, exact, accuracy); + } + } + } + assert_eq!(sketch.quantile(0.0).unwrap(), values.first().copied()); + assert_eq!(sketch.quantile(1.0).unwrap(), values.last().copied()); + sketch +} + +#[test] +fn value_accuracy_across_distributions() { + let mut state = 123; + let mut uniform = Vec::new(); + let mut log_normal = Vec::new(); + let mut pareto = Vec::new(); + let mut repeated = Vec::new(); + for i in 0..4097 { + let u = ((next_random(&mut state) >> 11) as f64 + 0.5) / (1u64 << 53) as f64; + let v = ((next_random(&mut state) >> 11) as f64 + 0.5) / (1u64 << 53) as f64; + uniform.push(1000.0 * (u - 0.5)); + log_normal.push((3.0 * (-2.0 * u.ln()).sqrt() * (std::f64::consts::TAU * v).cos()).exp()); + pareto.push(u.powf(-2.0)); + repeated.push([0.0, 0.001, 10.0, 10000.0][i % 4]); + } + for values in [uniform, log_normal, pareto, repeated] { + for direction in [CollapseDirection::Lowest, CollapseDirection::Highest] { + assert!(!check_quantiles(values.clone(), 2048, 0.01, direction).is_collapsed()); + assert!(check_quantiles(values.clone(), 32, 0.01, direction).is_collapsed()); + } + } +} + +#[test] +fn floating_point_domain_extremes_and_sparse_gaps() { + let mut state = 42; + let mut values = vec![ + f64::MAX, + -f64::MAX, + f64::MIN_POSITIVE, + -f64::MIN_POSITIVE, + 0.0, + ]; + for _ in 0..4092 { + let bits = next_random(&mut state); + let value = f64::from_bits(bits); + values.push(if value.is_finite() { value } else { 0.0 }); + } + for direction in [CollapseDirection::Lowest, CollapseDirection::Highest] { + assert!(!check_quantiles(values.clone(), 100_000, 0.01, direction).is_collapsed()); + assert!(check_quantiles(values.clone(), 8, 1e-9, direction).is_collapsed()); + } +} + +#[test] +fn logarithmic_boundaries_and_supported_accuracy_extremes() { + for accuracy in [ + 1e-9, + 0.001, + 0.01, + 1.0 / 3.0, + 0.9, + f64::from_bits(1.0f64.to_bits() - 1), + ] { + for exponent in [-1022, -100, -1, 0, 1, 100, 1023] { + let center = 2.0f64.powi(exponent); + let mut values = vec![ + center, + f64::from_bits(center.to_bits() - 1), + f64::from_bits(center.to_bits() + 1), + ]; + for factor in [1.0 - accuracy, 1.0 + accuracy] { + let value = center * factor; + if value.is_finite() { + values.push(value); + } + } + for direction in [CollapseDirection::Lowest, CollapseDirection::Highest] { + check_quantiles(values.clone(), 64, accuracy, direction); + check_quantiles(values.iter().map(|v| -v).collect(), 64, accuracy, direction); + } + } + } +} + +#[test] +fn collapse_direction_follows_numeric_order_for_each_sign() { + for sign in [-1.0, 1.0] { + for direction in [CollapseDirection::Lowest, CollapseDirection::Highest] { + let mut sketch = DdSketch::with_collapse_direction(4, 0.1, direction).unwrap(); + let mut values = Vec::new(); + for i in 0..17 { + let value = sign * 1.3f64.powi(i); + values.push(value); + sketch.update(value).unwrap(); + } + values.sort_by(f64::total_cmp); + assert!(sketch.is_collapsed()); + let (lost, retained) = match direction { + CollapseDirection::Lowest => (0.125, 0.9375), + CollapseDirection::Highest => (0.9375, 0.0625), + }; + assert_eq!(sketch.quantile_is_collapsed(lost).unwrap(), Some(true)); + assert_eq!(sketch.quantile_is_collapsed(retained).unwrap(), Some(false)); + assert_relative_close( + sketch.quantile(retained).unwrap().unwrap(), + values[(retained * 16.0) as usize], + 0.1, + ); + for q in [0.0, 1.0] { + assert_eq!(sketch.quantile_is_collapsed(q).unwrap(), Some(false)); + } + } + } +} + +#[test] +fn single_bucket_preserves_zeros_and_exact_extrema() { + for direction in [CollapseDirection::Lowest, CollapseDirection::Highest] { + let mut sketch = DdSketch::with_collapse_direction(1, 0.01, direction).unwrap(); + for value in [-1e300, -10.0, 0.0, 10.0, 1e300] { + sketch.update(value).unwrap(); + } + assert_eq!(sketch.num_buckets(), 3); + assert_eq!(sketch.quantile(0.5).unwrap(), Some(0.0)); + assert_eq!(sketch.quantile_is_collapsed(0.5).unwrap(), Some(false)); + assert_eq!(sketch.min_value(), Some(-1e300)); + assert_eq!(sketch.max_value(), Some(1e300)); + assert_eq!(sketch.quantile_is_collapsed(0.25).unwrap(), Some(true)); + assert_eq!(sketch.quantile_is_collapsed(0.75).unwrap(), Some(true)); + } +} diff --git a/tests-integration/tests/ddsketch_test/main.rs b/tests-integration/tests/ddsketch_test/main.rs new file mode 100644 index 00000000..767a8072 --- /dev/null +++ b/tests-integration/tests/ddsketch_test/main.rs @@ -0,0 +1,40 @@ +// 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. + +mod accuracy; +mod merge; +mod update; + +fn assert_relative_close(estimate: f64, exact: f64, accuracy: f64) { + assert!(estimate.is_finite()); + if exact == 0.0 { + assert_eq!(estimate, 0.0); + } else { + let error = (estimate - exact).abs() / exact.abs(); + assert!( + error <= accuracy + 2e-12, + "{estimate} vs {exact}: error {error}, accuracy {accuracy}" + ); + } +} + +fn next_random(state: &mut u64) -> u64 { + *state = state + .wrapping_mul(6364136223846793005) + .wrapping_add(1442695040888963407); + *state +} diff --git a/tests-integration/tests/ddsketch_test/merge.rs b/tests-integration/tests/ddsketch_test/merge.rs new file mode 100644 index 00000000..fe19f8ae --- /dev/null +++ b/tests-integration/tests/ddsketch_test/merge.rs @@ -0,0 +1,128 @@ +// 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. + +use datasketches::ddsketch::CollapseDirection; +use datasketches::ddsketch::DdSketch; +use datasketches::error::ErrorKind; + +use super::next_random; + +#[test] +fn partitioning_order_and_merge_trees_preserve_histogram() { + let mut state = 1337; + let values: Vec<_> = (0..4096) + .map(|i| { + let magnitude = 1.7f64.powi((next_random(&mut state) % 1200) as i32 - 600); + let value = match i % 3 { + 0 => -magnitude, + 1 => 0.0, + _ => magnitude, + }; + (value, next_random(&mut state) % 100 + 1) + }) + .collect(); + for limit in [1, 2, 16, 2048] { + for direction in [CollapseDirection::Lowest, CollapseDirection::Highest] { + let empty = DdSketch::with_collapse_direction(limit, 0.01, direction).unwrap(); + let mut direct = empty.clone(); + let mut reversed = empty.clone(); + let mut partitions = vec![empty.clone(); 16]; + for (i, &(value, weight)) in values.iter().enumerate() { + direct.update_weighted(value, weight).unwrap(); + partitions[i % 16].update_weighted(value, weight).unwrap(); + } + for &(value, weight) in values.iter().rev() { + reversed.update_weighted(value, weight).unwrap(); + } + assert_eq!(direct.serialize(), reversed.serialize()); + let mut sequential = empty.clone(); + for partition in &partitions { + let before = partition.serialize(); + sequential.merge(partition).unwrap(); + assert_eq!(partition.serialize(), before); + } + assert_eq!(sequential.serialize(), direct.serialize()); + while partitions.len() > 1 { + partitions = partitions + .chunks(2) + .map(|pair| { + let mut merged = DdSketch::deserialize(&pair[0].serialize()).unwrap(); + merged.merge(&pair[1]).unwrap(); + merged + }) + .collect(); + } + assert_eq!(partitions[0].serialize(), direct.serialize()); + let mut adopted = empty.clone(); + adopted.merge(&direct).unwrap(); + adopted.merge(&empty).unwrap(); + assert_eq!(adopted.serialize(), direct.serialize()); + } + } +} + +#[test] +fn collapsed_boundary_is_preserved_when_merging_new_values() { + for direction in [CollapseDirection::Lowest, CollapseDirection::Highest] { + let mut left = DdSketch::with_collapse_direction(5, 0.01, direction).unwrap(); + let mut right = left.clone(); + let mut direct = left.clone(); + for value in [-1000.0, -1.0, 1.0, 1000.0] { + left.update(value).unwrap(); + direct.update(value).unwrap(); + } + for value in [-5.0, -2.0, 0.0, 2.0, 5.0] { + right.update(value).unwrap(); + direct.update(value).unwrap(); + } + let mut reverse = right.clone(); + reverse.merge(&left).unwrap(); + left.merge(&right).unwrap(); + assert_eq!(left.serialize(), direct.serialize()); + assert_eq!(reverse.serialize(), direct.serialize()); + for i in 0..=32 { + let q = i as f64 / 32.0; + assert_eq!( + left.quantile_is_collapsed(q).unwrap(), + direct.quantile_is_collapsed(q).unwrap() + ); + } + } +} + +#[test] +fn incompatible_and_overflowing_merges_are_atomic() { + let mut sketch = DdSketch::default(); + sketch.update_weighted(2.0, u64::MAX).unwrap(); + let before = sketch.serialize(); + let mut overflowing = DdSketch::default(); + overflowing.update(-100.0).unwrap(); + for other in [ + DdSketch::new(128, 0.01).unwrap(), + DdSketch::new(2048, 0.02).unwrap(), + DdSketch::with_collapse_direction(2048, 0.01, CollapseDirection::Highest).unwrap(), + overflowing, + ] { + assert_eq!( + sketch.merge(&other).unwrap_err().kind(), + ErrorKind::InvalidArgument + ); + assert_eq!(sketch.serialize(), before); + } + let mut empty = DdSketch::default(); + assert!(empty.merge(&DdSketch::new(1, 0.01).unwrap()).is_err()); +} diff --git a/tests-integration/tests/ddsketch_test/update.rs b/tests-integration/tests/ddsketch_test/update.rs new file mode 100644 index 00000000..0464f7a8 --- /dev/null +++ b/tests-integration/tests/ddsketch_test/update.rs @@ -0,0 +1,167 @@ +// 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. + +use datasketches::ddsketch::CollapseDirection; +use datasketches::ddsketch::DdSketch; +use datasketches::error::ErrorKind; + +#[test] +fn empty_and_invalid_queries() { + let sketch = DdSketch::default(); + assert!(sketch.is_empty()); + assert_eq!(sketch.min_value(), None); + assert_eq!(sketch.max_value(), None); + assert_eq!(sketch.quantile(0.5).unwrap(), None); + assert_eq!(sketch.quantiles(&[]).unwrap(), None); + assert_eq!(sketch.quantile_is_collapsed(0.5).unwrap(), None); + for rank in [f64::NAN, f64::INFINITY, f64::NEG_INFINITY, -0.1, 1.1] { + assert_eq!( + sketch.quantile(rank).unwrap_err().kind(), + ErrorKind::InvalidArgument + ); + assert_eq!( + sketch.quantiles(&[0.5, rank]).unwrap_err().kind(), + ErrorKind::InvalidArgument + ); + assert_eq!( + sketch.quantile_is_collapsed(rank).unwrap_err().kind(), + ErrorKind::InvalidArgument + ); + } +} + +#[test] +fn invalid_configuration() { + for accuracy in [ + f64::NAN, + f64::INFINITY, + f64::NEG_INFINITY, + -0.1, + 0.0, + 1e-10, + 1.0, + ] { + assert_eq!( + DdSketch::new(2048, accuracy).unwrap_err().kind(), + ErrorKind::InvalidArgument + ); + } + assert!(DdSketch::new(0, 0.01).is_err()); +} + +#[test] +fn invalid_updates_and_overflow_are_atomic() { + let mut sketch = DdSketch::default(); + sketch.update_weighted(5.0, u64::MAX).unwrap(); + let before = sketch.serialize(); + for value in [f64::NAN, f64::INFINITY, f64::NEG_INFINITY] { + for weight in [0, 1] { + assert_eq!( + sketch.update_weighted(value, weight).unwrap_err().kind(), + ErrorKind::InvalidArgument + ); + assert_eq!(sketch.serialize(), before); + } + } + assert!(sketch.update(-10.0).is_err()); + assert_eq!(sketch.serialize(), before); + sketch.update_weighted(-10.0, 0).unwrap(); + assert_eq!(sketch.serialize(), before); + assert_eq!(sketch.total_weight(), u64::MAX); +} + +#[test] +fn weighted_updates_match_repeated_values() { + let mut weighted = DdSketch::new(16, 0.01).unwrap(); + let mut repeated = weighted.clone(); + for (value, weight) in [(-500.0, 7), (-2.0, 3), (0.0, 5), (3.0, 11), (1000.0, 4)] { + weighted.update_weighted(value, weight).unwrap(); + for _ in 0..weight { + repeated.update(value).unwrap(); + } + } + assert_eq!(weighted.serialize(), repeated.serialize()); + let ranks = [1.0, 0.5, 0.0, 0.1, 0.5]; + let scalar: Vec<_> = ranks + .iter() + .map(|&q| weighted.quantile(q).unwrap().unwrap()) + .collect(); + assert_eq!(weighted.quantiles(&ranks).unwrap(), Some(scalar)); + assert_eq!(weighted.quantiles(&[]).unwrap(), Some(vec![])); +} + +#[test] +fn signed_zero_and_subnormal_inputs() { + let mut sketch = DdSketch::default(); + for value in [-0.0, 0.0, -f64::from_bits(1), f64::from_bits(1)] { + sketch.update(value).unwrap(); + } + assert_eq!(sketch.zero_count(), 4); + assert_eq!(sketch.quantile(0.5).unwrap(), Some(0.0)); + assert_eq!(sketch.min_value(), Some(-f64::from_bits(1))); + assert_eq!(sketch.max_value(), Some(f64::from_bits(1))); + assert_eq!(sketch.quantile(0.0).unwrap(), sketch.min_value()); + assert_eq!(sketch.quantile(1.0).unwrap(), sketch.max_value()); + sketch.update(f64::MIN_POSITIVE).unwrap(); + sketch.update(-f64::MIN_POSITIVE).unwrap(); + assert_eq!(sketch.zero_count(), 4); + + let mut zero = DdSketch::default(); + zero.update(-0.0).unwrap(); + assert_eq!(zero.min_value().unwrap().to_bits(), 0.0f64.to_bits()); + assert_eq!(zero.max_value().unwrap().to_bits(), 0.0f64.to_bits()); +} + +#[test] +fn full_u64_weight_selects_the_exact_discrete_rank() { + let mut sketch = DdSketch::default(); + sketch.update_weighted(1.0, 1 << 63).unwrap(); + sketch.update_weighted(100.0, (1 << 63) - 1).unwrap(); + assert_eq!(sketch.total_weight(), u64::MAX); + assert_eq!(sketch.quantile(0.5).unwrap(), Some(1.0)); + assert!( + sketch + .quantile(f64::from_bits(0.5f64.to_bits() + 1)) + .unwrap() + .unwrap() + > 99.0 + ); + assert_eq!(sketch.quantile(f64::from_bits(1)).unwrap(), Some(1.0)); + assert_eq!(sketch.quantile(1.0).unwrap(), Some(100.0)); +} + +#[test] +fn reset_reuses_storage_and_clears_collapse() { + let mut sketch = + DdSketch::with_collapse_direction(8, 0.01, CollapseDirection::Highest).unwrap(); + for value in [-1e200, -1e-200, 0.0, 1e-200, 1e200] { + sketch.update(value).unwrap(); + } + assert!(sketch.is_collapsed()); + assert_eq!(sketch.num_buckets(), 17); + let size = sketch.estimated_size(); + sketch.reset(); + assert!(sketch.is_empty()); + assert!(!sketch.is_collapsed()); + assert_eq!(sketch.estimated_size(), size); + let mut fresh = DdSketch::with_collapse_direction(8, 0.01, CollapseDirection::Highest).unwrap(); + for value in [-20.0, 0.0, 10.0] { + sketch.update(value).unwrap(); + fresh.update(value).unwrap(); + } + assert_eq!(sketch.serialize(), fresh.serialize()); +} diff --git a/tests-integration/tests/serde_tests.rs b/tests-integration/tests/serde_tests.rs index bd723168..741c0a7c 100644 --- a/tests-integration/tests/serde_tests.rs +++ b/tests-integration/tests/serde_tests.rs @@ -51,6 +51,9 @@ mod countmin; #[path = "serde_tests/cpc.rs"] mod cpc; +#[path = "serde_tests/ddsketch.rs"] +mod ddsketch; + #[path = "serde_tests/frequencies.rs"] mod frequencies; diff --git a/tests-integration/tests/serde_tests/ddsketch.rs b/tests-integration/tests/serde_tests/ddsketch.rs new file mode 100644 index 00000000..3dab040d --- /dev/null +++ b/tests-integration/tests/serde_tests/ddsketch.rs @@ -0,0 +1,187 @@ +// 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. + +use datasketches::ddsketch::CollapseDirection; +use datasketches::ddsketch::DdSketch; +use datasketches::error::ErrorKind; + +#[test] +fn roundtrip_preserves_queries_and_future_operations() { + for direction in [CollapseDirection::Lowest, CollapseDirection::Highest] { + for values in [ + vec![], + vec![-0.0], + vec![f64::from_bits(1)], + vec![1.0, 1.01, 1.02], + vec![-f64::MAX, -1.0, 0.0, 1.0, f64::MAX], + vec![ + -f64::MIN_POSITIVE, + -f64::from_bits(1), + f64::from_bits(1), + f64::MIN_POSITIVE, + ], + ] { + let mut original = DdSketch::with_collapse_direction(8, 0.01, direction).unwrap(); + for value in values { + original.update_weighted(value, 3).unwrap(); + } + let bytes = original.serialize(); + let mut restored = DdSketch::deserialize(&bytes).unwrap(); + assert_eq!(restored.serialize(), bytes); + for q in [0.0, 0.01, 0.5, 0.9, 1.0] { + assert_eq!(original.quantile(q).unwrap(), restored.quantile(q).unwrap()); + assert_eq!( + original.quantile_is_collapsed(q).unwrap(), + restored.quantile_is_collapsed(q).unwrap() + ); + } + for value in [-1e200, -7.0, 0.0, 7.0, 1e200] { + original.update(value).unwrap(); + restored.update(value).unwrap(); + } + let other = restored.clone(); + original.merge(&other).unwrap(); + restored.merge(&other).unwrap(); + assert_eq!(original.serialize(), restored.serialize()); + + let mut output = vec![0; bytes.len() + 100]; + let capacity = output.capacity(); + DdSketch::deserialize(&bytes) + .unwrap() + .serialize_into(&mut output); + assert_eq!(output, bytes); + assert_eq!(output.capacity(), capacity); + } + } +} + +#[test] +fn version_one_image_is_readable_without_the_writer() { + // Independently constructed image: gamma = 2, positive bins (1, 2] and + // (2, 4] with weights 2 and 3. This pins the format as well as its meaning. + let mut bytes = Vec::from(*b"DDS1"); + bytes.extend_from_slice(&8u32.to_le_bytes()); + bytes.extend_from_slice(&(1.0f64 / 3.0).to_le_bytes()); + bytes.extend_from_slice(&[0, 0, 0, 0]); + bytes.extend_from_slice(&5u64.to_le_bytes()); + bytes.extend_from_slice(&0u64.to_le_bytes()); + bytes.extend_from_slice(&1.5f64.to_le_bytes()); + bytes.extend_from_slice(&3.0f64.to_le_bytes()); + bytes.extend_from_slice(&0i64.to_le_bytes()); + bytes.extend_from_slice(&0u32.to_le_bytes()); + bytes.extend_from_slice(&1i64.to_le_bytes()); + bytes.extend_from_slice(&2u32.to_le_bytes()); + bytes.extend_from_slice(&2u64.to_le_bytes()); + bytes.extend_from_slice(&3u64.to_le_bytes()); + let sketch = DdSketch::deserialize(&bytes).unwrap(); + assert_eq!(sketch.total_weight(), 5); + assert!((sketch.quantile(0.5).unwrap().unwrap() - 8.0 / 3.0).abs() < 1e-14); + assert_eq!(sketch.quantile(0.0).unwrap(), Some(1.5)); + assert_eq!(sketch.quantile(1.0).unwrap(), Some(3.0)); + assert_eq!(sketch.serialize(), bytes); +} + +#[test] +fn logarithmic_boundary_rounding_does_not_prevent_reading() { + let mut sketch = DdSketch::new(8, 1.0 / 3.0).unwrap(); + sketch.update(2.0).unwrap(); + // At a power of gamma, platform math can place the value on either side + // of the bucket boundary. Both images must remain usable after decoding. + for index in [1i64, 2] { + let mut bytes = sketch.serialize(); + bytes[64..72].copy_from_slice(&index.to_le_bytes()); + let mut restored = DdSketch::deserialize(&bytes).unwrap(); + assert_eq!(restored.quantile(0.5).unwrap(), Some(2.0)); + restored.update(8.0).unwrap(); + DdSketch::deserialize(&restored.serialize()).unwrap(); + } + let mut bytes = sketch.serialize(); + bytes[64..72].copy_from_slice(&3i64.to_le_bytes()); + assert!(DdSketch::deserialize(&bytes).is_err()); +} + +#[test] +fn malformed_headers_counts_and_extrema_are_rejected() { + let mut sketch = DdSketch::new(8, 0.1).unwrap(); + sketch.update(-10.0).unwrap(); + sketch.update(10.0).unwrap(); + let original = sketch.serialize(); + for end in 0..original.len() { + assert_eq!( + DdSketch::deserialize(&original[..end]).unwrap_err().kind(), + ErrorKind::InvalidData + ); + } + let mut extra = original.clone(); + extra.push(0); + assert!(DdSketch::deserialize(&extra).is_err()); + let corruptions: Vec<(usize, Vec)> = vec![ + (0, b"DDS2".to_vec()), + (4, 0u32.to_le_bytes().to_vec()), + (8, f64::NAN.to_le_bytes().to_vec()), + (8, 0.0f64.to_le_bytes().to_vec()), + (16, vec![2]), + (17, vec![2]), + (18, vec![1]), + (19, vec![1]), + (20, 0u64.to_le_bytes().to_vec()), + (28, u64::MAX.to_le_bytes().to_vec()), + (36, f64::NAN.to_le_bytes().to_vec()), + (36, 11.0f64.to_le_bytes().to_vec()), + (44, f64::INFINITY.to_le_bytes().to_vec()), + (44, (-0.0f64).to_le_bytes().to_vec()), + (52, i64::MIN.to_le_bytes().to_vec()), + (60, u32::MAX.to_le_bytes().to_vec()), + (64, i64::MAX.to_le_bytes().to_vec()), + (72, u32::MAX.to_le_bytes().to_vec()), + (76, 0u64.to_le_bytes().to_vec()), + (76, u64::MAX.to_le_bytes().to_vec()), + ]; + for (offset, replacement) in corruptions { + let mut bytes = original.clone(); + bytes[offset..offset + replacement.len()].copy_from_slice(&replacement); + assert_eq!( + DdSketch::deserialize(&bytes).unwrap_err().kind(), + ErrorKind::InvalidData, + "offset {offset}" + ); + } +} + +#[test] +fn mutated_images_are_rejected_or_remain_safe_to_use() { + let mut sketch = DdSketch::new(4, 0.1).unwrap(); + for value in [-1e300, -1.0, 0.0, 1.0, 1e300] { + sketch.update(value).unwrap(); + } + let original = sketch.serialize(); + for offset in 0..original.len() { + for bit in 0..8 { + let mut bytes = original.clone(); + bytes[offset] ^= 1 << bit; + if let Ok(mut restored) = DdSketch::deserialize(&bytes) { + assert_eq!(restored.serialize(), bytes); + for q in [0.0, 0.25, 0.5, 0.75, 1.0] { + assert!(restored.quantile(q).unwrap().unwrap().is_finite()); + } + restored.update(0.0).unwrap(); + let _ = restored.update(-f64::MAX); + DdSketch::deserialize(&restored.serialize()).unwrap(); + } + } + } +}