Opensource internals

ClickHouse Estimates Join Cardinality from Distinct-Value Counts

16 min read Jay Hubbert

Every join order ClickHouse picks is decided by a cost model built from per-column distinct-value counts, min/max ranges and row estimates.

TL;DR: Column statistics (NDV from a uniq sketch, min/max, null counts) become per-relation row estimates; an equality between two join keys gets selectivity 1 / max(NDV_left, NDV_right); the join’s output cardinality is that selectivity times the product of the two sides, floored by the preserved side for outer joins. The whole model lives in joinOrderCommon.h.

A join-order optimizer is a search algorithm wearing a costume. The search itself is generic — enumerate candidate trees, score them, pick the best — and all of the intelligence lives in the scoring function. Get the score wrong and the search is a very expensive way to pick a bad plan.

ClickHouse’s scoring function is remarkably small, and it is worth understanding in full, because the number it produces shows up in EXPLAIN output and people try to read it as a promise.

It is not a promise. It is a guess with a documented error bar, and the whole pipeline that produces it is:

text 
   per-part column statistics          per-relation estimate
   ┌───────────────────────────┐        ┌────────────────────────┐
   │ uniq sketch  -> NDV       │        │ estimated_rows         │
   │ minmax       -> [min,max] │  ────► │ column_stats[col].ndv  │
   │ basic        -> nulls     │        │  (per column)          │
   └───────────────────────────┘        └────────────────────────┘
                  │                                │
                  │  selectivity of a predicate     │  1 / max(NDV)
                  ▼                                ▼
        ┌──────────────────────────────────────────────────┐
        │  SelectivityEstimate { value, reported_value }  │
        └──────────────────────────────────────────────────┘
                  │
                  ▼
        ┌──────────────────────────────────────────────────┐
        │  estimateJoinCardinality()  -> rows of the join  │
        │  computeJoinCost()          -> cost of the join  │
        └──────────────────────────────────────────────────┘

Where the numbers come from

ClickHouse stores per-column statistics inside each MergeTree part. Three kinds matter here:

  • uniq / uniqCombined64 — distinct-value sketches. The plain uniq statistics object is literally the uniq aggregate function’s state; the V2 variant is uniqCombined64 with K = 12, which is exact for a handful of values, a hash set up to 256 of them, and a HyperLogLog above that.
  • minmax / basic — the minimum and maximum value, and (for basic) the null count and the default count.
  • tdigest, countmin — distribution sketches, used for < and = selectivity.

The per-part objects are merged into a per-column object. The NDV question has a fallback that is important to know about:

cpp 
// src/Storages/Statistics/Statistics.cpp:446
UInt64 ColumnStatistics::estimateCardinality() const
{
    if (const IStatistics * uniq_stats = findUniqStats(stats))
    {
        return uniq_stats->estimateCardinality();
    }
    /// if we don't have uniq statistics, we use a mock one, assuming there are 90% different unique values.
    return UInt64(static_cast<Float64>(rows) * ConditionSelectivityEstimator::default_cardinality_ratio);
}

With no uniq sketch, the column is assumed to have 10 % distinct values (default_cardinality_ratio = 0.1). That is a guess, and the guess propagates: the last fallback of estimateEqual divides the non-NULL row count by the distinct-value count to answer “how many rows have this value?”:

cpp 
// src/Storages/Statistics/Statistics.cpp:404
    if (uniq_stats != nullptr)
    {
        UInt64 cardinality = uniq_stats->estimateCardinality();
        if (cardinality == 0 || rows == 0)
            return 0;
        /// Uniq ignores NULLs, so divide non-NULL row count by distinct values.
        /// `getNonNullRowCount` returns `rows` when null-count tracking is absent.
        return static_cast<Float64>(getNonNullRowCount()) / static_cast<Float64>(cardinality);
    }

    return std::nullopt;

