tisonkun opened a new pull request, #313:
URL: https://github.com/apache/datasketches-rust/pull/313
Public operations, storage management, query helpers, and serialization are
interleaved in `sketch.rs`, making the implementation difficult to follow.
Group the public APIs first, keep complete algorithms together, and name query
helpers around interpolation intervals and cumulative weights.
Digest-independent interpolation becomes a module-level function; construction
remains associated with the type.
Keep codec implementation and format constants together at the end of the
same module to preserve its compilation boundary. Public APIs, query
arithmetic, validation, and serialization formats are unchanged.
## Performance methodology
Compared baseline `8bb74d7fe77d769ff9c20d2a3b8e8296928239ea` with
`354d742cfffcdfc886d805a08f794d3b3661e655` on an Apple M4 Max, macOS 26.6.2,
Rust 1.99.1 (`58cf8f9c3`, LLVM 23.1.1), using default release settings and the
`tdigest` feature.
- Use the same standalone harness and existing
`benchmarks/tdigest/support.rs` fixtures for both builds. Timing uses `Instant`
and `std::hint::black_box`, with the system allocator and allocation profiling
disabled. Builds have no custom Rust flags. This is a local desktop run,
without CPU affinity or an isolated host.
- Calibrate repetitions **only on the baseline**, targeting a whole batch of
approximately **300 ms**. Freeze that count for the baseline, candidate, and
control; do not independently scale each implementation or retune counts after
warmup. This differs from setting Divan's overall minimum runtime to 300 ms.
- Run two warmup rounds at one fifth of the fixed count, then seven measured
rounds at the full count. Each case runs in a fresh process. Reverse
baseline/control/candidate order on alternating rounds.
- Generate inputs and prepare prebuilt digests before timing. The general
query fixture uses `k=200` and 100,000 deterministic values. CDF/PMF split
points and sorted quantile ranks are evenly spaced in `(0, 1)`. Output
allocation and destruction are included.
- The merge-only case clones input pools of up to 256 compressed digests
outside timing; merging and destroying consumed states are inside timing. Its
control includes cloning inside timing. Each source digest represents 100,000
values.
- Serialized aggregation starts with a prepared eight-value image: decode,
owned collection, serialize, decode, and query the 0.9 quantile. Building that
fixture is excluded. The build and build-plus-CDF rows separately include
ingestion of 100,000 values and subsequent destruction, with pending
compression included in the CDF lifecycle.
- Include an **A/A control copied byte-for-byte from the baseline
executable**, measured at the same fixed count. Source and executable hashes
were checked against the measured revision.
## Results
Times below are **whole-batch medians**, not single-call latencies. `Paired
change = median_i(100 * (PR_i / main_i - 1))`; A/A applies the same formula to
the identical baseline copy. Positive means slower. A median of paired ratios
need not equal the ratio of the displayed median times. Counts were fixed after
baseline calibration, so measured baseline totals can drift from the 300 ms
target.
| Workload per repetition | Repetitions per sample | Main (ms) | PR (ms) |
Paired change | A/A change |
| --- | ---: | ---: | ---: | ---: | ---: |
| Frozen Rank | 12,241,214 | 303.1 | 304.1 | +0.68% | +1.05% |
| Mutable Rank, compressed state | 10,859,058 | 282.4 | 278.3 | -0.53% |
-1.16% |
| 256 mixed Rank queries | 43,990 | 302.7 | 311.3 | +1.95% | -0.99% |
| Rank on six values | 50,228,751 | 292.9 | 285.9 | -2.43% | +1.16% |
| Rank on repeated means | 13,595,031 | 300.7 | 300.4 | +0.00% | -0.77% |
| CDF, 1 split point | 6,172,792 | 301.0 | 301.1 | -0.47% | -0.53% |
| CDF, 6 split points | 2,233,848 | 291.0 | 292.7 | +0.01% | +0.84% |
| CDF, 100 split points | 691,840 | 311.5 | 294.6 | -3.95% | -3.28% |
| CDF, 512 split points | 148,389 | 299.4 | 292.1 | -1.81% | -0.38% |
| PMF, 1 split point | 6,142,334 | 299.6 | 301.0 | +0.19% | -0.39% |
| PMF, 6 split points | 2,198,425 | 296.7 | 295.1 | -0.33% | +0.67% |
| PMF, 100 split points | 656,161 | 296.1 | 287.5 | -2.94% | +1.35% |
| PMF, 512 split points | 143,748 | 296.8 | 291.7 | -2.16% | -1.93% |
| 100 sorted quantiles | 974,627 | 294.3 | 299.0 | +1.62% | -0.68% |
| Merge, input cloning excluded | 149,126 | 303.1 | 306.9 | +1.28% | -0.09% |
| Clone + merge | 145,630 | 300.3 | 301.6 | +0.71% | +1.78% |
| Decode 512 states of 64 values | 3,666 | 300.3 | 301.0 | +0.24% | -0.33% |
| Decode one 64-value state | 1,693,951 | 246.5 | 250.6 | +0.88% | -0.02% |
| Serialize a prebuilt digest | 1,209,004 | 304.4 | 307.0 | -0.09% | -0.57% |
| Serialized eight-value aggregation | 2,160,388 | 294.7 | 295.0 | -0.26% |
+0.37% |
| Build from 100,000 values | 111 | 299.7 | 296.1 | -1.40% | -1.56% |
| Build 100,000 values + CDF100 | 109 | 296.5 | 296.8 | -0.02% | -0.69% |
The largest paired median increase in this run is **1.95% for mixed Rank**,
with all seven paired increases between **0.62% and 2.88%**. The same
workload's A/A paired range is **−2.31% to +4.32%**; across workloads, A/A
paired medians range from **−3.28% to +1.78%**. The control demonstrates
measurement variability, but does not prove that every candidate difference is
noise or exclude a small performance effect. No speedup or universal
zero-regression guarantee is claimed.
As a separate code-generation check, the CDF kernel, pairwise merge,
compression loop, centroid merge, and decoder have matching normalized
instruction sequences after resolving external call targets and excluding
relocation addresses and alignment padding. The decoder contains 1,666
instructions in each build. This checks the named kernels, not every inlined
caller or workload.
## Validation
- `cargo x check`
- `cargo x test`: 778 passing tests, including doctests and serialization
compatibility tests
- `cargo x lint`
<details>
<summary>Standalone fixed-work benchmark harness and runner</summary>
These temporary benchmark sources are provided here for reproducing the
measurement protocol; they are not part of the library diff. Create an
independent Cargo package named `tdigest-probe` (version `0.0.0`, edition
`2024`) with a path dependency on the checkout's `datasketches` package,
enabling only `tdigest`. The measured manifest also declares optional `divan =
"=0.1.21"` and `profile = ["dep:divan"]`; leave that feature disabled. Copy the
checkout's `benchmarks/tdigest/support.rs` unchanged to `src/support.rs`, and
use the following as `src/main.rs`.
Build the exact same harness against each revision with `cargo +1.99.1 build
--release`. Keep the executables immutable as `bin/baseline` and
`bin/candidate`, then copy `bin/baseline` to `bin/control`. Run the Python
driver beside `bin/` as `python3 run.py baseline,control,candidate results`.
The QoS call in this driver matches the macOS measurement host; input
preparation and process startup are outside the Rust timer.
```rust
use std::hint::black_box;
use std::time::{Duration, Instant};
use datasketches::tdigest::{TDigest, TDigestMut};
#[allow(dead_code)]
mod support;
#[cfg(feature = "profile")]
#[global_allocator]
static ALLOC: divan::AllocProfiler = divan::AllocProfiler::system();
fn measure<T>(count: usize, mut op: impl FnMut() -> T) -> Duration {
let start = Instant::now();
for _ in 0..count { black_box(op()); }
start.elapsed()
}
fn digest(xs: &[f64]) -> TDigest { support::build_digest(xs) }
fn main() {
let args: Vec<_> = std::env::args().collect();
let case = &args[1];
let count: usize = args[2].parse().unwrap();
let elapsed = match case.as_str() {
"rank" => { let d=support::prepared_digest(); measure(count, ||
black_box(&d).rank(black_box(0.53125)).unwrap()) }
"rank_mut" => { let mut d=support::prepared_digest().unfreeze();
measure(count, || black_box(&mut d).rank(black_box(0.53125)).unwrap()) }
"rank_mixed" => { let d=support::prepared_digest(); let
xs=support::values(256); measure(count, || { for &x in black_box(&xs)
{black_box(black_box(&d).rank(black_box(x)).unwrap());} }) }
"rank_small" => { let d=digest(&[0.,1.,2.,3.,4.,5.]); measure(count,
|| black_box(&d).rank(black_box(2.5)).unwrap()) }
"rank_repeated" => { let xs=(0..100_000).map(|i|(i%8) as
f64).collect::<Vec<_>>(); let d=digest(&xs); measure(count, ||
black_box(&d).rank(black_box(3.5)).unwrap()) }
"quantiles_100" => { let d=support::prepared_digest(); let
qs=(1..=100).map(|i|i as f64/101.).collect::<Vec<_>>(); measure(count, ||
black_box(&d).quantiles(black_box(&qs)).unwrap()) }
"merge" | "merge_clone" => {
let xs=support::values(200_000);
let mut left=support::build_mut_digest(&xs[..100_000]); let mut
right=support::build_mut_digest(&xs[100_000..]);
black_box(left.rank(0.5).unwrap());
black_box(right.rank(0.5).unwrap());
if case=="merge_clone" { measure(count, || {let mut
d=black_box(&left).clone(); d.merge(black_box(&right)); d}) }
else {
let mut elapsed=Duration::ZERO; let mut done=0;
while done<count {
let batch=(count-done).min(256);
let
inputs=(0..batch).map(|_|left.clone()).collect::<Vec<_>>();
let start=Instant::now();
for mut d in inputs {d.merge(black_box(&right));
black_box(d);}
elapsed+=start.elapsed(); done+=batch;
}
elapsed
}
}
"decode_groups" => {
let bytes=support::serialized_partial_digests_with(200,512,64);
measure(count, ||
black_box(&bytes).iter().map(|b|TDigestMut::deserialize(black_box(b)).unwrap()).collect::<Vec<_>>())
}
"decode_one" => {
let
bytes=support::serialized_partial_digests_with(200,1,64).pop().unwrap();
measure(count, ||
TDigestMut::deserialize(black_box(&bytes)).unwrap())
}
"serialize" => { let d=support::prepared_digest(); measure(count, ||
black_box(&d).serialize()) }
"aggregate_8" => {
let xs=support::values(8); let
bytes=support::build_mut_digest(&xs).serialize();
measure(count, || {
let mut
merged=std::iter::once(black_box(&bytes)).map(|b|TDigestMut::deserialize(b)).collect::<Result<TDigestMut,_>>().unwrap();
let result=merged.serialize();
TDigestMut::deserialize(black_box(&result)).unwrap().quantile(black_box(0.9)).unwrap()
})
}
"update" => { let xs=support::values(100_000); measure(count, ||
support::build_mut_digest(black_box(&xs))) }
"lifecycle" => { let xs=support::values(100_000); let
qs=(1..=100).map(|i|i as f64/101.).collect::<Vec<_>>(); measure(count, ||
support::build_mut_digest(black_box(&xs)).cdf(black_box(&qs)).unwrap()) }
other if other.starts_with("cdf_") || other.starts_with("pmf_") => {
let d=support::prepared_digest(); let
len:usize=other[4..].parse().unwrap();
let qs=(1..=len).map(|i|i as f64/(len+1) as
f64).collect::<Vec<_>>();
if other.starts_with("cdf_") {measure(count, ||
black_box(&d).cdf(black_box(&qs)).unwrap())}
else {measure(count, ||
black_box(&d).pmf(black_box(&qs)).unwrap())}
}
_ => panic!("unknown case"),
};
println!("{}", elapsed.as_nanos());
}
```
```python
from pathlib import Path
import subprocess,json,statistics,random,ctypes,sys,hashlib
root=Path(__file__).parent
assert ctypes.CDLL(None).pthread_set_qos_class_self_np(0x19,0)==0
variants=sys.argv[1].split(',') if len(sys.argv)>1 else
['baseline-system','initial-system']
tag=sys.argv[2] if len(sys.argv)>2 else 'system'
cases=['rank','rank_mut','rank_mixed','rank_small','rank_repeated','cdf_1','cdf_6','cdf_100','cdf_512','pmf_1','pmf_6','pmf_100','pmf_512','quantiles_100','merge','merge_clone','decode_groups','decode_one','serialize','aggregate_8','update','lifecycle']
if len(sys.argv)>3:cases=sys.argv[3].split(',')
def measure(name,case,n):return
int(subprocess.check_output([str(root/'bin'/name),case,str(n)],text=True))
counts={}
for c in cases:
n=1
while True:
ns=measure(variants[0],c,n)
if ns>=30_000_000:break
n=max(n+1,min(n*10,round(n*50_000_000/max(ns,1))))
n=max(1,round(n*300_000_000/ns))
ns=measure(variants[0],c,n)
n=max(1,round(n*300_000_000/ns))
counts[c]=n
print('calibrated',c,n,flush=True)
(root/(tag+'-counts.json')).write_text(json.dumps(counts,indent=2))
rows=[]
for i in range(-2,7):
row={}
for c in cases:
order=variants.copy()
if i%2:order.reverse()
for name in order:
ns=measure(name,c,counts[c] if i>=0 else max(1,counts[c]//5))
row.setdefault(name,{})[c]=ns
if i>=0:
rows.append(row);(root/(tag+'-rounds.json')).write_text(json.dumps(rows,indent=2))
print('round',i,'complete',flush=True)
results=[]
for c in cases:
base=variants[0]
med={n:statistics.median(r[n][c] for r in rows)/1e6 for n in variants}
changes={n:statistics.median((r[n][c]/r[base][c]-1)*100 for r in rows) for
n in variants}
ranges={n:[min((r[n][c]/r[base][c]-1)*100 for r in
rows),max((r[n][c]/r[base][c]-1)*100 for r in rows)] for n in variants}
results.append({'case':c,'count':counts[c],'median_ms':med,'changes_percent':changes,'paired_range_percent':ranges})
print(c,counts[c],{n:round(v,3) for n,v in med.items()},{n:round(v,2) for
n,v in changes.items()},flush=True)
(root/(tag+'-results.json')).write_text(json.dumps(results,indent=2))
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
