Skip to content

refactor(tdigest): organize public APIs and internal helpers - #313

Merged
tisonkun merged 2 commits into
apache:mainfrom
tisonkun:codex/tdigest-organization
Oct 27, 2026
Merged

tisonkun merged 2 commits into
apache:mainfrom
tisonkun:codex/tdigest-organization

Conversation

@tisonkun

Copy link
Copy Markdown
Member

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
Standalone fixed-work benchmark harness and runner

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.

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

@tisonkun
tisonkun enabled auto-merge (squash) October 27, 2026 15:17
@tisonkun
tisonkun merged commit 0cf3cf6 into apache:main Oct 27, 2026
10 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant