How ClickHouse Runs UNION, INTERSECT and EXCEPT

TL;DR: The three set operations are three different plans. UNION ALL merges the branches’ output ports into one pipe and is done. UNION DISTINCT adds a hash-based DISTINCT, but first a preliminary DISTINCT that runs on the parallel streams and may abandon itself if the input turns out to be mostly unique. INTERSECT/EXCEPT build a hash set of the right side and then stream the left side through a filter.

sql
SELECT country FROM hits_2024
UNION DISTINCT
SELECT country FROM hits_2025;

The word “set operation” suggests one mechanism. There are three, and they have almost nothing in common beyond the syntax.

UNION ALL: merge the pipes

UNION ALL is the simplest thing in the database. Each branch is planned independently, its header is reconciled with the others’, and the pipelines are merged:

cpp
// src/Interpreters/InterpreterSelectWithUnionQuery.cpp:289
        std::vector<std::unique_ptr<QueryPlan>> plans(num_plans);
        SharedHeaders headers(num_plans);

        for (size_t i = 0; i < num_plans; ++i)
        {
            plans[i] = std::make_unique<QueryPlan>();
            nested_interpreters[i]->buildQueryPlan(*plans[i]);

            if (!blocksHaveEqualStructure(*plans[i]->getCurrentHeader(), *result_header))
            {
                auto actions_dag = ActionsDAG::makeConvertingActions(
                        plans[i]->getCurrentHeader()->getColumnsWithTypeAndName(),
                        result_header->getColumnsWithTypeAndName(),
                        ActionsDAG::MatchColumnsMode::Position,
                        context);
                auto converting_step = std::make_unique<ExpressionStep>(plans[i]->getCurrentHeader(), std::move(actions_dag));
                converting_step->setStepDescription("Conversion before UNION");
                plans[i]->addStep(std::move(converting_step));
            }

            headers[i] = plans[i]->getCurrentHeader();
        }

        auto max_threads = getEffectiveMaxThreads(settings);
        auto union_step = std::make_unique<UnionStep>(std::move(headers), max_threads, /* allow_narrowing = */ true);

        query_plan.unitePlans(std::move(union_step), std::move(plans));

The Conversion before UNION step is what makes the branches line up: the branches are optimised independently, so one of them may have constant-folded a column into a Const while its sibling did not, and the types may differ (Int32 vs Int64). The conversion is by position, not by name.

UnionStep then unites the pipelines. The union itself is a pipe-level operation — the branches’ output ports are simply gathered into one pipe:

