How DataFusion Approximates Quantiles with a Mergeable t-Digest (and Distinct Counts with HyperLogLog)

TL;DR: approx_percentile_cont keeps a bounded set of centroids — a mean and a weight each — instead of the raw values. The state is just those centroids plus sum/count/min/max, so a partial aggregate can hand it to a final aggregate over the wire and the final aggregate folds the centroid lists together. The catch: merging re-compresses, so the result depends slightly on how the data was partitioned. HyperLogLog, used by approx_distinct, is exactly mergeable (register-wise max), and DataFusion even gives it a sparse representation so a group with three distinct values does not cost 16 KiB.

The problem: an exact median is not mergeable

SELECT percentile_cont(0.5) WITHIN GROUP (ORDER BY x) FROM t needs every value of x, sorted. That is fine until you want to parallelize it: the partial phase on partition 0 has 10 million values, the partial phase on partition 1 has 10 million more, and the final phase has to combine them into one answer. If the aggregate function is “sort everything”, the state is “everything”, and the shuffle moves all the data.

An aggregate is only parallelizable if its state is small and mergeable. sum is the archetype: state is one i64, merge is addition. count likewise. percentile_cont is not: its state is the whole multiset.

Sketch algorithms fix this by keeping an approximate state that is small and mergeable. DataFusion ships two of them: a t-digest for quantiles (datafusion/functions-aggregate-common/src/tdigest.rs) and a HyperLogLog for distinct counts (datafusion/functions-aggregate/src/hyperloglog.rs).

The t-digest data structure

The t-digest (Dunning’s 2019 paper, arXiv:1902.04023) is a “clustering” sketch: instead of storing values, it stores centroids, each of which is a weighted average:

// datafusion/functions-aggregate-common/src/tdigest.rs:54
pub struct Centroid {
    mean: f64,
    weight: f64,
}

and the digest itself is:

// datafusion/functions-aggregate-common/src/tdigest.rs:98
pub struct TDigest {
    centroids: Vec<Centroid>,
    max_size: usize,     // default 100 (DEFAULT_MAX_SIZE, tdigest.rs:39)
    sum: f64,
    count: f64,
    max: f64,
    min: f64,
}

The magic is which values get merged together. A naive “keep the first 100 distinct values” is useless for quantiles. A t-digest merges neighbouring values, but only when the resulting cluster is small enough, and the “small enough” threshold depends on where in the distribution the cluster sits. That is the scale function:

// datafusion/functions-aggregate-common/src/tdigest.rs:188
fn k_to_q(k: u64, d: usize) -> f64 {
    let k_div_d = k as f64 / d as f64;
    if k_div_d >= 0.5 {
        let base = 1.0 - k_div_d;
        1.0 - 2.0 * base * base
    } else {
        2.0 * k_div_d * k_div_d
    }
}

k is the index of the centroid, d is max_size. The quantile q a centroid is allowed to “cover” is k_to_q(k, d): near the tails the curve is flat, so a single centroid there may represent a tiny slice of the distribution, and near the median it is steep, so centroids there are allowed to be fat. Practically: the tails are precise and the middle is fuzzy, which is exactly what you want from a quantile sketch — nobody cares whether p50 is off by 0.1%, everybody cares whether p99.9 is right.

The compression loop is a single pass over the sorted values, accumulating weights until the limit for the current centroid index is exceeded:

// datafusion/functions-aggregate-common/src/tdigest.rs:285 (inside merge_sorted_f64)
if weight_so_far <= q_limit_times_count {
    sums_to_merge += next_sum;
    weights_to_merge += next.weight();
} else {
    result.sum += curr.add(sums_to_merge, weights_to_merge);
    compressed.push(curr.clone());
    q_limit_times_count = Self::k_to_q(k_limit, self.max_size) * result.count();
    k_limit += 1;
    curr = next;
}

Note q_limit_times_count is a fraction of the total count, so the digest re-balances itself as more data arrives: the same digest that held 100 values in 20 centroids will hold 100 million values in 100 centroids, still with the tail clusters small.

Making the digest mergeable

Merging two digests means merging their centroid lists. The lists are sorted by mean, so the merge is a sorted merge followed by a re-compression pass with the same scale function. For more than two digests, DataFusion does a bottom-up merge sort over the concatenated centroid lists:

// datafusion/functions-aggregate-common/src/tdigest.rs:357
pub fn merge_digests<'a>(digests: impl IntoIterator<Item = &'a TDigest>) -> TDigest {
    // ... concatenate all centroids ...
    let mut digests_per_block: usize = 1;
    while digests_per_block < starts.len() {
        for i in (0..starts.len()).step_by(digests_per_block * 2) {
            // ... Self::external_merge(&mut centroids, first, middle, last, &mut merge_buffer)
        }
        digests_per_block *= 2;
    }
    // ... then re-compress under the k_to_q limits ...
}

external_merge (tdigest.rs:310) is a textbook two-pointer merge into a scratch buffer, then copy back. Bottom-up merging (digests_per_block *= 2) keeps the total work O(n log n) in the number of digests, and it is done on flat centroid arrays, not on nested structures — no per-digest allocation in the inner loop.

Serializing the state so it can cross a shuffle

The state has to be Arrow data. TDigest::to_scalar_state (tdigest.rs:576) packs it into six scalars:

    ┌────────┬────────┬────────┬───────┬────────┬────────┐
    │max_size│  sum   │ count  │  max  │  min   │centroid│
    └────────┴────────┴────────┴───────┴────────┴────────┘
                                                     │
                               ┌─────────────────────┘
                               ▼
                          ┌ List ───┐
                          │ mean    │
                          │ weight  │  <- Centroid 1
                          │ mean    │
                          │ weight  │  <- Centroid 2
                          └─────────┘

and from_scalar_state (tdigest.rs:605) reverses it. The centroids are packed as a flat List<Float64> of [mean, weight, mean, weight, ...] — a deliberately boring layout, because it round-trips through Arrow without custom serialization.

DataFusion’s aggregate wires this up with seven state columns (datafusion/functions-aggregate/src/approx_percentile_cont.rs:250): the six digest fields plus the percentile itself. The percentile has to travel with the state because the final phase does not have the original arguments — it only has the state. STATE_PERCENTILE_IDX = 6 (approx_percentile_cont.rs:52) marks that column, and merge_batch reads it back out:

// datafusion/functions-aggregate/src/approx_percentile_cont.rs:461
fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
    if let Some(percentile_array) = states.get(STATE_PERCENTILE_IDX) {
        self.percentile.resolve(percentile_array)?;
    }
    let tdigest_states = &states[..STATE_PERCENTILE_IDX];
    let states = (0..tdigest_states[0].len())
        .map(|index| {
            tdigest_states.iter()
                .map(|array| ScalarValue::try_from_array(array, index))
                .collect::<Result<Vec<_>>>()
                .map(|state| TDigest::from_scalar_state(&state))
        })
        .collect::<Result<Vec<_>>>()?;
    self.merge_digests(&states);
    Ok(())
}

Note that merge_batch reconstructs a TDigest per row of the incoming state, then merges them all. That is the price of a “state is Arrow arrays” design: the per-group state is unpacked, then folded.

The honest caveat: t-digest is approximately mergeable

Here is the thing to internalize. A t-digest is not a CRDT. Merging digests D1 and D2 in the order (D1, D2) does not produce the same centroids as merging in the order (D2, D1), because the re-compression step merges neighbouring centroids differently depending on where the boundary falls. The estimate is stable to within the sketch’s error bar, but it is not bit-exact.

Concretely: SELECT approx_percentile_cont(0.5) FROM t may return a slightly different number depending on how many partitions the table had. The discrepancy shrinks as max_size grows — more centroids means less merging, means less re-compression ambiguity. The default is 100 centroids (DEFAULT_MAX_SIZE, tdigest.rs:39), and the user can raise it as the third argument: approx_percentile_cont(0.9, 500).

This is a design trade, not a bug: exactness would require shipping all the data, which is exactly what the sketch exists to avoid.

The DataFusion-specific wart: sorting every batch

update_batch is:

// datafusion/functions-aggregate/src/approx_percentile_cont.rs:427
fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
    if values.len() > 1 {
        self.resolve_percentile(&values[1])?;
    }
    let mut values = Arc::clone(&values[0]);
    if values.null_count() > 0 {
        values = filter(&values, &is_not_null(&values)?)?;
    }
    let sorted_values = &arrow::compute::sort(&values, None)?;
    let sorted_values = ApproxPercentileAccumulator::convert_to_float(sorted_values)?;
    self.digest = self.digest.merge_sorted_f64(&sorted_values);
    Ok(())
}