Before reaching that fallback, estimateEqual tries the distribution sketches, in descending order of accuracy — basic first, then a t-digest, then a count-min sketch:

cpp 
// src/Storages/Statistics/Statistics.cpp:389
    const IStatistics * uniq_stats = findUniqStats(stats);
    if (stats_desc.data_type->isValueRepresentedByNumber() && uniq_stats != nullptr && stats.contains(StatisticsType::TDigest))
    {
        /// 2048 is the default number of buckets in TDigest. In this case, TDigest stores exactly one value (with many rows) for every bucket.
        if (uniq_stats->estimateCardinality() < 2048)
        {
            return stats.at(StatisticsType::TDigest)->estimateEqual(val);
        }
    }

That “2048” is a nice piece of engineering folklore: a t-digest with more buckets than there are distinct values has one value per bucket, so it can answer an equality exactly.

From statistics to a relation estimate

The unit the join optimizer consumes is a RelationStats:

cpp 
// src/Processors/QueryPlan/Optimizations/RelationStatistics.h:16
struct RelationStats
{
    std::optional<UInt64> estimated_rows = {};
    std::optional<Float64> avg_row_bytes = {};
    std::unordered_map<String, ColumnStats> column_stats = {};

    String table_name;
    bool imprecise_estimate = false;

    /// Diagnostic annotation of where `estimated_rows` came from; see `RowEstimateSource`.
    /// `NoSource` means the producer of the estimate did not track it; set it wherever it is known.
    RowEstimateSource source = RowEstimateSource::NoSource;
};

and the per-column part of it is just an NDV plus some optional extras:

cpp 
// src/Storages/Statistics/ConditionSelectivityEstimator.h:14
struct ColumnStats
{
    UInt64 num_distinct_values = 0;
    /// Average uncompressed size of one value; 0 means unknown.
    Float64 avg_bytes = 0;
    /// Value range from `basic`/`minmax` statistics; unset when unknown.
    std::optional<Field> min_value = {};
    std::optional<Field> max_value = {};
    /// Fraction of NULL values; unset when unknown.
    std::optional<Float64> null_fraction = {};
};

estimateReadRowsCount is the function that produces a RelationStats for any plan node. For a MergeTree read it does three things: analyse the ranges, ask the statistics estimator for a selectivity-adjusted row count, and fall back to the primary index’s row count if there is no statistics.

cpp 
// src/Processors/QueryPlan/Optimizations/RelationStatisticsEstimator.cpp:238
        const bool use_statistics = reading->getContext()->getSettingsRef()[Setting::use_statistics];
        if (use_statistics)
        {
            if (auto estimator = reading->getConditionSelectivityEstimator(reading->getAllColumnNames(), analyzed_result))
            {
                auto relation_profile = estimator->estimateRelationProfile(reading->getStorageMetadata(), filter, prewhere_node);
                RelationStats stats{
                    .estimated_rows = relation_profile.rows,
                    .column_stats = relation_profile.column_stats,
                    .table_name = table_display_name,
                    .source = RowEstimateSource::Statistics};
                LOG_TRACE(getLogger("optimizeJoin"), "estimate statistics {}", dumpStatsForLogs(stats));
                return stats;
            }
        }

If the filter is not usable by the primary index, the estimate is withheld rather than faked:

cpp 
// src/Processors/QueryPlan/Optimizations/RelationStatisticsEstimator.cpp:263
        /// If any conditions are pushed down to storage but not used in the index,
        /// we cannot precisely estimate the row count
        if (has_filter && !is_filtered_by_index)
            return RelationStats{
                .estimated_rows = {},
                .table_name = table_display_name,
                .imprecise_estimate = true,
                .source = RowEstimateSource::NoStatistics};

An unknown row count is not the same as a row count of zero, and the difference matters: the cost model treats an unknown side as one row so that a cross product with an unknown side does not look free, but it must not be treated as empty.

Aggregation subqueries

