Opensource internals

How ReplacingMergeTree Collapses Duplicates and What SELECT FINAL Really Does

20 min read Jay Hubbert

How ReplacingMergeTree picks a winning row per sorting key, how the version column breaks ties, and how SELECT FINAL rebuilds that decision at query time.

TL;DR: A ReplacingMergeTree merge keeps one row per sorting key: the one with the highest version, or the last one inserted if the versions tie. A delete is a row with is_deleted = 1 and a high version. SELECT FINAL does not look at the merged state — it re-runs the same collapse at query time over every part, which is why it is expensive.

The problem: an update is a delete plus an insert

ClickHouse’s storage model has no UPDATE. A MergeTree part is immutable; the only way to change a row is to write a new row and let a merge throw the old one away.

For a slowly-changing dimension — a user’s profile, a product’s price, a device’s last-seen timestamp — that is the natural model. The user wants:

For each key, show me the newest version of the row.

ReplacingMergeTree is the engine that does this. Its merge algorithm is simple to state and fiddly to implement:

  • Group rows by sorting key.
  • Within a group, keep exactly one.
  • Which one? The one with the highest version. If versions tie, the last inserted.

The engine declaration

ReplacingMergeTree is a MergeTree with two extra optional parameters: a version column and an is_deleted column.

cpp 
// src/Storages/MergeTree/MergeTreeData.h:478
    struct MergingParams
    {
        /// Merging mode. See above.
        enum Mode
        {
            Ordinary            = 0,    /// Enum values are saved. Do not change them.
            Collapsing          = 1,
            Summing             = 2,
            Aggregating         = 3,
            Replacing           = 5,
            Graphite            = 6,
            VersionedCollapsing = 7,
            Coalescing          = 8,
        };

        Mode mode{};

        /// For Collapsing and VersionedCollapsing mode.
        String sign_column;

        /// For Replacing mode. Can be empty for Replacing.
        String is_deleted_column;

        /// For Summing mode. If empty - columns_to_sum is determined automatically.
        Names columns_to_sum;

        /// For Replacing and VersionedCollapsing mode. Can be empty for Replacing.
        String version_column;

And the parameters are parsed positionally from the engine arguments, which is a small parsing puzzle — the last argument may be the version column or the is_deleted column, and you cannot tell them apart without knowing how many there are:

cpp 
// src/Storages/MergeTree/registerStorageMergeTree.cpp:633
    else if (merging_params.mode == MergeTreeData::MergingParams::Replacing)
    {
        // if there is args and number of optional parameter is higher than 1
        // is_deleted is not allowed with the 'allow_deprecated_syntax_for_merge_tree' settings
        if (arg_cnt - arg_num == 2 && !engine_args[arg_cnt - 1]->as<ASTLiteral>() && is_extended_storage_def)
        {
            if (!tryGetIdentifierNameInto(engine_args[arg_cnt - 1], merging_params.is_deleted_column))
                throw Exception(ErrorCodes::BAD_ARGUMENTS, "is_deleted column name must be an identifier {}", verbose_help_message);
            --arg_cnt;
        }

        /// If the last element is not index_granularity or replica_name (a literal), then this is the name of the version column.
        if (arg_cnt && !engine_args[arg_cnt - 1]->as<ASTLiteral>())
        {
            if (!tryGetIdentifierNameInto(engine_args[arg_cnt - 1], merging_params.version_column))
                throw Exception(ErrorCodes::BAD_ARGUMENTS, "Version column name must be an identifier {}", verbose_help_message);
            --arg_cnt;
        }
    }

Note the order: is_deleted is peeled off first (when there are two identifier arguments), then version. But the documented syntax is ReplacingMergeTree(ver, is_deleted) — so the last argument is is_deleted and the one before it is the version. The code peels is_deleted from the end and then version from the new end, which is consistent.

The types are checked at creation time:

cpp 
// src/Storages/MergeTree/MergeTreeData.cpp:2410
    if (mode == MergingParams::Replacing)
    {
        if (!version_column.empty() && version_column == is_deleted_column)
            throw Exception(ErrorCodes::BAD_ARGUMENTS, "The version and is_deleted column cannot be the same column ({})", version_column);

        check_is_deleted_column(true, "ReplacingMergeTree");
        check_version_column(true, "ReplacingMergeTree");
    }

is_deleted must be UInt8 and the version column must satisfy canBeUsedAsVersion() — an integer or a date type. is_deleted also requires a version column:

cpp 
// src/Storages/MergeTree/MergeTreeData.cpp:2316
    /// Check that if the is_deleted column is needed, it exists and is of type UInt8. If exist, version column must be defined too but version checks are not done here.
    auto check_is_deleted_column = [this, & columns](bool is_optional, const std::string & storage)
    {
        if (is_deleted_column.empty())
        {
            if (is_optional)
                return;

            throw Exception(ErrorCodes::LOGICAL_ERROR, "`is_deleted` ({}) column for storage {} is empty", is_deleted_column, storage);
        }

        if (version_column.empty() && !is_optional)
            throw Exception(ErrorCodes::LOGICAL_ERROR, "Version column ({}) for storage {} is empty while is_deleted ({}) is not.",
                            version_column, storage, is_deleted_column);

That is a semantic requirement, not a syntactic one: a delete row without a version cannot be ordered against the rows it is supposed to replace.

A typical table looks like this:

sql 
CREATE TABLE user_profile
(
    user_id     UInt64,
    name        String,
    email       String,
    updated_at  DateTime,   -- the version column
    is_deleted  UInt8       -- the tombstone flag
)
ENGINE = ReplacingMergeTree(updated_at, is_deleted)
ORDER BY user_id;

An update is then two inserts: the new row with a higher updated_at, and — if the user was deleted — a row with is_deleted = 1 and an even higher updated_at.

The merge algorithm

The merge is implemented as a merging algorithm over the sorted streams of the parts being merged. The heart of it is the version comparison:

cpp 
// src/Processors/Merges/Algorithms/ReplacingSortedAlgorithm.cpp:269
        bool replace_with_current_row = false;
        if (selected_row.empty())
        {
            replace_with_current_row = true;
        }
        else
        {
            /// Three-way comparison of the current row's version with the selected row's version.
            /// Without a version column all rows count as having equal versions, so the selection
            /// falls through to the physical-order rule below.
            int version_cmp = version_column_number == -1 ? 0
                : current->all_columns[version_column_number]->compareAt(
                    current->getRow(), selected_row.row_num,
                    *(*selected_row.all_columns)[version_column_number],
                    /* nan_direction_hint = */ 1);

            if (version_cmp > 0)
            {
                replace_with_current_row = true;
            }
            else if (version_cmp == 0)
            {
                /// Rows with equal versions are selected by their physical order: the row written last wins.
                /// The queue emits rows with equal sort key ordered by source index, i.e. by data part
                /// (parts are ordered from the oldest to the newest one), and within one source in the reading order.
                /// In the direct reading order the current row is always "newer", so it replaces the selected one
                /// (a non-strict comparison in terms of the version). In the reverse reading order rows within
                /// one source arrive backwards, so the current row replaces the selected one only when
                /// it comes from a newer data part.
                chassert(current_row.source_stream_index >= selected_row.source_stream_index);
                replace_with_current_row = !read_in_reverse || current_row.source_stream_index > selected_row.source_stream_index;
            }
        }

Three things to unpack.

The comparison is three-way, not “greater than”. compareAt returns -1, 0 or +1, and the tie case is handled separately. A naive if (current_version > selected_version) would keep the first row on a tie; the three-way form lets the tie fall through to a physical-order rule.

The tie-break is physical order, and the direction matters. The queue emits rows with equal sort keys ordered by source index, and source index 0 is the oldest part. So “the current row replaces the selected one” means “the newer part wins”, which is the intuitive behaviour. But in reverse reading order, rows within one source arrive backwards, so the naive rule would pick the oldest row of the newest part. The read_in_reverse flag flips the comparison accordingly: within one part, the first row encountered in reverse order wins.

The nan_direction_hint = 1 matters. The version column’s type only has to satisfy canBeUsedAsVersion(), and DataTypeNumber<T> — which includes Float32 and Float64 — says yes:

cpp 
// src/DataTypes/DataTypesNumber.h:26
    bool canBeUsedAsVersion() const override { return true; }

So a float version column is legal, and float versions can be NaN. The nan_direction_hint tells compareAt how to order NaN against other values, so that a NaN version does not produce an inconsistent ordering between the merge and the query path.

Now, the delete flag. A row with is_deleted = 1 is a tombstone, and with cleanup enabled it is dropped rather than kept:

cpp 
// src/Processors/Merges/Algorithms/ReplacingSortedAlgorithm.cpp:89
/// True when the group's winning row must not reach the output: a `CLEANUP` tombstone, or a row
/// the filter rejects.
bool ReplacingSortedAlgorithm::isSelectedRowSkipped() const
{
    if (cleanup && is_deleted_column_number != -1
        && assert_cast<const ColumnUInt8 &>(*(*selected_row.all_columns)[is_deleted_column_number]).getData()[selected_row.row_num])
        return true;

    return isRowFiltered(selected_row);
}

Note cleanup &&. Without cleanup, the tombstone wins the group and is emitted as the surviving row — the row is still there, marked deleted. This is the documented default behaviour, and it is deliberate:

cpp 
// src/Storages/MergeTree/registerStorageMergeTree.cpp:3166
By default, ClickHouse will keep the last row for a key even if that row is a delete row. This is so that any future rows with lower versions can
be safely inserted and the delete row will still be applied.

The reasoning is about out-of-order arrival. If a delete with version 100 arrives, and later a row with version 50 arrives (a late-arriving update), the delete must still win — otherwise the resurrect would be applied. Keeping the tombstone preserves the ability to re-apply it. OPTIMIZE ... FINAL CLEANUP (or the allow_experimental_replacing_merge_with_cleanup setting) is what actually removes the tombstone rows.

The skip-run optimization

The merge algorithm has a fast path that is worth understanding, because it explains a lot of ClickHouse’s performance characteristics.

If there is no version column, no is_deleted column, and no row-source bookkeeping, then processing a row has no effect other than replacing selected_row with it. So a whole run of equal keys can be skipped — jump straight to the last row of the run:

cpp 
// src/Processors/Merges/Algorithms/ReplacingSortedAlgorithm.cpp:56
    /// With a version or an is_deleted column every row of a run must be examined, and row
    /// sources for a vertical merge must be recorded per row. Without them the only effect of
    /// processing a row is replacing `selected_row` with it, so runs can be fast-forwarded.
    /// In the reverse reading order the first row of a run within a source wins instead of the
    /// last one, so the fast-forward to the last row of the run does not apply either.
    can_skip_to_run_end = version_column_number == -1 && is_deleted_column_number == -1
        && out_row_sources_buf == nullptr && !enable_vertical_final && !read_in_reverse;
    uses_runs_of_equal_keys = can_skip_to_run_end;

And the actual jump:

cpp 
// src/Processors/Merges/Algorithms/ReplacingSortedAlgorithm.cpp:310
        /// All rows of one batch come consecutively from the same cursor. When only the last
        /// row of a run of equal keys is kept and processing a row has no other effects, jump
        /// to the last row of the run of the current key within the batch: the intermediate
        /// rows would each merely replace `selected_row` with the next one. Sources with a
        /// non-zero part level are excluded to keep the exact behavior of
        /// `rowsHaveDifferentSortColumns`, which does not compare rows of such sources at all.
        if (skip_runs_of_equal_keys && current_batch_size > 1 && !current->permutation
            && sources_origin_merge_tree_part_level[current->order] == 0)
        {
            size_t run_begin = current->getPos();
            size_t run_bound = run_begin + current_batch_size;

            /// The last row of the cursor may need to be skipped, leave it to the per-row check.
            if (run_bound == current->getSize() && skipLastRowFor(current->order))
                --run_bound;

            if (run_begin + 1 < run_bound)
            {
                auto & key_runs = source_key_runs[current->order];
                if (key_runs.end != run_bound)
                {
                    key_runs.runs.reset(current->desc.size());
                    key_runs.end = run_bound;
                }
                size_t run_end = getEqualRangeEndAssumeSorted(key_runs.runs, current->sort_columns, current->desc, run_begin, run_bound);
                if (run_end > run_begin + 1)
                {
                    /// Jump to the last row of the run; the loop processes it as usual.
                    queue.next(run_end - 1 - run_begin);
                    continue;
                }
            }
        }

The sources_origin_merge_tree_part_level[current->order] == 0 guard is subtle and important: a part at level > 0 was produced by a merge, so it already has one row per key — there are no duplicate runs in it to skip, and skipping them would change the result of rowsHaveDifferentSortColumns.

The skip_runs_of_equal_keys flag has a further precondition:

cpp 
// src/Processors/Merges/Algorithms/ReplacingSortedAlgorithm.cpp:67
void ReplacingSortedAlgorithm::initialize(Inputs inputs)
{
    IMergingAlgorithmWithSharedChunks::initialize(std::move(inputs));

    /// Skipping runs needs the queue to actually detect batches. A batch longer than one row is
    /// not evidence of that on its own: a queue with one cursor left always reports its whole
    /// remainder as one batch, since there is nothing to compare it against. Without this
    /// condition the probe for the end of a run would run on every row of a single-input merge -
    /// which is every `INSERT` into a `ReplacingMergeTree` with `optimize_on_insert` (on by
    /// default), where `MergeTreeDataWriter::mergeBlock` merges one already sorted block. When
    /// that block holds no runs of equal keys, the detection is disabled and the merge has to
    /// cost exactly what it costs with the plain heap.
    skip_runs_of_equal_keys = can_skip_to_run_end && batch_detection_enabled;
    source_key_runs.assign(sources.size(), {});
}

This is a good example of the kind of detail that makes a fast path fast: the probe for “where does this run end” is itself O(run length) in the worst case, so it must be disabled when there is nothing to gain. A single-input merge — which is what optimize_on_insert does — would otherwise pay the probe cost on every row.

What SELECT FINAL actually does

Here is the crucial point, and it is the one most users get wrong.

SELECT FINAL does not read the merged state of the table. The parts on disk are exactly as they were written. FINAL re-runs the collapse at query time, over the raw parts, and only then applies the predicate.

The pipe construction makes it explicit — FINAL reads all parts and inserts a merging transform:

cpp 
// src/Processors/QueryPlan/MergeTreeFinalMerge.cpp:84
            case MergeTreeData::MergingParams::Replacing:
                return std::make_shared<ReplacingSortedTransform>(header, num_outputs,
                            sort_description, merging_params.is_deleted_column, merging_params.version_column, max_block_size_rows, /*max_block_size_bytes=*/0, /*max_dynamic_subcolumns*/std::nullopt, /*out_row_sources_buf_*/ nullptr, /*use_average_block_sizes*/ false, /*cleanup*/ !merging_params.is_deleted_column.empty(), enable_vertical_final, read_in_reverse);

and the full-merge pipe reads everything in order and merges it:

cpp 
// src/Processors/QueryPlan/MergeTreeFinalMerge.cpp:164
Pipe buildFullFinalMergePipe(
    const StorageMetadataPtr & metadata_snapshot,
    MergeTreeData::MergingParams merging_params,
    size_t max_block_size_rows,
    bool enable_vertical_final,
    ContextPtr context,
    std::optional<ActionsDAG> & out_projection,
    const std::function<Pipe()> & read_all_parts_in_order)
{
    auto sorting_expr = metadata_snapshot->getSortingKey().expression;
    auto sort_description = buildFinalSortDescription(metadata_snapshot, context->getSettingsRef());

    Pipe pipe = read_all_parts_in_order();
    if (pipe.empty())
        return pipe;

    pipe.addSimpleTransform([sorting_expr](const SharedHeader & header)
                            { return std::make_shared<ExpressionTransform>(header, sorting_expr); });

So FINAL costs a full merge of every part at query time, with all the memory and CPU that implies, and it happens on every query. OPTIMIZE TABLE ... FINAL pays the same cost once, in the background, and writes the result down.

There is a special case for is_deleted, though. When a part range does not intersect any other part range — no key can have a duplicate — the merge is skipped, and the is_deleted filter has to be applied by hand, because the merging algorithm would have applied it:

cpp 
// src/Processors/QueryPlan/MergeTreeFinalMerge.cpp:279
Pipe readNonIntersectingFinalWithEngineFilter(
    const MergeTreeData::MergingParams & merging_params,
    const Names & origin_column_names,
    ContextPtr context,
    const std::function<Pipe(const Names & columns)> & read)
{
    /// `Collapsing` does not expose unmatched negative-sign rows on FINAL, and `Replacing` with an is-deleted
    /// column hides deleted rows; a non-intersecting range skips the merge, so add that filter here. Other
    /// engines drop nothing on FINAL beyond the deduplication a single part already did.
    if (merging_params.mode == MergeTreeData::MergingParams::Collapsing)
    {
        auto columns_with_sign = origin_column_names;
        if (std::ranges::find(columns_with_sign, merging_params.sign_column) == columns_with_sign.end())
            columns_with_sign.push_back(merging_params.sign_column);
        Pipe pipe = read(columns_with_sign);
        auto [expression, filter_name] = createExpressionForPositiveSign(merging_params.sign_column, pipe.getHeader(), context);
        pipe.addSimpleTransform([&](const SharedHeader & header)
            { return std::make_shared<FilterTransform>(header, expression, filter_name, true); });
        return pipe;
    }
    if (!merging_params.is_deleted_column.empty())
    {
        auto columns_with_is_deleted = origin_column_names;
        if (std::ranges::find(columns_with_is_deleted, merging_params.is_deleted_column) == columns_with_is_deleted.end())
            columns_with_is_deleted.push_back(merging_params.is_deleted_column);
        Pipe pipe = read(columns_with_is_deleted);
        auto [expression, filter_name] = createExpressionForIsDeleted(merging_params.is_deleted_column, pipe.getHeader(), context);
        pipe.addSimpleTransform([&](const SharedHeader & header)
            { return std::make_shared<FilterTransform>(header, expression, filter_name, true); });
        return pipe;
    }
    return read(origin_column_names);
}

is_deleted = 0 is the filter. A tombstone is hidden on FINAL — which is the opposite of the merge’s default behaviour, where the tombstone wins the group and is emitted.

Vertical FINAL

The default FINAL implementation copies the surviving row’s data into the output. enable_vertical_final avoids that: instead of merging the data, the algorithm records which row indices survived, and a later transform picks them out of the original chunks:

cpp 
// src/Processors/Merges/Algorithms/ReplacingSortedAlgorithm.cpp:127
void ReplacingSortedAlgorithm::insertRowImpl()
{
    flushCurrentRowSources(/*keep_selected_row=*/ true);

    if (enable_vertical_final)
    {
        /// We just record the position to be selected in the chunk
        if (!selected_row.owned_chunk->replace_final_selection)
            selected_row.owned_chunk->replace_final_selection = ColumnUInt64::create();

        selected_row.owned_chunk->replace_final_selection->insert(selected_row.row_num);

        /// This is the last row we can select from `selected_row.owned_chunk`, keep it to emit later
        if (selected_row.current_cursor == nullptr)
            to_be_emitted.push(std::move(selected_row.owned_chunk));
    }
    else
    {
        merged_data->insertRow(*selected_row.all_columns, selected_row.row_num, selected_row.owned_chunk->getNumRows());
    }

    selected_row.clear();
}

The chunks carry a side-channel column of surviving row indices, and a SelectByIndicesTransform materializes the result at the end:

cpp 
// src/Processors/QueryPlan/MergeTreeFinalMerge.cpp:106
    pipe.addTransform(get_merging_processor());
    if (enable_vertical_final)
        pipe.addSimpleTransform([](const SharedHeader & header_)
                                { return std::make_shared<SelectByIndicesTransform>(header_); });

The win: the merge does not have to copy the surviving rows’ column data. The data stays in the original chunks, and only an index is recorded. For a table where most rows survive, that saves a full copy of the table.

The cost: the can_skip_to_run_end fast path is disabled, because a vertical final has to record the position of every surviving row, and the row-source bookkeeping has to be per-row.

The lazy FINAL: replacing a merge with an aggregation

There is a much more aggressive optimization, and it is the most interesting part of this story. It is enabled when the query has a WHERE clause with a selective primary-key condition, and the engine can prove that only a small set of keys can possibly have duplicates.

Instead of merging all the parts, the engine:

  1. Reads only the rows matching the primary-key condition, projecting just the primary key columns.
  2. Builds a Set of those keys.
  3. Uses that set to figure out which parts actually need to be merged.
  4. For the parts that do, computes the winner with an argMax aggregation rather than a merge.

The tiebreaker is the interesting bit. The winner of a group is “the row with the highest version; on a tie, the one inserted last”. Insert order is captured by a global row index, and the two are packed into a single comparable value:

cpp 
// src/Processors/Sources/LazyReadReplacingFinalSource.cpp:143
    /// When there's a version column, compute a tiebreaker for argMax so that
    /// equal versions resolve to the last inserted row (highest __global_row_index).
    /// For version types ≤ 64 bits, pack (version, __global_row_index) into UInt128.
    /// For wider types, use tuple(version, __global_row_index).
    /// When there's no version column, just use max(__global_row_index).
    bool has_version = !merging_params.version_column.empty();
    static constexpr auto tiebreaker_column_name = "__lazy_final_tiebreaker";

The packing, for a 64-bit-or-narrower version:

cpp 
// src/Processors/Sources/LazyReadReplacingFinalSource.cpp:184
            const auto * version_128 = &dag.addFunction(to_uint128, {version_node}, {});
            const auto * row_index_128 = &dag.addFunction(to_uint128, {row_index_node}, {});
            auto shift_type = std::make_shared<DataTypeUInt8>();
            auto shift_column = shift_type->createColumnConst(0, Field(UInt8(64)));
            const auto * shift_amount = &dag.addColumn(std::move(shift_column), std::move(shift_type), "__shift_64");
            const auto * shifted = &dag.addFunction(bit_shift_left, {version_128, shift_amount}, {});
            const auto * tiebreaker = &dag.addFunction(plus_func, {shifted, row_index_128}, {});
            tiebreaker = &dag.addAlias(*tiebreaker, tiebreaker_column_name);

(version << 64) + row_index in a UInt128. The version occupies the high 64 bits, the row index the low 64, so a plain unsigned comparison of the tiebreaker is exactly “compare versions, then compare insertion order”.

Signed versions need a fixup, because the sign bit would otherwise make every negative version look larger than every positive one:

cpp 
// src/Processors/Sources/LazyReadReplacingFinalSource.cpp:168
            /// For signed types (and Date32 which is Int32 internally),
            /// widen to Int64 first, then flip the sign bit
            /// to convert signed order to unsigned order.
            if (which.isNativeInt() || which.isDate32())
            {
                auto to_int64 = FunctionFactory::instance().get("toInt64", nullptr);
                auto reinterpret_func = FunctionFactory::instance().get("reinterpretAsUInt64", nullptr);
                auto bitxor_func = FunctionFactory::instance().get("bitXor", nullptr);
                auto sign_bit_type = std::make_shared<DataTypeUInt64>();
                auto sign_bit_column = sign_bit_type->createColumnConst(0, Field(UInt64(1) << 63));
                const auto * sign_bit_node = &dag.addColumn(std::move(sign_bit_column), std::move(sign_bit_type), "__sign_bit");
                const auto * version_int64 = &dag.addFunction(to_int64, {version_node}, {});
                const auto * version_uint64 = &dag.addFunction(reinterpret_func, {version_int64}, {});
                version_node = &dag.addFunction(bitxor_func, {version_uint64, sign_bit_node}, {});
            }

x ^ (1 << 63) is the standard “flip the sign bit” transform that maps signed order onto unsigned order. For versions wider than 64 bits, the tiebreaker falls back to a tuple:

cpp 
// src/Processors/Sources/LazyReadReplacingFinalSource.cpp:194
        else
        {
            auto tuple_func = FunctionFactory::instance().get("tuple", query_context);
            const auto * tiebreaker = &dag.addFunction(tuple_func, {version_node, row_index_node}, {});
            tiebreaker = &dag.addAlias(*tiebreaker, tiebreaker_column_name);
            dag.getOutputs().push_back(tiebreaker);
        }

And the aggregation itself:

cpp 
// src/Processors/Sources/LazyReadReplacingFinalSource.cpp:215
        if (has_version)
        {
            auto tiebreaker_type = header->getByName(tiebreaker_column_name).type;
            argmax_row_index_name = fmt::format("argMax(__global_row_index, {})", tiebreaker_column_name);

            /// argMax(__global_row_index, __tiebreaker)
            AggregateDescription desc;
            desc.function = AggregateFunctionFactory::instance().get(
                "argMax", NullsAction::EMPTY, {uint64_type, tiebreaker_type}, {}, properties);
            desc.argument_names = {"__global_row_index", tiebreaker_column_name};
            desc.column_name = argmax_row_index_name;
            aggregates.push_back(std::move(desc));

            /// argMax(is_deleted, __tiebreaker) if is_deleted column exists
            if (!merging_params.is_deleted_column.empty())
            {
                argmax_is_deleted_name = fmt::format("argMax({}, {})", merging_params.is_deleted_column, tiebreaker_column_name);
                auto is_deleted_type = header->getByName(merging_params.is_deleted_column).type;

Two argMax aggregates: one for the winning row’s global index, one for the winning row’s is_deleted flag. Both use the same tiebreaker, so they agree on which row won. The result is a grouped aggregation over the primary key with an argMax — which is a hash-table aggregate, not a sort-merge.

The whole optimization is gated on being able to prove the merge is unnecessary for most parts. The optimizeLazyFinal pass splits the parts into “intersecting” (key ranges overlap, so a merge is needed) and “non-intersecting” (no overlap, so no duplicate is possible):

cpp 
// src/Processors/QueryPlan/Optimizations/optimizeLazyFinal.cpp:489
    /// Split parts into non-intersecting (unique key ranges, no FINAL needed) and
    /// intersecting (overlapping, need FINAL). This avoids running the expensive
    /// aggregation-based FINAL on parts that have no duplicates.
    /// When all parts are non-intersecting, replaceNodeWithPlan is called inside
    /// and fully_replaced is set — in that case we're done.
    auto split_result = trySplitNonIntersectingParts(
        reading_step, analyzed_result, filter_step, read_node, query_plan, /*allow_partial_split=*/ !stops_reading_early);

The gate is a ratio — the optimization only pays off if enough of the table is pruned:

cpp 
// src/Processors/QueryPlan/Optimizations/optimizeLazyFinal.cpp:509
    /// Use parts from the analyzed result (possibly narrowed to intersecting-only by the split).
    /// These are PK-filtered parts with narrowed mark ranges from selectRangesToRead.
    auto parts_for_set = std::make_shared<RangesInDataParts>(
        analyzed_result ? analyzed_result->parts_with_ranges : reading_step->getParts());

The set is built from the primary key columns only — the PK is a prefix of the sorting key, so the remaining sorting key columns are useless for deciding which keys have duplicates:

cpp 
// src/Processors/QueryPlan/Optimizations/optimizeLazyFinal.cpp:514
    /// Build the set for primary key columns only (PK is a prefix of sorting key;
    /// the remaining sorting key columns are useless for index analysis).
    SizeLimits set_size_limits(optimization_settings.max_rows_for_lazy_final, optimization_settings.max_bytes_for_lazy_final, OverflowMode::BREAK);

And there is a hard limit on the set size, so the optimization cannot blow up memory on a broad query.

Reverse-order reading

One more modern wrinkle. SELECT FINAL with ORDER BY on a descending sorting key can be answered by reading the parts backwards and taking the first row of each group rather than the last:

cpp 
// src/Processors/QueryPlan/ReadFromMergeTree.cpp:2481
    const bool read_in_reverse = reader_settings.read_in_order && query_info.input_order_info && query_info.input_order_info->direction < 0;
    const auto read_type = read_in_reverse ? ReadType::InReverseOrder : ReadType::InOrder;

and the merging algorithm has to be told, because the tie-break rule flips:

cpp 
// src/Processors/QueryPlan/MergeTreeFinalMerge.cpp:56
    /// Only the Replacing merging algorithm supports reading in the reverse order
    /// (see `ReadFromMergeTree::requestReadingInOrder`).
    if (read_in_reverse && merging_params.mode != MergeTreeData::MergingParams::Replacing)
        throw Exception(ErrorCodes::LOGICAL_ERROR, "Reading in reverse order with FINAL is supported only for ReplacingMergeTree");

Only Replacing supports it, because only Replacing’s tie-break rule is “the row written last wins” — a rule that can be evaluated in either direction. Collapsing and Summing aggregate, and an aggregate is not direction-independent.

Trade-offs

ReplacingMergeTree is eventually consistent. The collapse happens during merges, which happen in the background, on a schedule you do not control. Until a merge happens, the table physically contains both the old and the new row. A query without FINAL sees both. The documentation says it plainly:

cpp 
// src/Storages/MergeTree/registerStorageMergeTree.cpp:3239
At merge time, the ReplacingMergeTree identifies duplicate rows, using the values of the `ORDER BY` columns (used to create the table) as a unique identifier, and retains only the highest version. This, however, offers eventual correctness only - it does not guarantee rows will be deduplicated, and you should not rely on it. Queries can, therefore, produce incorrect answers due to update and delete rows being considered in queries.

FINAL is expensive. It re-merges at query time. The lazy-FINAL optimization helps a lot when the query is selective, but a FINAL over the whole table is a full merge on every query.

FINAL changes what predicates mean. A predicate is applied after the collapse, so SELECT ... WHERE version = 5 FINAL finds the row whose winning version is 5 — not a row that had version 5 and lost. Without FINAL, the same predicate might match the loser.

The difference is visible in one query:

sql 
-- Without FINAL, both versions are present until a merge happens.
SELECT user_id, updated_at FROM user_profile WHERE user_id = 42;

-- With FINAL, only the winner survives.
SELECT user_id, updated_at FROM user_profile WHERE user_id = 42 FINAL;

The version must be monotonic. “No matter the operation on the data, the version should be increased.” A version that goes backwards resurrects stale data, and there is no way for ClickHouse to detect that.

Deletes are two-phase. By default the tombstone is kept (so that late lower-version rows cannot resurrect the row), and only OPTIMIZE ... FINAL CLEANUP removes it. Space is held until then.

The bottom line

ReplacingMergeTree is the smallest interesting member of the MergeTree family: one row per sorting key, the highest version wins, ties broken by insertion order. The implementation is a merging algorithm that has to be careful about three things — the three-way version comparison, the tie-break rule’s dependence on read direction, and the “no false skip” requirement that makes the run-skipping fast path safe.

SELECT FINAL is where the interesting engineering lives, because it has to reproduce a merge’s result at query time. ClickHouse’s progression here is instructive: from “merge everything on every query”, to “merge everything but copy only the survivors” (vertical final), to “prove that most parts need no merge at all, and compute the winners with a grouped argMax over a UInt128 tiebreaker” (lazy final). Each step trades a specific, well-understood correctness argument for a large constant-factor win — and the tiebreaker packing, (version << 64) + row_index, with the sign-bit flip for signed versions, is the kind of detail that only works because someone reasoned about it very carefully.

ClickhouseInternalsMergetreeReplacingmergetreeFinal