DataFusion's DISTINCT Is a GROUP BY, and approx_distinct Is a HyperLogLog Over Hashes

TL;DR: SELECT DISTINCT a, b is not a separate operator — the optimizer rewrites it into GROUP BY a, b, which is the same hash-table machinery as any other aggregation. When the exact answer does not fit in memory, approx_distinct swaps the exact set for a 16384-register HyperLogLog, and the sketch is indexed by the hash of the value: hashing is what gives you a uniform 64-bit key for free, and a collision just costs you a little accuracy.

Part one: DISTINCT is a GROUP BY

The instinct is to treat DISTINCT as its own operator with its own algorithm. DataFusion does not, and the reason is that a distinct set is a grouping: SELECT DISTINCT a, b returns exactly the groups of GROUP BY a, b, one row each.

The optimizer rule says so outright:

rust
// datafusion/optimizer/src/replace_distinct_aggregate.rs:31-64
/// Optimizer that replaces logical [[Distinct]] with a logical [[Aggregate]]
///
/// ```text
/// SELECT DISTINCT a, b FROM tab
/// ```
///
/// Into
/// ```text
/// SELECT a, b FROM tab GROUP BY a, b
/// ```

rust
// replace_distinct_aggregate.rs:87-127
            LogicalPlan::Distinct(Distinct::All(input)) => {
                let group_expr = expand_wildcard(input.schema(), &input, None)?;

                if group_expr.is_empty() {
                    // Special case: there are no columns to group by, so we can't replace it by a group by
                    // however, we can replace it by LIMIT 1 because there is either no output or a single empty row
                    return Ok(Transformed::yes(Arc::unwrap_or_clone(input)));
                }

                let field_count = input.schema().fields().len();
                for dep in input.schema().functional_dependencies().iter() {
                    // If the input is already unique on all of its columns (e.g.
                    // it is a GROUP BY over exactly these columns), the DISTINCT
                    // is a no-op and we can simply remove it. The dependency mode
                    // must be `Single`: a `Multi` dependence (e.g. a former key
                    // downgraded by a join) means equal rows may occur multiple
                    // times, so the DISTINCT still has work to do.
                    if dep.mode == Dependency::Single
                        && dep.source_indices.len() >= field_count
                        && dep.source_indices[..field_count]
                            .iter()
                            .enumerate()
                            .all(|(idx, f_idx)| idx == *f_idx)
                    {
                        return Ok(Transformed::yes(Arc::unwrap_or_clone(input)));
                    }
                }

                // Replace with aggregation:
                let aggr_plan = LogicalPlan::Aggregate(Aggregate::try_new(
                    input,
                    group_expr,
                    vec![],
                )?);

Three cases in twenty lines:

No columns at all (a table with no columns): there is at most one row, so the answer is LIMIT 1:

rust
// replace_distinct_aggregate.rs:93-101
                if group_expr.is_empty() {
                    // Special case: there are no columns to group by, so we can't replace it by a group by
                    // however, we can replace it by LIMIT 1 because there is either no output or a single empty row
                    return Ok(Transformed::yes(LogicalPlan::Limit(Limit {
                        skip: None,
                        fetch: Some(Box::new(lit(1i64))),
                        input,
                    })));
                }

The input is already unique on all columns (functional dependency with mode Single): DISTINCT is a no-op and is deleted. The Single vs Multi distinction matters — a Multi dependency means “these columns are functionally determined by something else”, so equal rows can still repeat.
Everything else: rewrite to GROUP BY on the expanded wildcard.

DISTINCT ON is more interesting, because it is not a plain grouping:

rust
// replace_distinct_aggregate.rs:41-51
/// On the other hand, for a `DISTINCT ON` query the replacement is
/// a bit more involved and effectively converts
/// ```text
/// SELECT DISTINCT ON (a) b FROM tab ORDER BY a DESC, c
/// ```
/// into
/// ```text
/// SELECT b FROM (
///     SELECT a, FIRST_VALUE(b ORDER BY a DESC, c) AS b
///     FROM tab
///     GROUP BY a
/// )
/// ORDER BY a DESC
/// ```

DISTINCT ON (a) becomes GROUP BY a plus first_value(b ORDER BY ...). The group key supplies the uniqueness; the ordering-sensitive aggregate supplies the payload.

`count(DISTINCT x)` becomes two phases

count(DISTINCT b) cannot be a plain group by — you need the distinct values of b within each group a. DataFusion splits it into an inner group by over the finer grain and an outer aggregate over the coarse one:

rust
// datafusion/optimizer/src/single_distinct_to_groupby.rs:40-56
/// single distinct to group by optimizer rule
///  ```text
///    Before:
///    SELECT a, count(DISTINCT b), sum(c)
///    FROM t
///    GROUP BY a
///
///    After:
///    SELECT a, count(alias1), sum(alias2)
///    FROM (
///      SELECT a, b as alias1, sum(c) as alias2
///      FROM t
///      GROUP BY a, b
///    )
///    GROUP BY a
///  ```

The inner GROUP BY a, b produces one row per (a, distinct b), and the outer count(alias1) counts them. Note that the inner group by has no aggregate at all — it is a pure distinct — which is why the outer aggregate’s input is one row per distinct value.

The rule is careful about when the rewrite pays. It only applies if every distinct aggregate shares the same argument expression, and it refuses to extend the rewrite to a plan that only qualifies because of a non-distinct count:

rust
// single_distinct_to_groupby.rs:176-196
/// Whether the rewrite is worth extending to a plan that only qualifies because
/// of the non-distinct `count`.
///
/// The rewrite is not free: every other aggregate moves down to the inner group
/// by, which has a row per `(group, distinct value)` pair rather than per group,
/// and keeps its state at that finer grain. What pays for that is taking the
/// distinct aggregate off `GroupsAccumulatorAdapter` and its one boxed
/// accumulator per group. An aggregate that already has a specialized
/// `GroupsAccumulator` never went near the adapter, so there is nothing to buy
/// and only the inner group by to pay for.

`DISTINCT` on an idempotent aggregate is free

Not every DISTINCT needs rewriting. min(DISTINCT x) is min(x), because the merge is idempotent. DataFusion encodes this as a property of the aggregate function itself:

rust
// datafusion/optimizer/src/eliminate_aggregate_distinct.rs:30-52
/// Optimizer rule that removes a `DISTINCT` modifier that cannot change the
/// result of the aggregate it is attached to.
///
/// `min`, `max`, `bool_and`, `bit_or` and friends have an idempotent merge, so
/// `min(DISTINCT x)` and `min(x)` return the same value. ...

and approx_distinct declares itself insensitive to duplicates, because folding a value into a HyperLogLog twice changes nothing:

rust
// datafusion/functions-aggregate/src/approx_distinct.rs:903-906
    fn distinct_handling(&self) -> DistinctHandling {
        // Updating an HLL register with a value already seen is a no-op.
        DistinctHandling::Insensitive
    }

The physical side: early termination

Once DISTINCT is a GROUP BY with no aggregates, the physical plan can stop as soon as it has enough groups. AggregateExec has a predicate for exactly that shape:

rust
// datafusion/physical-plan/src/aggregates/mod.rs:1775-1813
    /// true, if this Aggregate has a group-by with no required or explicit ordering,
    /// no filtering and no aggregate expressions
    /// This method qualifies the use of the LimitedDistinctAggregation rewrite rule
    /// on an AggregateExec.
    pub fn is_unordered_unfiltered_group_by_distinct(&self) -> bool {

and the stream checks the soft limit after every batch:

rust
// datafusion/physical-plan/src/aggregates/single_stream.rs:105-114
/// After each input batch, the stream checks whether the soft limit has been
/// reached. If so, it emits the accumulated groups and stops reading input.
///
/// This operator does not guarantee an exact limit because a single batch can
/// cross the threshold. The downstream limit operator enforces the exact result
/// size.

So SELECT DISTINCT x FROM t LIMIT 10 becomes a group-by that reads one batch, maybe two, and stops.

Part two: when the exact set does not fit

The exact rewrite has a cost: the hash table holds one entry per distinct value, forever. SELECT DISTINCT user_id FROM events over ten billion events needs a table with as many entries as there are distinct users. That is a memory wall, not a CPU wall.

approx_distinct replaces the exact set with a fixed-size sketch. The sketch is a HyperLogLog with 2^14 registers:

rust
// datafusion/functions-aggregate/src/hyperloglog.rs:26-48
/// Specifically, like Redis's version, this HLL structure uses
/// 2**14 = 16384 registers, which means the standard error is
/// 1.04/(16384**0.5) = 0.8125%. Unlike Redis, the register takes
/// up full [`u8`] size instead of a raw int* and thus saves some
/// tricky bit shifting techniques used in the original version.
/// This results in a memory usage increase from 12Kib to 16Kib.
/// Also only the dense version is adopted, so there's no automatic
/// conversion, largely to simplify the code.
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;
pub(crate) 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;

16 KiB per group, fixed, regardless of cardinality.

The trick: hash the value, don’t encode it

Here is the core design decision, and it is the one worth understanding.

A HyperLogLog needs, for each distinct value, a uniformly random 64-bit key. It then splits that key into a register index (the low 14 bits) and a “rho” (the position of the first zero bit among the remaining 50 bits).

The value itself cannot be that key. It might be a 200-byte string, a Decimal256, a nested Struct. There is no canonical 64-bit projection of those. You could invent one, but it would either be lossy in a way you have to reason about, or it would cost an allocation per value.

Hashing gives you the key for free, and DataFusion already has a vectorized hashing engine, because hash aggregation needs exactly the same thing:

rust
// datafusion/functions-aggregate/src/approx_distinct.rs:203-225
impl Accumulator for HLLAccumulator {
    fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
        let array = values[0].as_ref();
        self.hashes.clear();
        self.hashes.resize(array.len(), 0);
        create_hashes([array], &HLL_HASH_STATE, &mut self.hashes)?;

        match array.logical_nulls() {
            None => {
                for &hash in &self.hashes {
                    self.hll.add_hashed(hash);
                }
            }
            Some(nulls) => {
                for row in 0..array.len() {
                    if nulls.is_valid(row) {
                        self.hll.add_hashed(self.hashes[row]);
                    }
                }
            }
        }
        Ok(())
    }

One vectorized pass over the array produces 64 bits per row. Then folding into the sketch is two instructions:

rust
// hyperloglog.rs:116-125
    /// Adds a pre-computed hash value directly to the HyperLogLog.
    ///
    /// The hash should be computed using [`HLL_HASH_STATE`], the same hasher used
    /// by [`Self::add`].
    #[inline]
    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);
    }