Read the cost carefully: every batch is fully sorted (arrow::compute::sort), then copied into a Vec<f64>, and then merge_sorted_f64 builds a brand new TDigest (new Vec<Centroid>, new compression pass) rather than incrementally absorbing the values into the existing centroids. So the cost per batch is O(n log n) for the sort, O(n) for the copy, plus a full re-compression of the (up to 100) existing centroids.

A textbook t-digest implementation keeps the digest sorted by centroid mean and inserts new values by binary search + local re-compression, so a batch of n values costs O(n log(max_size)), not O(n log n). DataFusion’s version is simpler to reason about (and reuses Arrow’s vectorized sort), but it is noticeably more expensive than it needs to be for wide batches. If you are benchmarking approx_percentile_cont on a wide table, this is the first place to look.

Bonus: HyperLogLog, and the sparse trick

approx_distinct uses a completely different sketch — one that is exactly mergeable. The state is 2^14 registers of one byte each:

// datafusion/functions-aggregate/src/hyperloglog.rs:43
const HLL_P: usize = 14_usize;
const HLL_Q: usize = 64_usize - HLL_P;
pub(crate) const NUM_REGISTERS: usize = 1_usize << HLL_P;   // 16384

Each distinct value is hashed; the low 14 bits pick a register, and the remaining bits’ leading-zero count updates it:

// datafusion/functions-aggregate/src/hyperloglog.rs:121
pub(crate) fn add_hashed(&mut self, hash: u64) {
    let index = (hash & HLL_P_MASK) as usize;
    let p = ((hash >> HLL_P) | (1_u64 << HLL_Q)).trailing_zeros() + 1;
    self.registers[index] = self.registers[index].max(p as u8);
}

Merging is register-wise max:

// datafusion/functions-aggregate/src/hyperloglog.rs:140
pub fn merge(&mut self, other: &HyperLogLog<T>) {
    for i in 0..self.registers.len() {
        self.registers[i] = self.registers[i].max(other.registers[i]);
    }
}

max is commutative, associative and idempotent — a textbook CRDT. The estimate is therefore exactly partition-independent, with a standard error of 1.04/sqrt(16384) = 0.8125% (hyperloglog.rs:31-38).

The clever part is what DataFusion added on top. A dense sketch is 16 KiB per group, which is absurd for a GROUP BY where most groups have three distinct values. So approx_distinct keeps an adaptive per-group state:

// datafusion/functions-aggregate/src/approx_distinct.rs:304
const SPARSE_LIMIT: usize = 256;

enum GroupHll {
    Sparse(Vec<u64>),      // distinct hashes seen so far
    Dense(Box<HyperLogLog<u8>>),
}

A group starts sparse, accumulating raw hashes; when it exceeds 256 distinct values it is promoted to a dense sketch (compact_or_promote, approx_distinct.rs:360). Because folding the same hashes into a dense sketch produces exactly the same registers as adding the values one by one, the estimate is identical either way. On the read side, a sparse group is estimated without materializing 16 KiB:

// datafusion/functions-aggregate/src/hyperloglog.rs:177
pub(crate) fn count_from_hashes(hashes: &[u64]) -> usize {
    // sort (index, rho) pairs, keep the max rho per index,
    // build the register histogram, then apply the HLL estimator
}

The comment says it plainly: “This is used to cheaply estimate the many small groups produced by a high-cardinality GROUP BY, where allocating and scanning a 16 KiB sketch per group would dominate the runtime.” And because the sparse state is a list of hashes, the serialized partial state is also small — the final phase receives a few hundred bytes instead of 16 KiB per group (GroupHll::serialize, approx_distinct.rs:438).

Trade-offs, summarized

• t-digest: accurate at the tails, fuzzy in the middle; state is mergeable but the merge is not associative, so results are partition-sensitive; DataFusion’s implementation sorts every batch, which dominates its cost.
• HyperLogLog: exactly mergeable, bounded 0.81% error, but the error is relative — useless for small cardinalities (a set of 3 distinct values estimates as 3 ± 0.8, which rounds to 3, fine, but a set of 1 estimates as 1 with a large relative error), and it cannot tell you which values.
• Both trade exactness for a state that fits in a shuffle. The general lesson is the same: an aggregate becomes parallelizable exactly when its state becomes small, and its merge becomes an associative operation over that state. t-digest is associative approximately; HyperLogLog is associative exactly. DataFusion ships both, and picks per aggregate.