Opensource internals

Polars Estimates Cardinality With HyperLogLog

17 min read Jay Hubbert

How Polars estimates distinct counts: a 16 KiB HyperLogLog in polars-compute, a 256-byte sketch in polars-utils, and the count-sketch used to find heavy keys.

TL;DR: Polars has two cardinality sketches. The one in polars-compute has 16 384 one-byte registers, uses the low 14 bits of the hash as the register index and the number of leading zeros of the remaining bits as the register’s value, and estimates the cardinality with Ertl’s tau/sigma correction instead of the bias-correction table of HLL++. It is always dense — there is no sparse representation. The sketch the streaming engine actually uses is much smaller: 256 registers, one byte each, about 6.5% error. Both are used to make decisions: dictionary-encode a Parquet column, pick the build side of a hash join, publish a bloom filter, or size the hot table of a group-by.

Counting distinct values exactly means keeping a set of them, and the set is as big as the answer. A dataframe engine that wants to know “is this column worth dictionary-encoding?” or “which side of this join should I build the hash table on?” usually does not need the exact answer — it needs an answer that is right most of the time and costs a constant amount of memory no matter how many rows go past. That is what a cardinality sketch is, and Polars keeps several of them.

The register trick

The idea behind HyperLogLog is old and simple. Hash every value to 64 bits. Use p bits of the hash to pick one of m = 2^p registers. Use the position of the first 1 bit in the remaining bits as the value stored in that register, keeping the maximum ever seen. A register’s value is then “how far into the hash space did the most extreme value that landed here go”, and the distribution of those values across registers is what encodes the answer.

Polars’ implementation is in crates/polars-compute/src/hyperloglogplus.rs, and the constants that fix its accuracy and its memory are four lines:

rust 
/// The greater is P, the smaller the error.
const HLL_P: usize = 14_usize;
/// The number of bits of the hash value used determining the number of leading zeros
const HLL_Q: usize = 64_usize - HLL_P;
const NUM_REGISTERS: usize = 1_usize << HLL_P;
/// Mask to obtain index into the registers
const HLL_P_MASK: u64 = (NUM_REGISTERS as u64) - 1;

crates/polars-compute/src/hyperloglogplus.rs:31

And the whole update is three lines:

rust 
    /// Adds an element to the HyperLogLog.
    pub fn add(&mut self, obj: &T) {
        let hash = self.hash_value(obj);
        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);
    }

crates/polars-compute/src/hyperloglogplus.rs:88

Two details make this work. First, hash & HLL_P_MASK takes the low 14 bits as the index — the register choice is independent of the rank bits, which is what makes the estimator unbiased. Second, the | (1_u64 << HLL_Q) sets a sentinel bit at position 50, so trailing_zeros() is always defined (a hash of all zeros would otherwise have 64 trailing zeros and overflow the u8 register) and the rank is capped at 51, which fits in a byte.

The registers are one byte each, so the whole sketch is 16 KiB:

rust 
#[derive(Clone, Debug)]
pub struct HyperLogLog<T>
where
    T: Hash + ?Sized,
{
    registers: [u8; NUM_REGISTERS],
    phantom: PhantomData<T>,
}

crates/polars-compute/src/hyperloglogplus.rs:38

The hash is seeded with a fixed seed, not a random one:

rust 
/// Fixed seed for the hashing so that values are consistent across runs
///
/// Note that when we later move on to have serialized HLL register binaries
/// shared across cluster, this SEED will have to be consistent across all
/// parties otherwise we might have corruption. So ideally for later this seed
/// shall be part of the serialized form (or stay unchanged across versions).
const SEED: PlFixedStateQuality = PlFixedStateQuality::with_seed(0);

crates/polars-compute/src/hyperloglogplus.rs:57

That is a deliberate trade: a random seed would make two sketches of the same data disagree, which would break the merge operation and any attempt to compare estimates across runs. The comment shows the author thinking about shipping register arrays between machines.

Merging is the operation a sketch is really built for, and it is trivial because the registers are maxima:

rust 
    /// Merge the other [`HyperLogLog`] into this one
    pub fn merge(&mut self, other: &HyperLogLog<T>) {
        assert!(
            self.registers.len() == other.registers.len(),
            "unexpected got unequal register size, expect {}, got {}",
            self.registers.len(),
            other.registers.len()
        );
        for i in 0..self.registers.len() {
            self.registers[i] = self.registers[i].max(other.registers[i]);
        }
    }

crates/polars-compute/src/hyperloglogplus.rs:113

You can sketch a billion rows on four machines and combine the four sketches into one, and the answer is the answer for all of them.