Read that as: use 14 bits to pick a register, count the leading zeros of the other 50, and keep the maximum seen for that register. That is the whole algorithm. The register value is “the longest run of leading zeros observed for values landing in this bucket”, which is a proxy for how close the value was to the bucket boundary — a proxy for “is this value probably unique”.

Why hashing is the right move, not a hack:

Uniformity. A good hash makes the register indices uniform and the leading-zero counts geometric, which is exactly the distribution the estimator assumes. The error bound is only valid under that assumption.
Collisions are already priced in. If two distinct values collide in all 64 bits, the sketch merges them — the count is off by one, in a scheme whose error is already ±0.8%. A 64-bit collision between two values in a billion-row stream has probability ~2.7e-8; it does not need to be prevented, only to be unlikely.
The cost is already paid. create_hashes is the same function hash aggregation uses for its group keys, and it is vectorized over the whole batch.
It works for any type. Strings, decimals, lists, structs, dictionaries — all of them hash to 64 bits through the same path.

The price is that the sketch is no longer a function of the values, only of their hashes. Two different values with the same hash are indistinguishable. That is precisely the approximation being bought.

Two hashers, and why the seed is fixed

DataFusion has two hash functions, and the split is deliberate:

rust
// datafusion/common/src/hash_utils.rs:34-48
/// [`RandomState`] is optimized for speed and suitable for hash tables and
/// bloom filters. [`QualityRandomState`] is optimized for statistical quality
/// and suitable for algorithms such as HyperLogLog. The tradeoff is that the
/// fast variant gives up some statistical quality, while the quality variant
/// is slightly slower.
///
/// See: <https://docs.rs/foldhash/0.2.0/src/foldhash/lib.rs.html#17-21>
pub type RandomState = FixedState;
pub type QualityRandomState = foldhash::quality::FixedState;