A subquery that aggregates is a relation too, and its row count is the number of groups:

cpp 
// src/Processors/QueryPlan/Optimizations/RelationStatisticsEstimator.cpp:78
        UInt64 key_number_of_distinct_values = key_stats->second.num_distinct_values;

        if (input_stats.estimated_rows)
            key_number_of_distinct_values = std::min(key_number_of_distinct_values, *input_stats.estimated_rows);

        aggregation_stats.column_stats[key].num_distinct_values = key_number_of_distinct_values;

        /// For now assume that aggregation columns are independent, so multiply their NDVs
        if (total_number_of_distinct_values)
            *total_number_of_distinct_values *= static_cast<Float64>(key_number_of_distinct_values);

The independence assumption is worth flagging: GROUP BY a, b is estimated as NDV(a) * NDV(b), which is wrong whenever a and b are correlated, and it is clamped to the input row count so it cannot exceed reality by more than the correlations allow.

Selectivity

Now the interesting part: what does a.x = b.y keep?

The answer is 1 / max(NDV(a.x), NDV(b.y)), and the implementation says so explicitly:

cpp 
// src/Processors/QueryPlan/Optimizations/joinOrderCommon.h:68
/// Accounts for an equality between key columns with the largest NDV `max_ndv` (0 if unknown)
/// and the largest NDV upper bound `max_ndv_upper_bound` (0 if unknown), see `ColumnDistinctValues`.
inline void applyEquiKeyDistinctValues(SelectivityEstimate & estimate, UInt64 max_ndv, UInt64 max_ndv_upper_bound)
{
    estimate.has_equi = true;
    if (max_ndv > 0)
    {
        estimate.value = std::min(estimate.value, 1.0 / static_cast<double>(max_ndv));
        estimate.reliable = true;
    }
    if (max_ndv_upper_bound > 0)
        estimate.reported_value = std::min(estimate.reported_value, 1.0 / static_cast<double>(max_ndv_upper_bound));
}

The intuition is the standard one: if the key has D distinct values spread uniformly over N rows, each value appears N/D times, so a lookup of one value keeps a fraction 1/D of the other side.

Note there are two numbers, not one, and the distinction is subtle but important:

cpp 
// src/Processors/QueryPlan/Optimizations/joinOrderCommon.h:18
struct SelectivityEstimate
{
    double value = 1.0;
    /// Like `value`, but a key column without NDV statistics uses the row count of its relation
    /// (an upper bound of its NDV) instead. Reported by `EXPLAIN` as `estimated (NDV)` and used for the
    /// match fraction of semi/anti joins; the inner join cost model does not use it, because it says
    /// nothing about which side is the key.
    double reported_value = 1.0;
    bool reliable = false;
    bool has_equi = false;
};
  • value is the measured selectivity: 1/NDV when the NDV is known.
  • reported_value is the upper bound version: 1/row_count when the NDV is not known, since a column cannot have more distinct values than the relation has rows.

reliable says which one applies, and the difference changes the cardinality formula, because reliability is also the signal for “which side is the key”.

The cardinality formula

Here is the entire model:

cpp 
// src/Processors/QueryPlan/Optimizations/joinOrderCommon.h:135
inline double estimateJoinedRows(
    const SelectivityEstimate & selectivity, std::optional<UInt64> lhs_rows, std::optional<UInt64> rhs_rows)
{
    double lhs = static_cast<double>(lhs_rows.value_or(1));
    double rhs = static_cast<double>(rhs_rows.value_or(1));
    if (!lhs_rows || !rhs_rows)
        return selectivity.reported_value * lhs * rhs;
    if (selectivity.reliable)
        return selectivity.value * lhs * rhs;
    if (selectivity.has_equi)
        return std::max(lhs, rhs);
    return lhs * rhs;
}