The estimator, and what “HLL++” means here

The original HyperLogLog paper needed two corrections: a small-range correction (when the sketch is mostly empty registers) and a large-range correction (when the linear-count estimator breaks down and the two-largest harmonic mean takes over). HLL++ added a third thing: an empirically measured bias correction, a table of corrections per register value that removes the systematic error of the estimator in the small-to-medium range.

Polars implements none of that machinery. Its estimator is Ertl’s, from “New cardinality estimation algorithms for HyperLogLog sketches” (arXiv:1702.01284), and it folds the range corrections into the estimator itself:

rust 
    /// Guess the number of unique elements seen by the HyperLogLog.
    pub fn count(&self) -> usize {
        let histogram = self.get_histogram();
        let m = NUM_REGISTERS as f64;
        let mut z = m * hll_tau((m - histogram[HLL_Q + 1] as f64) / m);
        for i in histogram[1..=HLL_Q].iter().rev() {
            z += *i as f64;
            z *= 0.5;
        }
        z += m * hll_sigma(histogram[0] as f64 / m);
        (0.5 / 2_f64.ln() * m * m / z).round() as usize
    }

crates/polars-compute/src/hyperloglogplus.rs:121

The histogram counts how many registers hold each value. z is built from the register values in a harmonic-mean-like way; hll_tau is the correction for the case where the sketch has seen a lot of high values, and hll_sigma the correction for the case where most registers are still zero. Both are iterative series that stop when they converge:

rust 
/// Helper function sigma as defined in
/// "New cardinality estimation algorithms for HyperLogLog sketches"
/// Otmar Ertl, arXiv:1702.01284
#[inline]
fn hll_sigma(x: f64) -> f64 {
    if x == 1. {
        f64::INFINITY
    } else {
        let mut y = 1.0;
        let mut z = x;
        let mut x = x;
        loop {
            x *= x;
            let z_prime = z;
            z += x * y;
            y += y;
            if z_prime == z {
                break;
            }
        }
        z
    }
}

crates/polars-compute/src/hyperloglogplus.rs:135

The point of the tau/sigma estimator is that it is unbiased over the whole range, which makes the empirical bias table of HLL++ unnecessary. What the module does not have — and this is where the name is slightly misleading — is the other half of HLL++: the sparse representation. The registers are always a dense [u8; 16384], so every sketch costs 16 KiB, even if it has seen three values. There is no switch from a sparse list of (index, value) pairs to the dense array when the list gets too long. For a column with a handful of distinct values, Polars pays 16 KiB anyway.

The accuracy is the usual HyperLogLog accuracy, and the test suite states it in the only units that matter:

rust 
    fn compare_with_delta(got: usize, expected: usize) {
        let expected = expected as f64;
        let diff = (got as f64) - expected;
        let diff = diff.abs() / expected;
        // times 6 because we want the tests to be stable
        // so we allow a rather large margin of error
        // this is adopted from redis's unit test version as well
        let margin = 1.04 / ((NUM_REGISTERS as f64).sqrt()) * 6.0;

crates/polars-compute/src/hyperloglogplus.rs:221

1.04 / sqrt(16384) is 0.81% — the standard error of the estimator — and the test allows six times that, because a test that fails one run in a million is not a test.

The smaller sketch the streaming engine uses

The 16 KiB sketch is fine for a one-shot approx_n_unique. It is not fine for the streaming engine, where a sketch per partition per join side adds up, and where the answer only has to be good enough to make a decision. So polars-utils carries a second, much smaller implementation:

rust 
/// HyperLogLog in Practice: Algorithmic Engineering of
/// a State of The Art Cardinality Estimation Algorithm
/// Stefan Heule, Marc Nunkesser, Alexander Hall
///
/// We use m = 256 which gives a relative error of ~6.5% of the cardinality
/// estimate. We don't bother with stuffing the counts in 6 bits, byte access is
/// fast.
///
/// The bias correction described in the paper is not implemented, so this is
/// somewhere in between HyperLogLog and HyperLogLog++.
#[derive(Clone)]
pub struct CardinalitySketch {
    buckets: [u8; 256],
}

crates/polars-utils/src/cardinality_sketch.rs:20

The comment is unusually honest: this is a HyperLogLog without the bias correction, so “somewhere in between”. 256 bytes and 6.5% error is a completely different point on the accuracy/memory curve from 16 KiB and 0.8%, and for “should I publish this bloom filter?” the cheaper answer is the right one.

The update is the same register trick with different constants:

rust 
    /// Add a new hash to the sketch.
    #[inline]
    pub fn insert(&mut self, mut h: u64) {
        const ARBITRARY_ODD: u64 = 0x902813a5785dc787;
        // We multiply by this arbitrarily chosen odd number and then take the
        // top bits to ensure the sketch is influenced by all bits of the hash.
        h = h.wrapping_mul(ARBITRARY_ODD);
        let idx = (h >> 56) as usize;
        let p = 1 + (h << 8).leading_zeros() as u8;
        self.buckets[idx] = self.buckets[idx].max(p);
    }

crates/polars-utils/src/cardinality_sketch.rs:40

And the estimator is the classical one with the small-range correction:

rust 
    pub fn estimate(&self) -> usize {
        let m = 256.0;
        let alpha_m = 0.7123 / (1.0 + 1.079 / m);

        let mut sum = 0.0;
        let mut num_zero = 0;
        for x in self.buckets.iter() {
            sum = alg_add_f64(sum, inv_pow2(*x));
            num_zero += (*x == 0) as usize;
        }

        let est = (alpha_m * m * m) / sum;
        let corr_est = if est <= 5.0 / 2.0 * m && num_zero != 0 {
            // Small cardinality estimate, full 64-bit logarithm is overkill.
            m * (m as f32 / num_zero as f32).ln() as f64
        } else {
            est
        };

        if num_zero == self.buckets.len() {
            0
        } else {
            ((corr_est + 0.5) as usize).max(1)
        }
    }

crates/polars-utils/src/cardinality_sketch.rs:65

Two details are worth noticing. The hash is multiplied by an odd constant before the bits are used, which is a cheap way to mix the hash so that all of its bits influence the register choice. And the sum of 2^-bucket is accumulated with alg_add_f64, which on nightly builds uses Rust’s algebraic float addition so that the result does not depend on the order of the additions:

rust 
#[inline(always)]
pub fn alg_add_f64(a: f64, b: f64) -> f64 {
    #[cfg(feature = "nightly")]
    {
        a.algebraic_add(b)
    }
    #[cfg(not(feature = "nightly"))]
    {
        a + b
    }
}

crates/polars-utils/src/algebraic_ops.rs:2

With 256 terms of very different magnitudes, the order of the additions is not irrelevant to the last bits of the sum, and the last bits of the sum are the estimate.

The min of one is also there: an empty sketch estimates 0, and a sketch that has seen anything at all estimates at least 1.

Where the estimates are used

approx_n_unique

The obvious entry point. The chunked implementation is four lines:

rust 
    fn approx_n_unique(&self) -> IdxSize {
        let mut hllp = HyperLogLog::new();
        self.iter().for_each(|item| hllp.add(&item.to_total_ord()));
        hllp.count() as IdxSize
    }

crates/polars-core/src/chunked_array/ops/approx_n_unique.rs:17

to_total_ord is the float wrapper from the sorting story: for f64 it produces a TotalOrdWrap, whose Hash canonicalises -0.0 to 0.0 and all NaNs to one value. Without it, -0.0 and 0.0 would be counted as two distinct values and every NaN would be its own distinct value, which is rarely what the user meant.

Deciding whether to dictionary-encode a Parquet column

polars-compute exposes the sketch as an array-level cardinality estimate, and the Parquet writer uses it to decide whether dictionary encoding is worth it:

rust 
    let estimated_cardinality = polars_compute::cardinality::estimate_cardinality(array);

    if array.len() > 128 && (estimated_cardinality as f64) / (array.len() as f64) > 0.75 {
        return None;
    }

crates/polars-parquet/src/arrow/write/dictionary.rs:256

If more than 75% of the values are distinct, a dictionary is just an indirection with extra steps, so the column is written as plain values. The estimate does not have to be exact for that decision; it has to be right near the boundary, and 0.8% error is plenty.

The estimator itself is a dispatch on the physical type, and it handles strings by hashing the bytes:

rust 
fn binary_offset_array_estimate<O: Offset>(array: &BinaryArray<O>) -> usize {
    let mut hll = HyperLogLog::new();

    if array.has_nulls() {
        for v in array.iter() {
            let v = v.unwrap_or_default();
            hll.add(v);
        }
    } else {
        for v in array.values_iter() {
            hll.add(v);
        }
    }

    hll.count()
}

crates/polars-compute/src/cardinality.rs:127

Note that nulls are hashed as the default value — the empty string, or zero for a primitive — so they contribute at most one more distinct value, and none at all if the default value is in the data anyway. Booleans are handled exactly rather than sketched, because there are only two of them.

Choosing the build side of a hash join

This is where the small sketch earns its keep. The streaming engine samples both sides of an equi-join and estimates how many bytes each side’s hash table would need:

rust 
        let rows = rows as f64;
        Ok(Self {
            key_ratio: (sketch.estimate() as f64 / rows).min(1.0),
            min_hash_key_ratio: min_hash.and_then(|m| m.key_ratio()),
            key_width: key_bytes as f64 / rows,
            row_width: row_bytes as f64 / rows,
        })
    }

    /// Bytes retained when building a table of the distinct keys of `rows` rows.
    pub fn distinct_keys_build_bytes(&self, rows: usize) -> f64 {
        rows as f64 * self.key_ratio * (self.key_width + KEY_SLOT_OVERHEAD)
    }