/// Fixed quality hash state used by HyperLogLog sketches.
///
/// The seed is part of the HLL wire/storage semantics: serialized sketches only
/// remain mergeable if every producer uses the same hash state.
pub const HLL_RANDOM_STATE: QualityRandomState = QualityRandomState::with_seed(0);

Aggregation uses foldhash’s fast variant, because a hash table only needs good-enough distribution. HyperLogLog uses the quality variant, because the estimator’s error bound depends on the hash being genuinely uniform. And the seed is 0, fixed forever, because a serialized sketch is 16 KiB of opaque bytes: if two nodes hashed differently, their sketches could not be merged.

Estimating without materializing registers

The estimator is the classic one, with the modern bias corrections:

rust
// hyperloglog.rs:206-217
#[inline]
fn count_from_histogram(histogram: &[u32; HLL_Q + 2]) -> usize {
    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
}

tau and sigma are Ertl’s corrections for small and large cardinalities. But note what this function needs: a histogram of register values, i.e. the 16 KiB register array.

For the per-group case, DataFusion avoids materializing that array for small groups. GroupHll is a sparse/dense hybrid:

rust
// datafusion/functions-aggregate/src/approx_distinct.rs:296-334
/// Maximum number of distinct hashes kept in the sparse representation of a
/// per-group sketch before it is promoted to a dense [`HyperLogLog`].
///
/// A dense sketch always occupies [`NUM_REGISTERS`] (16 KiB) regardless of how
/// many values it has seen. The vast majority of groups in a high-cardinality
/// `GROUP BY` only observe a handful of distinct values, so keeping their state
/// as a small list of hashes saves a huge amount of memory (both while
/// aggregating and when serializing the partial state for the final phase).
const SPARSE_LIMIT: usize = 256;