Read it as four cases:

  1. One side unknown → the result is unknown too, and the formula falls back to the upper-bound selectivity.
  2. Reliable NDV on both sides → selectivity * lhs * rhs. The textbook answer.
  3. Equi-join, but NDV unknown → max(lhs, rhs). This is the foreign-key/primary-key heuristic: if b.y is a unique key, then joining a to b produces at most a’s row count, and the model assumes the larger side is preserved.
  4. No equi condition at all → the full cross product lhs * rhs.

Case 3 is the “we don’t know which side is the key, so assume the bigger side survives” fallback, and the comment above the function explains why it is shaped that way:

cpp 
// src/Processors/QueryPlan/Optimizations/joinOrderCommon.h:125
/// Expected number of rows an inner join of `lhs_rows` x `rhs_rows` keeps under `selectivity`.
/// When a side has no row estimate we cannot tell which side is the key, so the join is costed as
/// before the FK->PK heuristic: the missing row estimate counts as 1 row and the selectivity is the
/// NDV upper bound (`reported_value`), whatever `reliable` says. A join with an unknown relation must
/// not look more expensive than that, otherwise the only relations with known sizes (for example,
/// those with hash-table statistics hints) are pushed to the end of the join order, and a key whose
/// relation has a known size must keep making the join with it cheap even when the other side is a
/// sub-join of unknown size. With both sizes known, an equi join without reliable NDV is assumed to
/// be FK->PK (the smaller side is a unique key), so it keeps the larger side. Without any equi
/// condition (cross or range-only join) the result is the full product.

That is a long comment, and it is worth reading twice, because it encodes a failure mode that cost real debugging time: if a join with an unknown side looks more expensive than a join with two known sides, the optimizer will shove the known-size relations to the end of the order — which is exactly the wrong answer for a star query.

Outer joins

The outer-join rules are a floor, applied on top of the inner formula:

cpp 
// src/Processors/QueryPlan/Optimizations/joinOrderCommon.h:195
    double joined_rows = std::max(estimateJoinedRows(selectivity, left_rows, right_rows), 1.0);

    if (join_kind == JoinKind::Left)
        joined_rows = std::max(joined_rows, lhs);
    if (join_kind == JoinKind::Right)
        joined_rows = std::max(joined_rows, rhs);
    if (join_kind == JoinKind::Full)
        joined_rows = std::max(joined_rows, lhs + rhs);

LEFT keeps at least every left row (null-padded when unmatched), RIGHT at least every right row, FULL at least the sum.

Semi and anti joins

Semi and anti joins are filters, not expansions, and treating them like outer joins is a classic optimizer bug — it makes a selective semi join look like a row multiplier, so the optimizer refuses to push it down. ClickHouse models them explicitly:

cpp 
// src/Processors/QueryPlan/Optimizations/joinOrderCommon.h:173
    if (strictness == JoinStrictness::Semi || strictness == JoinStrictness::Anti)
    {
        /// Preserved side is the left input for LEFT (and Inner/Cross, defensively), the right
        /// input for RIGHT; the other side is only probed for existence.
        const bool preserve_left = !isRight(join_kind);
        const double preserved = preserve_left ? lhs : rhs;
        const double other = preserve_left ? rhs : lhs;
        /// Expected fraction of preserved rows with at least one match. The selectivity is ~1/ndv,
        /// so `ndv_selectivity * other` approximates matches per preserved row; cap at 1. Without
        /// reliable statistics use the row-count upper bound of NDV (`reported_value`), which treats
        /// the join as FK->PK: e.g. a semi join of 1000 rows against 10 keys keeps ~10 rows.
        const double ndv_selectivity = selectivity.reliable ? selectivity.value : selectivity.reported_value;
        const double match_fraction = std::min(1.0, ndv_selectivity * other);
        const double kept = (strictness == JoinStrictness::Semi)
            ? preserved * match_fraction
            : preserved * (1.0 - match_fraction);

A semi join of 1000 rows against a 10-key side keeps about 1000 * min(1, 0.1 * 10) = 1000 rows, i.e. everything, because the 10 keys cover all 1000 rows. An anti join of the same shape keeps 0.

NDV propagation through the join tree

A cardinality estimate is only as good as the NDVs feeding it, and NDVs change as you join. Two mechanisms keep them honest.

Merging. When a join is built, the two children’s column statistics are merged, and the NDVs of the equi-join key columns are tightened to the minimum of the two sides:

cpp 
// src/Processors/QueryPlan/Optimizations/RelationStatistics.cpp:35
    const UInt64 minimum = std::min(left_stats.num_distinct_values, right_stats.num_distinct_values);
    if (update_left)
        left_stats.num_distinct_values = minimum;
    if (update_right)
        right_stats.num_distinct_values = minimum;

Which side gets tightened depends on the join kind, because an outer join preserves one side’s keys untouched:

cpp 
// src/Processors/QueryPlan/Optimizations/RelationStatistics.cpp:26
    else if (strictness != JoinStrictness::Anti)
    {
        /// An outer join preserves the named side, so only the non-preserved side can lose key values.
        update_left = kind == JoinKind::Inner || kind == JoinKind::Right
            || kind == JoinKind::Cross || kind == JoinKind::Comma;
        update_right = kind == JoinKind::Inner || kind == JoinKind::Left
            || kind == JoinKind::Cross || kind == JoinKind::Comma;
    }

Capping. A join cannot produce more distinct values in a column than it produces rows, and the constructor enforces it:

cpp 
// src/Processors/QueryPlan/Optimizations/joinOrder.cpp:99
    /// Cap all NDVs at the estimated output rows.
    if (cardinality_)
    {
        for (auto & [_, stats] : column_stats)
            stats.num_distinct_values = std::min(stats.num_distinct_values, *cardinality_);
    }

This is the mechanism that stops a chain of joins from inflating the NDV without bound: after a ⋈ b produces 100 rows, every column of the result has an NDV of at most 100, and the next join’s selectivity is computed from that.

Reading the numbers in EXPLAIN

The estimates are surfaced, with a documented meaning:

cpp 
// src/Parsers/ParserExplainQuery.cpp:783
- `Cost` — the cost of the whole join subtree under this step, which is the value the optimizer
  minimizes when it compares candidate join orders. The cost of a join is its estimated number of
  matched row pairs, `<selectivity> * <left_rows> * <right_rows>`, plus the cost of its inputs.
- `Selectivity` — the estimated fraction of the Cartesian product of the two sides that survives
  the join condition. It is derived from the number of distinct values (NDV) of the join keys:
  a key equality keeps about `1 / max(NDV_left, NDV_right)` of the pairs, and the smallest
  fraction over the join conditions is used.
- `Output rows` — the estimated number of rows the join produces:
  `<selectivity> * <left_rows> * <right_rows>` for an inner join, floored at `<left_rows>` for
  `LEFT`, at `<right_rows>` for `RIGHT`, and at `<left_rows> + <right_rows>` for `FULL`, because an
  outer join keeps every row of its preserved side. When a `SEMI` or `ANTI` join takes part in the
  reordering, it is estimated as a fraction of its preserved side,
  `<preserved_rows> * min(1, <selectivity> * <other_rows>)` for `SEMI` and the remaining rows for
  `ANTI`; otherwise it uses the formula of its join kind.
- `Left` / `Right` — the estimated number of rows entering the join from each side.

and the ~~ prefix on ResultRows marks an imprecise estimate, which happens when any leaf of the subtree had no statistics.

A worked example: two tables of 100 rows each, joined on a column with 100 distinct values (so NDV = 100 on both sides):

text 
│  Cost: estimated 100.00
│  Selectivity: estimated (NDV) 0.01
│  Output rows: estimated 100.00
│  Left: rows estimated 100.00
│  Right: rows estimated 100.00

0.01 = 1/100 is the selectivity, 0.01 * 100 * 100 = 100 is the output row count, and 100 is also the cost (the matched-pair count) because the inputs are free.

Where the estimates are not used

This is worth being explicit about, because people assume the estimate drives everything.

The join-order optimizer’s cost model is the primary consumer. The same estimate is also handed to the physical hash join as a hint for how much to preallocate:

cpp 
// src/Processors/QueryPlan/JoinStepLogical.cpp:2596
        if (join_step->right_relation.estimated_rows)
            join_step->join_algorithm_params->rhs_size_estimation = join_step->right_relation.estimated_rows;

and the same numbers are cached: a hash table built from a given subtree and a given set of key columns is remembered with its measured row count, and that measurement is fed back into the next query’s estimate:

cpp 
// src/Processors/QueryPlan/Optimizations/optimizeJoin.cpp:550
    std::optional<size_t> num_rows_from_cache = graph.context->statistics_context.getCachedHint(node);
    if (graph.context->join_settings.use_hash_table_stats_for_join_reordering && num_rows_from_cache
        && (!stats.estimated_rows || num_rows_from_cache.value() < stats.estimated_rows.value()))
    {
        /// A measured row count beats statistics: take the minimum and mark it a precise cache value.
        stats.estimated_rows = num_rows_from_cache;
        stats.imprecise_estimate = false;
        stats.source = RowEstimateSource::HashTableCache;
    }

That last snippet is the real feedback loop: a measured row count replaces a modelled one, and the next query’s plan is better because of it. The setting that turns it on is use_hash_table_stats_for_join_reordering, on by default since 26.1.

Getting better numbers

The estimate is only as good as the statistics underneath it. The knobs:

sql 
-- Turn on column statistics for the joined columns.
ALTER TABLE t1 MATERIALIZE STATISTICS ALL;

-- Check what the optimizer believes about a table.
SELECT * FROM system.columns
WHERE table = 't1' AND column = 'id';
sql 
-- Statistics are used by default; turn them off to see the difference.
SELECT * FROM a JOIN b ON a.id = b.id SETTINGS use_statistics = 0;

The diagnostic the optimizer itself emits is the most useful one — if a relation has no column statistics, the log says so by name:

cpp 
// src/Processors/QueryPlan/Optimizations/optimizeJoin.cpp:981
            "Join order optimization uses imprecise row count estimates derived from the primary index "
            "because the following relation(s) have no column statistics available for join reordering: {}. "
            "The chosen join order may be suboptimal. Consider creating column statistics for the joined tables "
            "(for example, `ALTER TABLE <table> MATERIALIZE STATISTICS ALL`) and enabling the 'use_statistics' setting",

Trade-offs

The model is one formula long, and that is the point. 1/max(NDV) is a crude estimator: it assumes uniform distribution of keys, ignores correlation between two key columns, ignores the actual overlap of the two key sets, and ignores the physical layout. A real optimizer would use histograms and intersection estimates. ClickHouse trades that accuracy for a cost model that is fast enough to call for every join in every query — the whole estimate is a handful of floating-point operations per candidate pair.

The fallbacks are conservative on purpose. When the NDV is unknown, the model does not use 1/row_count as if it were measured; it marks the estimate unreliable and switches to the FK→PK heuristic. Under-estimating selectivity makes a join look cheaper than it is, which is the dangerous direction; the model’s fallbacks are arranged so that a missing statistic never makes a join look free.

Statistics are a maintenance cost. They are stored per part, they must be merged on part merges, they must be rebuilt after a ALTER, and they are cached in memory. The use_statistics setting exists because for some workloads the modelling error is larger than the benefit.

The feedback loop is the real win. The hash-table statistics cache turns the optimiser from “guessing from sketches” into “remembering what actually happened”, for the exact subtree-and-key combination that was measured.

ClickhouseQuery planningStatisticsCardinalityOptimizer