feat: add DDSketch

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.

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.
 

benchmarks/Cargo.toml

@@ -27,6 +27,7 @@ datasketches = { workspace = true, features = [
   "bloom",
   "countmin",
   "cpc",
+  "ddsketch",
   "frequencies",
   "hll",
   "kll",

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
+        });
+}

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;

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()
+    });
+}

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());
+}

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<f64> {
+    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
+}

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);
+            }
+        });
+}

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;

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 }

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(())
+}