cpp
// src/QueryPipeline/Pipe.cpp:323
    for (auto & pipe : pipes)
    {
        if (!allow_empty_header || !pipe.header->empty())
            assertCompatibleHeader(*pipe.header, *res.header, "Pipe::unitePipes");

        res.processors->insert(res.processors->end(), pipe.processors->begin(), pipe.processors->end());
        res.output_ports.insert(res.output_ports.end(), pipe.output_ports.begin(), pipe.output_ports.end());

        res.max_parallel_streams += pipe.max_parallel_streams;

No processor, no copy, no buffering: the output ports of every branch hang off the same pipe and the executor reads from them round-robin. A UNION ALL in ClickHouse is an interleave, not a concatenation. The rows of branch 1 and branch 2 come out in whatever order the scheduler produces them.

That also means UNION ALL cannot preserve ordering, and the plan says so explicitly — the UnionStep may narrow the number of streams, which is only allowed if no downstream consumer depends on the order:

cpp
// src/Processors/QueryPlan/UnionStep.cpp:126
    /// The `max_streams_for_union_step*` cap only applies to steps built for SQL
    /// `UNION ALL` / `UNION DISTINCT`, and only while no downstream step relies on
    /// per-stream sortedness of the union output. For all other cases the narrowing
    /// must be skipped: shuffling streams through `ConcatProcessor` would break the
    /// ordering invariants of `GroupingAggregatedTransform` (memory-efficient
    /// distributed aggregation), `MergingSortedTransform`, and similar order-sensitive
    /// consumers. We still validate the ratio so misconfiguration is reported on every
    /// query rather than only when a narrowable `UNION` happens to be present.

UNION DISTINCT: a hash set, twice

UNION DISTINCT is UNION ALL plus a DISTINCT. The interesting part is that there are two DISTINCTs:

cpp
// src/Interpreters/InterpreterSelectWithUnionQuery.cpp:317
        const auto & query = query_ptr->as<ASTSelectWithUnionQuery &>();
        if (query.union_mode == SelectUnionMode::UNION_DISTINCT)
        {
            /// Add distinct transform
            DistinctStep::Settings distinct_settings(settings);

            /// UNION concatenates its branches' streams instead of merging them, so a preliminary
            /// DISTINCT runs in parallel and shrinks what the final single-stream DISTINCT must merge.
            if (preliminaryDistinctIsUseful(max_threads))
            {
                auto pre_distinct_step = std::make_unique<DistinctStep>(
                    query_plan.getCurrentHeader(),
                    distinct_settings,
                    0,
                    result_header->getNames(),
                    true);
                pre_distinct_step->setStepDescription("Preliminary DISTINCT");
                query_plan.addStep(std::move(pre_distinct_step));
            }

            auto distinct_step = std::make_unique<DistinctStep>(
                query_plan.getCurrentHeader(),
                std::move(distinct_settings),
                0,
                result_header->getNames(),
                false);

            query_plan.addStep(std::move(distinct_step));
        }

The shape of the plan is:

text
   branch 1 ─┐
   branch 2 ─┼─► UNION (interleaved streams) ─► Preliminary DISTINCT ─► Final DISTINCT ─► result
   branch 3 ─┘        (N streams)              (N streams, parallel)  (1 stream)

The preliminary DISTINCT is per-stream and runs in parallel. It is not required for correctness — the final DISTINCT deduplicates exactly — it exists to shrink the volume that the final one has to merge. Its defining property is that it is allowed to fail.

The preliminary DISTINCT can give up

A DISTINCT is a hash table. If the input turns out to be mostly unique, the hash table is pure overhead: it grows with every row and removes almost nothing. So the preliminary DISTINCT watches the survival rate and, if it is high, drops the table and passes the rest of the stream through:

cpp
// src/Processors/Transforms/DistinctTransform.cpp:230
    if (abandon_controller)
    {
        /// The rate is measured against the rows the transform received: the rows dropped as `NULL` keys
        /// (in the `skip_null_keys` mode, inside the filter) count as removed by the deduplication, so a
        /// stream that mostly consists of `NULL` keys keeps the transform even when the non-`NULL` part is
        /// unique - dropping the `NULL` rows is exactly the reduction the consumer benefits from.
        if (abandon_controller->update(num_rows, chunk.getNumRows(), distinct_set->getTotalByteCount()))
        {
            LOG_TRACE(getLogger("DistinctTransform"),
                "Switching DISTINCT to pass-through: input is mostly unique (retained keys: {}, set memory: {})",
                distinct_set->getTotalRowCount(), formatReadableSizeWithBinarySuffix(distinct_set->getTotalByteCount()));

            /// The new rows of the current chunk are still emitted (the following chunks flow
            /// through unfiltered).
            distinct_set.reset();
            ProfileEvents::increment(ProfileEvents::DistinctTransformsAbandonedDeduplication);
            return;
        }
    }

The thresholds are concrete:

cpp
// src/Processors/Transforms/DistinctTransform.h:30
    /// Number of chunks to observe before the rate is checked.
    static constexpr size_t OBSERVATION_CHUNK_COUNT = 5;

    /// The observation itself retains memory: until the first check, the hash table keeps every unique
    /// key seen, which for wide keys is chunk count * block size * key size per stream. One chunk of
    /// rows already gives a meaningful rate, so once the set is this large the check starts immediately
    /// instead of waiting out the chunk window. Integer keys stay on the full window (their set is
    /// about half this size at the fifth chunk).
    static constexpr size_t MAX_OBSERVATION_SET_BYTES = 16 * 1024 * 1024;

    /// Fraction of the observed rows that survived deduplication. Above this rate the removal is too
    /// small to help the consumer, while the hash table keeps growing with the unique rows.
    static constexpr double UNIQUE_RATE_THRESHOLD = 0.9;

Five chunks, or 16 MB of hash table, whichever comes first. Above a 90 % survival rate, the deduplication is abandoned.

This is only sound because a deduplicating consumer follows. The allow_abandoning flag is therefore only set for the preliminary step:

cpp
// src/Processors/QueryPlan/DistinctStep.cpp:290
        /// The preliminary deduplication is best-effort (a deduplicating consumer follows), so on
        /// mostly-unique input the transform may abandon it and free its hash table - unless a limit
        /// hint is set: an abandoned transform cannot count the distinct rows to stop the input early.
        const bool allow_abandoning = pre_distinct && build_settings.allow_preliminary_distinct_abandoning && limit_hint == 0;

Note the limit_hint == 0 condition: SELECT DISTINCT ... LIMIT n needs the count of distinct rows to know when to stop, so it may not abandon.

Memory pressure also releases it

The same step releases its set when the query’s memory usage exceeds a threshold, for the same reason — the final DISTINCT will catch anything the preliminary one misses:

cpp
// src/Processors/Transforms/DistinctTransform.cpp:187
    if (max_bytes_before_pass_through)
    {
        distinct_set->prepareForInsert(chunk);

        /// Preliminary hashing shares the query's remaining spill-threshold budget with the final
        /// transform and other operators.
        const UInt64 query_memory_usage = std::max<Int64>(0, getCurrentQueryMemoryUsage());
        const UInt64 available_memory = max_bytes_before_pass_through - std::min(max_bytes_before_pass_through, query_memory_usage);

        const size_t filtering_memory = distinct_set->estimateFilteringMemory(chunk);
        const size_t growth_memory = distinct_set->estimateGrowthMemory(chunk);
        if (filtering_memory > available_memory || growth_memory > available_memory - filtering_memory)
        {

Note that it predicts the memory: estimateGrowthMemory and estimateFilteringMemory are called before the insert, so the decision to give up is made before the allocation happens, not after.

What a DISTINCT actually hashes

The final DISTINCT hashes the whole row — all the columns, not a key. The key columns are chosen by calculateDistinctKeyColumnsPositions, which excludes constant columns:

cpp
// src/Processors/Transforms/DistinctSetFilter.cpp:27
ColumnNumbers calculateDistinctKeyColumnsPositions(const Block & header, const Names & columns)
{
    const size_t num_columns = columns.empty() ? header.columns() : columns.size();
    ColumnNumbers key_columns_pos;
    key_columns_pos.reserve(num_columns);
    for (size_t i = 0; i < num_columns; ++i)
    {
        const auto pos = columns.empty() ? i : header.getPositionByName(columns[i]);
        const auto & col = header.getByPosition(pos).column;
        if (col && !isColumnConst(*col))
            key_columns_pos.emplace_back(pos);
    }
    return key_columns_pos;
}

The hashing itself is the Set machinery — the same one IN uses — with one of several methods chosen from the key types. When no structured method fits, it falls back to hashing an unambiguously serialised tuple, and the representation of the key changes accordingly:

cpp
// src/Processors/Transforms/DistinctSetFilter.cpp:406
DistinctKeyRepresentation DistinctSetFilter::getKeyRepresentation() const
{
    chassert(!data->empty());
    return data->type == SetVariants::Type::hashed ? DistinctKeyRepresentation::Hash128 : DistinctKeyRepresentation::Columns;
}

The LowCardinality shortcut

There is one clever optimisation here that is worth its own section. If the key is a single LowCardinality column, the distinctness of a row is determined by its dictionary index, not by its value. So instead of hashing values, the transform keeps a bitmap of “which dictionary indices have I seen” and compares bits:

cpp
// src/Processors/Transforms/DistinctSetFilter.cpp:134
std::pair<IColumn::Filter, size_t> DistinctLowCardinalityFilter::buildMask(const ColumnLowCardinality & column, size_t num_rows)
{
    const auto & dictionary = column.getDictionary();
    const auto dict_size = dictionary.size();

    const DictionariesState::LCDictionaryKey dict_key{dictionary.getHash(), dict_size};

    /// The dictionary identity includes its size, so each bitmap is allocated once and never resized.
    auto [it, inserted] = dictionaries_state->lc_dict_states.try_emplace(dict_key, dict_size);
    auto & state = it->second;

The mask it produces is a pre-filter: rows whose dictionary index has already been seen cannot be new distinct values, so they are removed before the hash table is even consulted. And if every index of the dictionary has been seen, the chunk cannot contain any new value at all:

cpp
// src/Processors/Transforms/DistinctSetFilter.cpp:149
    /// If we've already seen all dictionary indices for this dictionary, then no row in this chunk
    /// (and also other chunks with the same dictionary) can produce a new distinct value.
    if (state.seen_count == dict_size)
        return {{}, 0};

The controller decides whether the shortcut is worth it at all, by measuring how often a new dictionary index appears:

cpp
// src/Processors/Transforms/DistinctSetFilter.cpp:42
void LCOptimizationController::update(size_t num_rows, size_t new_indices_in_chunk)
{
    if (state != State::Observing)
        return;

    ++chunks_observed;
    rows_observed += num_rows;
    new_indices_observed += new_indices_in_chunk;

    if (chunks_observed >= OBSERVATION_CHUNK_COUNT)
    {
        double new_index_rate = static_cast<double>(new_indices_observed) / static_cast<double>(rows_observed);

        /// Disable when the mask is almost a no-op: nearly every row introduces
        /// a new dictionary index, so the bitmap bookkeeping is pure overhead.
        if (new_index_rate >= NEW_INDEX_RATE_THRESHOLD)
            state = State::Disabled;
        else
            state = State::Enabled;
    }
}

If nearly every row has a new dictionary index, the bitmap is a no-op and the bookkeeping is pure overhead — so it is switched off.

INTERSECT and EXCEPT

INTERSECT and EXCEPT are not implemented as a UNION plus a filter. They have their own plan step and their own processor.

The step unites the two branches into one pipe and then attaches a transform with two input ports:

cpp
// src/Processors/QueryPlan/IntersectOrExceptStep.cpp:125
        /// For the case of union.
        cur_pipeline->addTransform(std::make_shared<ResizeProcessor>(getOutputHeader(), cur_pipeline->getNumStreams(), 1));
    }

    /// Zero means the step was deserialized on a worker; use the executing server's own setting.
    size_t new_max_threads = max_threads ? max_threads : settings.max_threads;
    *pipeline = QueryPipelineBuilder::unitePipelines(std::move(pipelines), new_max_threads, &processors);
    auto transform = std::make_shared<IntersectOrExceptTransform>(getOutputHeader(), current_operator);
    processors.push_back(transform);
    pipeline->addTransform(std::move(transform));

IntersectOrExceptTransform has a three-stage state machine: read the left side, read the right side, then filter the left side.

cpp
// src/Processors/Transforms/IntersectOrExceptTransform.cpp:129
void IntersectOrExceptTransform::work()
{
    if (stage == Stage::ReadLeftInput)
    {
        if (current_input_chunk.hasRows())
        {
            left_input_chunk = std::move(current_input_chunk);
            has_left_input_chunk = true;
            stage = Stage::ReadRightInput;
        }
    }
    else if (stage == Stage::ReadRightInput)
    {
        has_right_input_rows |= current_input_chunk.hasRows();
        accumulate(std::move(current_input_chunk));
    }
    else
    {
        filter(current_input_chunk);
        current_output_chunk = std::move(current_input_chunk);
    }

    has_input = false;
}

The right side is accumulated into a hash set — the same SetVariants machinery, keyed on the whole row:

cpp
// src/Processors/Transforms/IntersectOrExceptTransform.cpp:209
template <typename Method>
void IntersectOrExceptTransform::addToSet(Method & method, const ColumnRawPtrs & columns, size_t rows, SetVariants & variants) const
{
    chassert(!isAllOperator(), "addToSet must only be used for DISTINCT operators");

    typename Method::State state(columns, key_sizes, nullptr);

    for (size_t i = 0; i < rows; ++i)
    {
        if ((i & 0xFFF) == 0)
        {
            if (i > 0) [[unlikely]]
                FailPointInjection::pauseFailPoint(FailPoints::intersect_or_except_transform_pause);
            if (isCancelled())
                return;
        }

        state.emplaceKey(method.data, i, variants.string_pool);
    }
}

and then the left side is streamed through a membership filter:

cpp
// src/Processors/Transforms/IntersectOrExceptTransform.cpp:231
template <typename Method>
size_t IntersectOrExceptTransform::buildFilter(
    Method & method, const ColumnRawPtrs & columns, IColumn::Filter & filter, size_t rows, SetVariants & variants) const
{
    /// ALL operators use the counting path, so only DISTINCT operators reach here; the branch
    /// below therefore only needs to distinguish EXCEPT_DISTINCT from INTERSECT_DISTINCT.
    chassert(!isAllOperator(), "buildFilter must only be used for DISTINCT operators");

    typename Method::State state(columns, key_sizes, nullptr);
    size_t new_rows_num = 0;

    for (size_t i = 0; i < rows; ++i)
    {
        if ((i & 0xFFF) == 0 && isCancelled())
        {
            std::fill(filter.begin() + i, filter.end(), 0);
            return new_rows_num;
        }

        auto find_result = state.findKey(method.data, i, variants.string_pool);
        filter[i] = (current_operator == ASTSelectIntersectExceptQuery::Operator::EXCEPT_DISTINCT)
            ? !find_result.isFound()
            : find_result.isFound();

So INTERSECT DISTINCT keeps the left rows whose key is in the right set, and EXCEPT DISTINCT keeps the ones whose key is not.

The ALL variants need counts

EXCEPT ALL and INTERSECT ALL are multiset operations: A EXCEPT ALL B removes one occurrence of each row of B from A. So the set has to become a counting map, and the filter decrements the count as it goes:

cpp
// src/Processors/Transforms/IntersectOrExceptTransform.cpp:176
template <typename Method>
size_t IntersectOrExceptTransform::filterWithCounts(
    Method & method, const ColumnRawPtrs & columns, IColumn::Filter & filter, size_t rows, CountingSetVariants & variants) const
{
    typename Method::State state(columns, key_sizes, nullptr);
    const bool is_except = (current_operator == Operator::EXCEPT_ALL);
    size_t new_rows_num = 0;

    for (size_t i = 0; i < rows; ++i)
    {
        if ((i & 0xFFF) == 0 && isCancelled())
        {
            std::fill(filter.begin() + i, filter.end(), 0);
            return new_rows_num;
        }

        auto find_result = state.findKey(method.data, i, variants.string_pool);

        /// A remaining right-side occurrence of this row.
        bool matched = find_result.isFound() && find_result.getMapped() > 0;
        if (matched)
            --find_result.getMapped();

        /// EXCEPT ALL keeps unmatched rows; INTERSECT ALL keeps matched rows.
        filter[i] = matched != is_except;

The count is the mapped value of the hash table; --find_result.getMapped() consumes one occurrence. A row whose count has reached zero no longer matches, which is exactly the multiset semantics.

The DISTINCT variants get a second DISTINCT

INTERSECT DISTINCT and EXCEPT DISTINCT produce a set, but the transform above only removes duplicates of the right side. The left side may still contain duplicates of a row that survived. So a final DISTINCT is added:

cpp
// src/Interpreters/InterpreterSelectIntersectExceptQuery.cpp:163
    const auto & query = query_ptr->as<ASTSelectIntersectExceptQuery &>();
    if (query.final_operator == ASTSelectIntersectExceptQuery::Operator::INTERSECT_DISTINCT
        || query.final_operator == ASTSelectIntersectExceptQuery::Operator::EXCEPT_DISTINCT)
    {
        /// Add distinct transform
        auto distinct_step = std::make_unique<DistinctStep>(
            query_plan.getCurrentHeader(),
            DistinctStep::Settings(settings),
            0,
            result_header->getNames(),
            false);

        query_plan.addStep(std::move(distinct_step));
    }

Note the false in the last argument — that is pre_distinct, and for this final DISTINCT it is false, so it is exact and cannot abandon itself.

Trade-offs

What the design buys:

UNION ALL is free. No processor, no copy, no buffer. The branches keep their own parallelism.
The preliminary DISTINCT is a real optimisation and it is safe, because the final DISTINCT is exact. It turns a single-threaded hash table over the whole union into a per-stream hash table over a shrunken stream.
The abandon heuristic removes the pathological case. A DISTINCT over a mostly-unique column is a hash table that removes nothing and costs everything; the preliminary step notices and gets out of the way.
INTERSECT/EXCEPT stream the left side. The right side is materialised into a set, but the left side is not: it is filtered as it arrives.

What it costs:

UNION ALL is not a concatenation. The rows come out interleaved, and the plan explicitly gives up the right to narrow the stream count. An ORDER BY after a UNION ALL is a real sort of the whole result.
UNION DISTINCT hashes every row twice in the general case: once in the preliminary pass, once in the final. The preliminary pass exists to make the second pass cheaper, and it is only worth it when there are enough threads (preliminaryDistinctIsUseful returns max_threads > 1).
The abandon heuristic has a memory cost of its own. It needs five chunks (or 16 MB) of hash table before it can make the call, and the hash table it is trying to avoid is exactly what it has to build first.
INTERSECT/EXCEPT materialise the right side. For a large right side this is the same memory problem a hash join has, and there is no spilling variant.
INTERSECT/EXCEPT read the right side first, which means the left side’s first block waits for the whole right side to arrive.

Wrapping up

The one thing to remember: UNION ALL is a pipe merge, UNION DISTINCT is a pipe merge plus a hash set, and INTERSECT/EXCEPT is a hash set of one side plus a streaming filter of the other. Only the second of those has a preliminary pass, and only the preliminary pass is allowed to be wrong.

The part people get wrong: UNION ALL does not preserve the order of the branches, and it does not even preserve the order within a branch if the branch has several streams. If you need ordered output, the ORDER BY has to come after the union.

What to go do about it: for a UNION DISTINCT over many branches, look at the pipeline in EXPLAIN PIPELINE: if there is a “Preliminary DISTINCT” step, the final one has less work to do. If the preliminary one is missing, the union had a single thread, and the final DISTINCT is doing all the work on one thread.