crates/polars-stream/src/nodes/joins/utils.rs:356

Two ratios are computed. key_ratio is distinct keys over rows, straight from the cardinality sketch. min_hash_key_ratio is the same question asked differently: how many of the rows are distinct keys, counting each key once regardless of how often it appears. That is what the min-hash sketch is for:

rust 
    /// Add a new hash to the sketch.
    #[inline]
    pub fn insert(&mut self, h: u64) {
        const IDX_ODD: u64 = 0x85b0fc582ca15d65; // Randomly chosen.
        const TAG_ODD: u64 = 0x4a0da407caf4e9ad; // Randomly chosen.
        let idx = (h.wrapping_mul(IDX_ODD) >> (64 - NUM_ENTRIES.trailing_zeros())) as usize;
        let tag = (h.wrapping_mul(TAG_ODD) >> 32) as u32;
        let entry = &mut self.entries[idx];
        if tag < entry.tag {
            entry.tag = tag;
            entry.count = 0;
        }
        entry.count += (tag == entry.tag) as u32;
    }

crates/polars-utils/src/min_hash_sketch.rs:38

rust 
    /// Estimates the number of distinct keys as a fraction of the number of
    /// rows, weighting each distinct key equally.
    pub fn key_ratio(&self) -> Option<f64> {
        let mut nonempty = 0u64;
        let mut total = 0u64;
        for entry in self.entries.iter() {
            nonempty += (entry.count > 0) as u64;
            total += entry.count as u64;
        }
        (nonempty > 0).then(|| nonempty as f64 / total as f64)
    }

crates/polars-utils/src/min_hash_sketch.rs:60

Each entry keeps the smallest tag it has seen for that bucket and a count of how many rows hashed to that same tag. A key that appears many times wins its bucket’s tag competition and is counted once, so the ratio of non-empty entries to total count is the fraction of rows that are distinct keys.

Those two numbers feed the build-side decision:

rust 
                        let left_unique =
                            left.min_hash_key_ratio.unwrap_or(0.0) >= UNIQUE_KEY_RATIO;
                        let right_unique =
                            right.min_hash_key_ratio.unwrap_or(0.0) >= UNIQUE_KEY_RATIO;
                        let mut left_score = left.all_rows_build_bytes(params.sample_limit);
                        let mut right_score = right.all_rows_build_bytes(params.sample_limit);
                        if left.key_ratio <= REF_KEY_RATIO && right_unique {
                            left_score *= UNIQUE_BUILD_MARGIN / (left.key_ratio + 1e-6);
                        }
                        if right.key_ratio <= REF_KEY_RATIO && left_unique {
                            right_score *= UNIQUE_BUILD_MARGIN / (right.key_ratio + 1e-6);
                        }

crates/polars-stream/src/nodes/joins/equi_join.rs:521

If one side has unique keys and the other repeats them, the repeating side is assumed to reference the unique side, and it is penalised for being the build side. Getting this wrong costs memory and time, and the sample that produces these numbers is a sample — so the sketch is being used to make a decision that is right most of the time, with a bounded cost when it is wrong.

Publishing a bloom filter

The join runtime filter is a bloom filter built on the build side and probed by the other side. It is only worth publishing if it will reject enough probes, and the decision is made with a cardinality sketch of the build keys:

rust 
        let distinct = sketch.estimate();
        let weak = spec
            .probe_distinct
            .is_some_and(|probe| distinct as f64 > probe as f64 * MAX_BUILD_PROBE_DISTINCT_RATIO);
        let bloom = if weak {
            None
        } else if planned {
            // Every builder holding it was dropped above, so no thread inserts
            // anymore.
            shared
                .and_then(|bloom| {
                    debug_assert_eq!(Arc::strong_count(&bloom), 1);
                    Arc::into_inner(bloom).map(AtomicSplitBlockBloom::into_inner)
                })
                .filter(|bloom| distinct.saturating_mul(BLOOM_MIN_BITS_PER_KEY) <= bloom.num_bits())
        } else {
            KeyFilterSpec::bloom_for(distinct)
        };