/// Per-group HyperLogLog state used by [`HllGroupsAccumulator`].
///
/// Starts out as a compact list of the (deduplicated) hashes observed for the
/// group and only switches to a full dense [`HyperLogLog`] once it has seen more
/// than [`SPARSE_LIMIT`] distinct values. Folding the stored hashes into a dense
/// sketch produces exactly the same registers as adding the original values one
/// by one, so the cardinality estimate is identical to the per-group
/// [`Accumulator`] path.
#[derive(Clone, Debug)]
enum GroupHll {
    /// Distinct hashes seen so far. May contain duplicates between compactions.
    Sparse(Vec<u64>),
    Dense(Box<HyperLogLog<u8>>),
}

The estimate for a sparse group is computed directly from the hash list, without ever building a register array:

rust
// hyperloglog.rs:167-204
/// Estimate the cardinality of a set of precomputed hashes without
/// materializing a full [`NUM_REGISTERS`]-byte register array.
///
/// This is equivalent to adding every hash to a fresh [`HyperLogLog`] via
/// [`HyperLogLog::add_hashed`] and calling [`HyperLogLog::count`], but only does
/// work proportional to the number of hashes. It 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.
///
/// `hashes` may contain duplicates (duplicate hashes are idempotent).
pub(crate) fn count_from_hashes(hashes: &[u64]) -> usize {
    ...
    // For each touched register index keep the maximum rho. Sorting by
    // (index, rho) groups equal indices together with the max rho last.
    let mut idx_rho: Vec<(usize, u8)> =
        hashes.iter().map(|&hash| register_for_hash(hash)).collect();
    idx_rho.sort_unstable();

    let mut histogram = [0u32; HLL_Q + 2];
    let mut touched = 0u32;
    ...
    // All remaining registers are still zero.
    histogram[0] = NUM_REGISTERS as u32 - touched;
    count_from_histogram(&histogram)
}

Sorting the (index, rho) pairs and taking the max rho per index gives the same histogram as a full register array would — in O(k log k) time and O(k) memory instead of 16 KiB. For the typical group in a high-cardinality GROUP BY (a handful of distinct values), this is the difference between 16 KiB and 16 bytes.

The promotion rule is also careful: the sparse list is compacted (sorted and deduped) at twice the limit, and only promoted to dense if it still exceeds the limit:

rust
// approx_distinct.rs:357-377
    /// Deduplicate the sparse hash list and, if it still exceeds
    /// [`SPARSE_LIMIT`] distinct values, promote it to a dense sketch.
    #[cold]
    fn compact_or_promote(&mut self, cap_before: usize) -> isize {
        let GroupHll::Sparse(v) = self else {
            return 0;
        };
        v.sort_unstable();
        v.dedup();
        if v.len() > SPARSE_LIMIT {
            // cap_before is the capacity already reflected in allocated_bytes.
            // Any reallocation caused by the triggering push was never counted and
            // is also freed here, so the two cancel out.
            *self = GroupHll::Dense(Box::new(fold_sparse_to_hll(v)));
            (NUM_REGISTERS as isize) - ((cap_before * size_of::<u64>()) as isize)
        } else {

The memory accounting is delta-based: add_hash returns the change in heap bytes so the accumulator can track its footprint incrementally, and the dense promotion’s 16 KiB is charged at the moment it happens.

The grouped path

For SELECT k, approx_distinct(v) FROM t GROUP BY k there is a dedicated groups accumulator, one sketch per group:

rust
// approx_distinct.rs:460-490
/// A [`GroupsAccumulator`] for `approx_distinct` that keeps one adaptive
/// (sparse → dense) HyperLogLog sketch per group.
///
/// This is dramatically faster than the generic `GroupsAccumulatorAdapter`
/// fallback for high-cardinality `GROUP BY`s: it processes the whole input in a
/// single vectorized pass (no per-group `take`/slice and no dynamic dispatch),
/// and the sparse representation avoids allocating a 16 KiB sketch for every
/// group when most groups only see a few distinct values.
///
/// # Example
///
/// For `SELECT k, approx_distinct(v) FROM t GROUP BY k`, each group owns one
/// independent sketch:
///
/// ```text
/// group   state
/// a       Sparse([h1, h2, h3, h2])
/// b       Dense(HLL registers)
/// ...
/// ```

Its update loop pre-combines the filter and the value nulls into one mask, so the inner loop visits only the rows that should affect the sketch:

rust
// approx_distinct.rs:524-558
    fn update_batch(
        &mut self,
        values: &[ArrayRef],
        group_indices: &[usize],
        opt_filter: Option<&BooleanArray>,
        total_num_groups: usize,
    ) -> Result<()> {
        self.ensure_groups(total_num_groups);
        let array = values[0].as_ref();
        self.hashes.clear();
        self.hashes.resize(array.len(), 0);
        create_hashes([array], &HLL_HASH_STATE, &mut self.hashes)?;

        let mut delta: isize = 0;
        // Pre-combine value-nulls and filter into one mask so the update loop
        // only visits rows that should affect the sketch.
        let filter_nulls = opt_filter.map(filter_to_nulls);
        let value_nulls = array.logical_nulls();
        let combined_nulls =
            NullBuffer::union(filter_nulls.as_ref(), value_nulls.as_ref());
        match combined_nulls {
            None => {
                for (row, &hash) in self.hashes.iter().enumerate() {
                    delta += self.groups[group_indices[row]].add_hash(hash);
                }
            }
            Some(nulls) => {
                for row in nulls.valid_indices() {
                    delta += self.groups[group_indices[row]].add_hash(self.hashes[row]);
                }
            }
        }
        self.apply_delta(delta);
        Ok(())
    }

Note that the whole batch is hashed in one vectorized pass, and the group index array is used to scatter the hashes into per-group sketches. There is no per-group take and no dynamic dispatch per row — which is exactly the overhead the generic GroupsAccumulatorAdapter fallback would pay.

Exact when exact is cheap

The most interesting decision in approx_distinct is that DataFusion does not use a HyperLogLog when the value domain is small enough to count exactly. For booleans and 8/16-bit integers, an exact bitmap is smaller and more accurate than a sketch:

rust
// approx_distinct.rs:689-711
#[cold]
fn get_fixed_domain_approx_accumulator(
    data_type: &DataType,
) -> Result<Box<dyn Accumulator>> {
    match data_type {
        DataType::Boolean => Ok(Box::new(ApproxDistinctBitmapWrapper {
            inner: BooleanDistinctCountAccumulator::new(),
        })),
        DataType::UInt8 => Ok(Box::new(ApproxDistinctBitmapWrapper {
            inner: BoolArray256DistinctCountAccumulator::new(),
        })),
        DataType::Int8 => Ok(Box::new(ApproxDistinctBitmapWrapper {
            inner: BoolArray256DistinctCountAccumulatorI8::new(),
        })),
        DataType::UInt16 => Ok(Box::new(ApproxDistinctBitmapWrapper {
            inner: Bitmap65536DistinctCountAccumulator::new(),
        })),
        DataType::Int16 => Ok(Box::new(ApproxDistinctBitmapWrapper {
            inner: Bitmap65536DistinctCountAccumulatorI16::new(),
        })),

A boolean needs 2 bits, a u8 needs 256 bits (32 bytes), a u16 needs 65536 bits (8 KiB). Those are exact, and the ApproxDistinctBitmapWrapper exists only to convert the Int64 result into the UInt64 that approx_distinct declares as its return type. The same reasoning drives the exact count(DISTINCT x), which uses a HashSet for wide integers (datafusion/functions-aggregate-common/src/aggregate/count_distinct/native.rs:46-53).

Merging sketches

HyperLogLogs merge, which is what makes them usable in a distributed plan:

rust
// hyperloglog.rs:139-150
    /// 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]);
        }
    }

Register-wise maximum. That is the entire merge operation, and it is exactly why the sketch is 16 KiB of u8 rather than a compressed bitmap: the merge is a max over bytes, vectorizable and branch-free.

Sparse states merge too, by folding the stored hashes back in:

rust
// approx_distinct.rs:174-184
        match SerializedHll::decode(v)? {
            SerializedHll::Dense(registers) => {
                hll.merge(&HyperLogLog::new_with_registers(*registers));
            }
            SerializedHll::Sparse(chunks) => {
                for chunk in chunks {
                    hll.add_hashed(u64::from_le_bytes(*chunk));
                }
            }
        }

The wire format is self-describing: exactly NUM_REGISTERS bytes means a dense sketch; any multiple of 8 bytes means a sparse list of hashes.

One subtlety: the hash must be of the logical value

Because the sketch is keyed by hash, the hash must be a function of the value, not of its physical representation. Arrow’s Utf8View type makes this non-trivial: a short string can live inline in the view buffer, while a long one lives in a separate data buffer. If the two layouts hashed differently, the same string would count as two distinct values.

DataFusion has a dedicated test for exactly this:

rust
// approx_distinct.rs:1336-1361
    /// Regression: a short (≤ 12-byte) Utf8View string must hash identically
    /// in an all-inline batch and in a mixed batch that also contains a long
    /// string (which forces a data buffer).
    #[test]
    fn utf8view_groups_short_string_hashed_consistently_across_batches() {

and the underlying hasher has an explicit fast path for the all-inline case that hashes the view rather than the bytes:

rust
// datafusion/common/src/hash_utils.rs:463-490
fn hash_generic_byte_view_array<T: ByteViewType>(
    array: &GenericByteViewArray<T>,
    random_state: &impl HashState,
    hashes_buffer: &mut [u64],
    rehash: bool,
) {
    // instantiate the correct version based on presence of nulls and external buffers
    match (
        array.null_count() != 0,
        !array.data_buffers().is_empty(),
        rehash,
    ) {
        // no nulls or buffers ==> hash the inlined views directly
        // don't call the inner function as Rust seems better able to inline this simpler code (2-3% faster)
        (false, false, false) => {
            for (hash, &view) in hashes_buffer.iter_mut().zip(array.views().iter()) {
                *hash = view.hash_one(random_state);
            }
        }

Trade-offs

The error is not free, and not uniform. ±0.8125% standard error at P = 14, worst at small cardinalities (where tau has to correct a large bias) and at very large ones (where the 64-bit hash space saturates around ~10^19 distinct values). If you need an exact count, the GROUP BY rewrite or count(DISTINCT x) is right.
The hash must be stable. HLL_RANDOM_STATE is a const with seed 0. If it ever changed, sketches written by an older version would silently stop merging with newer ones — and would count the same value twice across a version boundary.
Collisions merge values. Two distinct values with the same 64-bit hash count as one. At a billion distinct values the collision probability is ~2.7e-8 — negligible, but it is the price of the trick, and it is why the hasher must be the quality variant.
Not every type is supported. Float16/32/64 are rejected outright by approx_distinct (“Support for ‘approx_distinct’ for data type Float64 is not implemented”), because hashing a float is ambiguous around 0.0/-0.0 and NaN payloads. The grouped path excludes them too.
Dense sketches are 16 KiB per group. The sparse/dense hybrid keeps the common case small, but a group that sees more than 256 distinct values pays the full 16 KiB — and the dense state is what gets serialized between the partial and final stages.
The exact DISTINCT rewrite has a memory wall of its own. It is semantically perfect but costs one hash table entry per distinct value, per partition. That is the problem approx_distinct exists to solve, and also why the SPARSE_LIMIT hybrid matters: a high-cardinality GROUP BY with approx_distinct per group would otherwise allocate 16 KiB times the number of groups.
The count(DISTINCT) rewrite moves every other aggregate to a finer grain. The inner group by “has a row per (group, distinct value) pair rather than per group”, which is why the rule checks that it actually buys something before applying it.

The takeaway

Two ideas, one shape. DISTINCT is not an algorithm, it is a GROUP BY with no aggregates — which means it inherits the entire hash-aggregation machinery, including the soft-limit early termination. approx_distinct is not an exact set, it is a fixed-size probabilistic summary whose state is indexed by a hash rather than by the value.

The hash is the enabling trick, and it is a good one because the cost of hashing has already been paid by the engine that sits next to it. Everything else — the sparse/dense hybrid, the register-wise-max merge, the fixed seed, the fixed-domain exact bitmaps, the Utf8View consistency test — is careful bookkeeping around the fact that once you throw away the value and keep only its hash, the only thing you can still guarantee is a number that is close.