crates/polars-stream/src/nodes/joins/runtime_filter.rs:328

If the build side has too many distinct keys relative to the probe side, the filter would not reject much, so it is dropped. The bloom filter itself is a split-block filter in the Parquet layout, 32-byte blocks and eight bits per key.

The F2 sketch: counting heavy keys

polars-utils also has a sketch for the second frequency moment:

rust 
const NUM_BUCKETS: usize = 64;

/// Estimates the second frequency moment F2 = sum_k f_k^2 of a stream of hashed
/// keys, where f_k is the number of times key k was inserted.
///
/// This is a single row of a Count-Sketch (the "fast AMS" sketch): every hash
/// adds +1 or -1 to one of 64 buckets, and the sum of squared buckets is an
/// unbiased estimate of F2 with a relative standard error of ~18%.
#[derive(Clone)]
pub struct F2Sketch {
    buckets: [i64; NUM_BUCKETS],
    num_inserts: u64,
}

crates/polars-utils/src/f2_sketch.rs:1

F2 is not a set-similarity measure; it is a measure of skew. If every key appears once, F2 equals the number of rows. If one key appears a thousand times, F2 is a million. So F2 tells you how heavy the tail of the frequency distribution is, and the streaming group-by uses exactly that to decide how big its hot table should be:

rust 
    fn maybe_grow_hot_tables(&mut self, max_size: usize) {
        let size = self.hot_table_size;
        if size >= max_size {
            return;
        }

        // Cheap early exit before estimating the number of distinct missed keys.
        let misses = self.miss_f2.num_inserts() as f64;
        if misses < HOT_TABLE_GROW_MIN_REPEAT * size as f64 {
            return;
        }

crates/polars-stream/src/nodes/group_by/sink.rs:209

Read the algebra. The model is: the missed rows consist of heavy keys that were each missed repeat times, plus the remaining distinct keys, each missed once. With D distinct keys (the cardinality sketch’s estimate, capped at the number of missed rows) and F2 the second moment over the missed rows only (the F2 sketch only sees the misses):

text 
misses = D + heavy * (repeat - 1)
F2     = D + heavy * (repeat^2 - 1)

Subtracting gives F2 - misses = heavy * repeat * (repeat - 1), and misses - D = heavy * (repeat - 1), so their quotient is repeat; and F2 - 2 * misses + D = heavy * (repeat - 1)^2, so the square of misses - D divided by that is heavy. Both are in the code:

rust 
        let f2 = self.miss_f2.estimate();
        let distinct = (self.key_sketch.estimate() as f64).min(misses);
        let excess = misses - distinct;
        let denom = f2 - 2.0 * misses + distinct;
        if excess <= 0.0 || denom <= 0.0 {
            return;
        }
        let repeat = (f2 - misses) / excess;
        let heavy = excess * excess / denom;

crates/polars-stream/src/nodes/group_by/sink.rs:225

If a few keys are missing over and over, repeat is large and the hot table should grow to keep them; if the misses are spread over many distinct keys, repeat is close to one and growing the table would be wasted memory. That is the whole reason the F2 sketch exists. A cardinality sketch tells you how many distinct keys there are; F2 tells you how skewed they are, and skew is what decides whether a cache is worth growing.

The bottom line

The sketches in Polars are not one algorithm but a family, chosen per decision:

SketchSizeErrorUsed for
HyperLogLog (polars-compute)16 KiB~0.8%approx_n_unique, Parquet dictionary decision
CardinalitySketch (polars-utils)256 B~6.5%join build side, bloom filter publishing, group-by distinct count
MinHashSketch2 KiB—fraction of rows that are distinct keys
F2Sketch512 B~18%heavy-key count in the streaming group-by

The trade-off is the usual one, and it is visible in the constants: the big sketch buys an order of magnitude of accuracy for 64 times the memory, and it is only used where the answer is shown to the user. The small sketch is used where the answer only has to tilt a decision — which side to build, whether to publish a filter, how big a table to allocate — and there the 6.5% error is free.

The register trick is what makes all of this possible: 64 bits of hash per value, a byte per register, and a maximum update, so that the whole cost of “how many distinct values are there” is one hash and one byte comparison per row, with a memory footprint that does not depend on the number of rows at all. The estimator on top is where the accuracy comes from, and Polars' choice of Ertl’s tau/sigma over HLL++’s empirical bias table is a nice example of a better algorithm replacing a lookup table.

PolarsInternalsJoinsGroup by