Opensource internals

ClickHouse Sorts in Parallel: Partial Sort, Merge Sorted Streams and External Sort

17 min read Jay Hubbert

ClickHouse sorts each stream on its own, keeps a running threshold to drop hopeless rows, then merges the sorted streams — spilling to disk when memory runs out.

TL;DR: ClickHouse never sorts the whole dataset at once. It sorts each chunk, throws away rows that can’t make the top-N, merges the sorted streams, and spills runs to disk when memory runs out.

ORDER BY is the operation every database is worst at. It’s the one requirement that’s inherently global: the first row of the result depends on every row of the input.

Everything else — filtering, projection, aggregation — is local and streams beautifully. Sorting doesn’t.

The classic answer is “sort in chunks, then merge the sorted chunks”. Every system does it. But there are two quite different versions of that, and ClickHouse implements both, in two quite different classes.

One version assumes the input is already sorted on a prefix of the sort key — a MergeTree table read in primary-key order, or the output of a previous ORDER BY. Then the merge is cheap and the only work is stitching together runs that are already in order.

The other version assumes nothing. It sorts each chunk independently and then does a k-way merge. This is the general case, and it’s where the memory pressure lives — a k-way merge of N sorted runs needs all N runs resident at once.

This is a tour of both, of the trick that lets ClickHouse throw away rows it knows are useless, and of the machinery that spills sorted runs to disk when memory runs out.

The plan: what gets built

SortingStep decides which of the two regimes applies. You can see the difference in the query:

sql 
-- Full sort: the input is in no particular order, so each stream is sorted
-- on its own and the sorted streams are merged.
SELECT * FROM hits ORDER BY UserID, EventTime LIMIT 100;
sql 
-- Finish sorting: the table is sorted by (a, b) and the query asks for
-- (a, b) too, so the merge only has to stitch runs together.
SELECT * FROM hits WHERE a = 1 ORDER BY a, b LIMIT 100;

The “finish sorting” case is when the input is already sorted on a prefix:

cpp 
// src/Processors/QueryPlan/SortingStep.cpp:381
void SortingStep::finishSorting(
    QueryPipelineBuilder & pipeline, const SortDescription & input_sort_desc, const SortDescription & result_sort_desc, const UInt64 limit_)
{
    pipeline.addSimpleTransform(
        [&](const SharedHeader & header, QueryPipelineBuilder::StreamType stream_type) -> ProcessorPtr
        {
            if (stream_type != QueryPipelineBuilder::StreamType::Main)
                return nullptr;

            return std::make_shared<PartialSortingTransform>(header, result_sort_desc, limit_, threshold_tracker);
        });

Two transforms per stream: a PartialSortingTransform that sorts each chunk as it arrives, and a FinishSortingTransform that merges the sorted chunks. The “full sort” case is the same first transform followed by a MergeSortingTransform:

cpp 
// src/Processors/QueryPlan/SortingStep.cpp:522
    if (!skip_partial_sort || limit_)
    {
        pipeline.addSimpleTransform(
            [&](const SharedHeader & header, QueryPipelineBuilder::StreamType stream_type) -> ProcessorPtr
            {
                if (stream_type != sorted_stream_type)
                    return nullptr;

                return std::make_shared<PartialSortingTransform>(header, result_sort_desc, limit_, threshold_tracker);
            });

Note the condition !skip_partial_sort || limit_. When there’s a LIMIT, the partial sort is always applied, even if the input is already sorted. That’s because the partial sort is what makes the limit cheap: it discards everything but the top limit rows of each chunk, so the merge only ever sees limit rows per chunk instead of the whole chunk.

PartialSortingTransform: sort the chunk, throw away the rest

The transform itself is short, and the interesting part is the threshold filter at the top:

cpp 
// src/Processors/Transforms/PartialSortingTransform.cpp:104
void PartialSortingTransform::transform(Chunk & chunk)
{
    if (chunk.getNumRows())
    {
        // The following code works with Blocks and will lose the number of
        // rows when there are no columns. We shouldn't get such block, because
        // we have to sort by at least one column.
        chassert(chunk.getNumColumns());
    }

    if (read_rows)
        read_rows->add(chunk.getNumRows());

    auto block = getInputPort().getHeader().cloneWithColumns(chunk.detachColumns());

    /// Materialize sort key columns that are ColumnReplicated to avoid index indirection during comparison and sorting.
    for (const auto & col_desc : description)
    {
        auto & column_entry = block.getByName(col_desc.column_name);
        column_entry.column = convertToFullColumnIfReplicationNotUseful(column_entry.column);
    }

    /** If we've saved columns from previously blocks we could filter all rows from current block
      * which are unnecessary for sortBlock(...) because they obviously won't be in the top LIMIT rows.
      */
    if (!sort_description_threshold_columns.empty() && !threshold_tracker)
    {
        UInt64 rows_num = block.rows();
        auto block_columns = extractRawColumns(block, description_with_positions);

        size_t result_size_hint = getFilterMask(
                block_columns, sort_description_threshold_columns,
                description, rows_num, filter, rows_to_compare, compare_results);

        /// Everything was filtered. Skip whole chunk.
        if (result_size_hint == 0)
            return;

        if (result_size_hint < rows_num)
        {
            for (auto & column : block)
                column.column = column.column->filter(filter, result_size_hint);
        }
    }

    sortBlock(block, description, limit);

Two things happen here, and the order matters.

First, the threshold filter. sort_description_threshold_columns holds the sort key of the current N-th best row seen so far — the “cutoff”. Every row in the new chunk that compares greater than the cutoff provably cannot be in the final top-N, so it’s dropped before the sort. The filter is computed with a vectorised column comparison:

cpp 
// src/Processors/Transforms/PartialSortingTransform.cpp:27
size_t getFilterMask(const ColumnRawPtrs & raw_block_columns, const Columns & threshold_columns,
                     const SortDescription & description, size_t num_rows, IColumn::Filter & filter,
                     PaddedPODArray<UInt64> & rows_to_compare, PaddedPODArray<Int8> & compare_results)
{
    filter.resize(num_rows);
    compare_results.resize(num_rows);

    if (description.size() == 1)
    {
        /// Fast path for single column
        raw_block_columns[0]->compareColumn(*threshold_columns[0], 0, nullptr, compare_results,
                              description[0].direction, description[0].nulls_direction);
    }
    else
    {
        rows_to_compare.resize(num_rows);
        iota(rows_to_compare.data(), num_rows, UInt64(0));

        size_t size = description.size();
        for (size_t i = 0; i < size; ++i)
        {
            raw_block_columns[i]->compareColumn(*threshold_columns[i], 0, &rows_to_compare, compare_results,
                                  description[i].direction, description[i].nulls_direction);

            if (rows_to_compare.empty())
                break;
        }
    }

    size_t result_size_hint = 0;

    for (size_t i = 0; i < num_rows; ++i)
    {
        /// Leave only rows that are less then row from rhs.
        filter[i] = compare_results[i] < 0;
        result_size_hint += filter[i];
    }

    return result_size_hint;
}

compareColumn is the columnar comparison primitive: it compares a whole column against a single-row column and writes a per-row sign into compare_results. For a single sort column it’s one vectorised pass. For multiple columns it narrows the candidate set on each column in turn — the rows_to_compare array is the survivor list, and it’s emptied when no rows survive.

Second, sortBlock(block, description, limit). The limit argument makes the sort itself partial: ClickHouse sorts the chunk but only materialises the first limit rows of the permutation.

cpp 
// src/Interpreters/sortBlock.cpp:384
void sortBlock(Block & block, const SortDescription & description, UInt64 limit, IColumn::PermutationSortStability stability)
{
    IColumn::Permutation permutation;

#ifndef NDEBUG
    block.checkNumberOfRows();
#endif
    getBlockSortPermutationImpl(block, description, stability, limit, permutation);

#ifndef NDEBUG
    checkSortedWithPermutation(block, description, limit, permutation);
#endif

    if (permutation.empty())
        return;

    bool is_identity_permutation = isIdentityPermutation(permutation, limit);
    if (is_identity_permutation && limit == 0)
        return;

    size_t output_rows = limit ? std::min(static_cast<size_t>(limit), permutation.size()) : permutation.size();
    Columns columns = block.getColumns();
    transformColumnsWithSharedIndex(
        columns,
        [&](const ColumnPtr & col) { return is_identity_permutation ? col->cut(0, output_rows) : col->permute(permutation, limit); });
    block.setColumns(columns);
}

Note the output_rows clamp. With a limit of 10 and a chunk of 100 000 rows, the output chunk has 10 rows.

The sort still had to consider all 100 000 rows (the top 10 of an unsorted chunk aren’t the first 10), but the permutation only materialises 10 entries and the columns are permuted with that.

So the memory cost of ORDER BY ... LIMIT 10 is proportional to the chunk size, not to the limit. Each chunk is sorted in full, truncated to 10 rows, and the merge only ever sees 10 rows per chunk.

Updating the threshold

After sorting, the transform updates the threshold to the current chunk’s limit-th row:

cpp 
// src/Processors/Transforms/PartialSortingTransform.cpp:151
    /// Check if we can use this block for optimization.
    if ((min_limit_for_partial_sort_optimization <= limit || threshold_tracker) && limit <= block.rows())
    {
        /** If we filtered more than limit rows from block take block last row.
          * Otherwise take last limit row.
          *
          * If current threshold value is empty, update current threshold value.
          * If min block value is less than current threshold value, update current threshold value.
          */
        size_t min_row_to_compare = limit - 1;
        auto raw_block_columns = extractRawColumns(block, description_with_positions);

        if (sort_description_threshold_columns.empty() ||
            compareWithThreshold(raw_block_columns, min_row_to_compare, sort_description_threshold_columns, description))
        {
            size_t raw_block_columns_size = raw_block_columns.size();
            Columns sort_description_threshold_columns_updated(raw_block_columns_size);

            for (size_t i = 0; i < raw_block_columns_size; ++i)
            {
                MutableColumnPtr sort_description_threshold_column_updated = raw_block_columns[i]->cloneEmpty();
                sort_description_threshold_column_updated->insertFrom(*raw_block_columns[i], min_row_to_compare);
                /// Without this, `assertTypeEquality` may fail when comparing a ColumnNullable against a ColumnReplicated (received from JOIN).
                sort_description_threshold_columns_updated[i]
                    = removeSpecialRepresentations(sort_description_threshold_column_updated->getPtr());
            }

            sort_description_threshold_columns = std::move(sort_description_threshold_columns_updated);

The threshold is monotonically non-increasing — it only ever gets better — so the filter gets more aggressive over time. The min_limit_for_partial_sort_optimization guard exists because for very small limits the bookkeeping isn’t worth it.

There’s a subtlety worth noticing: the filter is not stable. The comparison keeps rows that are strictly less than the threshold (compare_results[i] < 0), so a row equal to the threshold gets dropped. That’s fine for ORDER BY ... LIMIT without WITH TIES, and it’s why LimitTransform has a completely separate mechanism for WITH TIES.

FinishSortingTransform: merging streams that are already sorted

When the input is already sorted on a prefix of the key, the merge doesn’t need a heap at all — it just has to concatenate runs. The one hard part is handling a key that spans a chunk boundary.

cpp 
// src/Processors/Transforms/FinishSortingTransform.cpp:83
    /// Find the position of last already read key in current chunk.
    if (!chunks.empty())
    {
        size_t size = chunk.getNumRows();
        const auto & last_chunk = chunks.back();

        ssize_t low = -1;
        ssize_t high = size;
        while (high - low > 1)
        {
            ssize_t mid = (low + high) / 2;
            if (!less(last_chunk.getColumns(), chunk.getColumns(), last_chunk.getNumRows() - 1, mid, description_with_positions))
                low = mid;
            else
                high = mid;
        }

        size_t tail_pos = high;

        /// We need to save tail of chunk, because next chunk may starts with the same key as in tail
        /// and we should sort these rows in one portion.
        if (tail_pos != size)
        {
            auto source_columns = chunk.detachColumns();
            Columns tail_columns;

            for (auto & source_column : source_columns)
            {
                tail_columns.emplace_back(source_column->cut(tail_pos, size - tail_pos));
                source_column = source_column->cut(0, tail_pos);
            }

            chunks.emplace_back(std::move(source_columns), tail_pos);
            tail_chunk.setColumns(std::move(tail_columns), size - tail_pos);

            stage = Stage::Generate;
            return;
        }
    }

The binary search finds the first row of the new chunk whose key is greater than the last key of the previous chunk. Everything from there on has a key that might tie with the tail, so it’s cut off into tail_chunk and held back. The tail is re-merged with the next chunk, so that rows with equal keys end up in one sorted run rather than being split across two and appearing in the wrong order.

The stage = Stage::Generate is the switch that makes this work: the transform has to emit what it has so far, then go back to consuming.

MergeSortingTransform: accumulate, remerge, spill, merge

The general case is MergeSortingTransform, and it’s a state machine with three stages — Consume, Serialize, Generate — declared in the base class:

cpp 
// src/Processors/Transforms/SortingTransform.h:126
    enum class Stage : uint8_t
    {
        Consume = 0,
        Generate,
        Serialize,
    };

    Stage stage = Stage::Consume;

The base class prepare() is the dispatcher:

cpp 
// src/Processors/Transforms/SortingTransform.cpp:293
IProcessor::Status SortingTransform::prepare()
{
    if (stage == Stage::Serialize)
    {
        if (!processors.empty())
            return Status::UpdatePipeline;

        auto status = prepareSerialize();
        if (status != Status::Finished)
            return status;

        stage = Stage::Consume;
    }

    if (stage == Stage::Consume)
    {
        auto status = prepareConsume();
        if (status != Status::Finished)
            return status;

        stage = Stage::Generate;
    }

    /// stage == Stage::Generate

    if (!generated_prefix || !chunks.empty())
        return Status::Ready;

    if (!processors.empty())
        return Status::UpdatePipeline;

    return prepareGenerate();
}

The Status::UpdatePipeline returns are the graph-rewriting mechanism from the first article: when the sort decides it needs to spill to disk, it returns UpdatePipeline, the executor asks it for updatePipeline(), and the graph is rewired to insert a file sink and a file source.

Consume: accumulate chunks

cpp 
// src/Processors/Transforms/MergeSortingTransform.cpp:94
void MergeSortingTransform::consume(Chunk chunk)
{
    /** Algorithm:
      * - read to memory blocks from source stream;
      * - if too many of them and if external sorting is enabled,
      *   - merge all blocks to sorted stream and write it to temporary file;
      * - at the end, merge all sorted streams from temporary files and also from rest of blocks in memory.
      */

    /// If there were only const columns in sort description, then there is no need to sort.
    /// Return the chunk as is.
    if (description.empty())
    {
        generated_chunk = std::move(chunk);
        return;
    }

    removeConstColumns(chunk);
    compactReplicatedColumns(chunk);

    sum_rows_in_blocks += chunk.getNumRows();
    sum_bytes_in_blocks += chunk.allocatedBytes();
    chunks.push_back(std::move(chunk));

    /** If significant amount of data was accumulated, perform preliminary merging step.
      */
    if ((chunks.size() > 1
        && limit
        && limit * 2 < sum_rows_in_blocks   /// 2 is just a guess.
        && remerge_is_useful
        && max_bytes_before_remerge
        && sum_bytes_in_blocks > max_bytes_before_remerge) || (threshold_tracker && (static_cast<double>(sum_rows_in_blocks) > static_cast<double>(limit) * 1.5)))
    {
        remerge();
    }

    /** If too many of them and if external sorting is enabled,
      *  will merge blocks that we have in memory at this moment and write merged stream to temporary (compressed) file.
      * NOTE. It's possible to check free space in filesystem.
      */
    if (max_bytes_in_block_before_external_sort && sum_bytes_in_blocks > max_bytes_in_block_before_external_sort)
    {
        Int64 query_memory = getCurrentQueryMemoryUsage();
        if (!max_bytes_in_query_before_external_sort || query_memory > static_cast<Int64>(max_bytes_in_query_before_external_sort))
        {
            auto sink = prepareSpill();
            auto source = std::make_shared<BufferingFromFileSource>(sink->getPort().getSharedHeader(), sink->getHolder(), log);
            processors.emplace_back(source);
            processors.emplace_back(sink);
            stage = Stage::Serialize;
        }
    }
}

Three behaviours in one function.

Remerge. When there’s a LIMIT and enough data has accumulated, the in-memory chunks are merged now rather than later. This is the classic top-K trick: after merging, only limit rows survive, so the memory footprint of the accumulator collapses from “all data so far” to “the top limit rows”. The condition limit * 2 < sum_rows_in_blocks means “we have at least twice as many rows as we need”, so the remerge is worth it.

Spill. When the accumulated bytes exceed max_bytes_in_block_before_external_sort (and the whole query’s memory exceeds max_bytes_in_query_before_external_sort), the accumulated chunks are merged and written to a temporary file. The transform enters the Serialize stage, and the new processors — the file sink and the file source — are queued for updatePipeline().

The limit on remerge. remerge_is_useful is a latch:

cpp 
// src/Processors/Transforms/MergeSortingTransform.cpp:333
void MergeSortingTransform::remerge()
{
    LOG_DEBUG(log, "Re-merging intermediate ORDER BY data ({} blocks with {} rows) to save memory consumption", chunks.size(), sum_rows_in_blocks);

    /// NOTE Maybe concat all blocks and partial sort will be faster than merge?
    MergeSorter remerge_sorter(std::make_shared<const Block>(header_without_constants), std::move(chunks), description, max_merged_block_size, limit);

    Chunks new_chunks;
    size_t new_sum_rows_in_blocks = 0;
    size_t new_sum_bytes_in_blocks = 0;

    while (auto chunk = remerge_sorter.read())
    {
        new_sum_rows_in_blocks += chunk.getNumRows();
        new_sum_bytes_in_blocks += chunk.allocatedBytes();
        new_chunks.emplace_back(std::move(chunk));
    }

    LOG_DEBUG(log, "Memory usage is lowered from {} to {}", ReadableSize(sum_bytes_in_blocks), ReadableSize(new_sum_bytes_in_blocks));

    /// If the memory consumption was not lowered enough - we will not perform remerge anymore.
    if (remerge_lowered_memory_bytes_ratio > 0.0 && (static_cast<double>(new_sum_bytes_in_blocks) * remerge_lowered_memory_bytes_ratio > static_cast<double>(sum_bytes_in_blocks)))
    {
        remerge_is_useful = false;
        LOG_DEBUG(log, "Re-merging is not useful (memory usage was not lowered by remerge_sort_lowered_memory_bytes_ratio={})", remerge_lowered_memory_bytes_ratio);
    }

The default ratio is 2.0 (remerge_sort_lowered_memory_bytes_ratio), so the remerge must halve the memory to be considered worth doing. If it doesn’t, the flag is cleared and no further remerges happen — because a remerge that doesn’t reduce memory is pure loss: you paid for a merge and got nothing.

The default threshold is max_bytes_before_remerge_sort = 1000000000, one gigabyte.

Serialize and Generate

Serialize is the stage where the accumulated chunks are drained into the file sink, one sorted run at a time:

cpp 
// src/Processors/Transforms/MergeSortingTransform.cpp:252
void MergeSortingTransform::serialize()
{
    current_chunk = merge_sorter->read();
    if (!current_chunk)
        merge_sorter.reset();
    /// A read of `MergeSorter::Mode::MergeUniqueChunks` that consumes only duplicates returns no rows. The
    /// run gets no block for it, and the next call reads on.
    else if (!current_chunk.hasRows())
        current_chunk.clear();
}

Generate is where the final merge happens. If nothing was spilled, the answer is a plain in-memory merge of the accumulated chunks:

cpp 
// src/Processors/Transforms/MergeSortingTransform.cpp:263
void MergeSortingTransform::generate()
{
    if (!generated_prefix)
    {
        if (temporary_files_num == 0)
        {
            merge_sorter = std::make_unique<MergeSorter>(
                std::make_shared<const Block>(header_without_constants),
                std::move(chunks),
                description,
                max_merged_block_size,
                limit,
                merge_mode);
        }
        else
        {
            ProfileEvents::increment(external_merge_event);
            LOG_INFO(log, "There are {} temporary sorted parts to merge", temporary_files_num);

            processors.emplace_back(
                std::make_shared<MergeSorterSource>(
                    std::make_shared<const Block>(header_without_constants),
                    std::move(chunks),
                    description,
                    max_merged_block_size,
                    limit,
                    merge_mode));
        }

        generated_prefix = true;
    }

If there were temporary files, a MergeSorterSource is added to the pipeline and the graph is rewired so that the spilled runs (read back from disk) and the in-memory tail are merged together by a MergingSortedTransform — which is a proper k-way merge with a heap.

The k-way merge

MergingSortedTransform uses MergingSortedAlgorithm, which keeps one SortCursor per input stream in a heap and pops the smallest key. The interesting optimisation is the “totally less” shortcut:

cpp 
// src/Processors/Merges/Algorithms/MergingSortedAlgorithm.cpp:547
template <typename TSortingHeap>
IMergingAlgorithm::Status MergingSortedAlgorithm::mergeImpl(TSortingHeap & queue)
{
    /// Take rows in required order and put them into `merged_data`, while the rows are no more than `max_block_size`
    while (queue.isValid())
    {
        if (merged_data.hasEnoughRows())
            return Status(merged_data.pull());

        auto current = queue.current();

        if (current.impl->isLast() && current_inputs[current.impl->order].skip_last_row)
        {
            /// Get the next block from the corresponding source, if there is one.
            queue.removeTop();
            return Status(current.impl->order);
        }

        if (current.impl->isFirst()
            && !current_inputs[current.impl->order].skip_last_row /// Ignore optimization if last row should be skipped.
            && (queue.size() == 1
                || (queue.size() >= 2 && current.totallyLessOrEquals(queue.nextChild()))))
        {
            /** This is special optimization if current cursor is totally less than next cursor.
              * We want to insert current cursor chunk directly in merged data.
              *
              * First if merged_data is not empty we need to flush it.
              * We will get into the same condition on next mergeBatch call.
              *
              * Then we can insert chunk directly in merged data.
              */

            if (merged_data.mergedRows() != 0)
                return Status(merged_data.pull());

            /// Actually, current.impl->order stores source number (i.e. cursors[current.impl->order] == current.impl)
            size_t source_num = current.impl->order;

            auto chunk_num_rows = current_inputs[source_num].chunk.getNumRows();

            UInt64 total_merged_rows_after_insertion = merged_data.mergedRows() + chunk_num_rows;
            bool limit_reached = limit && total_merged_rows_after_insertion >= limit;

            if (limit && total_merged_rows_after_insertion > limit)
                chunk_num_rows -= total_merged_rows_after_insertion - limit;

            insertChunk(source_num);

            /// We will get the next block from the corresponding source, if there is one.
            queue.removeTop();

            auto status = Status(merged_data.pull(), limit_reached);

            if (!limit_reached)
                status.required_source = source_num;

            return status;
        }

totallyLessOrEquals asks: “is the last key of this stream less than or equal to the first key of the next-smallest stream?” If yes, then the entire chunk of the current stream comes before everything in the other streams, and it can be copied wholesale instead of row by row.

That turns the merge from O(rows log streams) comparisons into a memcpy for the common case. When the data is nearly sorted — which it usually is, because each run came from an already-sorted source — most chunks take the fast path.

Trade-offs

What this buys you.

The LIMIT optimisation is the big one. ORDER BY x LIMIT 10 over a hundred million rows costs: sort each chunk (in memory, chunk by chunk), truncate to 10 rows, merge 10 rows per chunk. Memory is O(chunk size × threads), not O(dataset). And the threshold filter improves on that further: once the cutoff is tight, whole chunks are discarded without being sorted at all — if (result_size_hint == 0) return;.

The remerge trick generalises the same idea beyond LIMIT. Even without a limit, merging periodically keeps the accumulator’s memory bounded and converts a “merge N runs at the end” problem into a series of smaller merges, which is also better for the page cache.

The external sort path is honest about the fact that ORDER BY is a blocking operation: if the data doesn’t fit, it goes to disk, in sorted runs, and the final merge is a k-way merge over those runs. The runs are written with the configured temporary-files codec (usually LZ4), so the disk traffic is compressed.

What it costs.

Sorting is still the slowest thing in the query, and the partial sort doesn’t help the general case. Without a LIMIT, every chunk must be fully sorted, and the merge must touch every row. sortBlock allocates a permutation of the chunk’s row indices — 8 bytes per row — and the column permutation that follows is a random-access pass over the column data, which is cache-hostile for wide rows.

The threshold filter has a cost of its own: a full vectorised comparison of the chunk against the cutoff, even when it rejects nothing. For a LIMIT query where the data isn’t sorted, the filter will usually reject nothing on the first chunk and progressively more later, so the average benefit depends heavily on the data.

The remerge is a gamble. It costs a merge, and it only pays off if the merge actually reduces memory by the configured ratio. The remerge_is_useful latch means the first unhelpful remerge disables all future ones — but the first one was already paid for. And the threshold for even attempting a remerge is one gigabyte of accumulated data, so the memory spike before the first remerge is real.

Finally, the tail-splitting logic in FinishSortingTransform is subtle and easy to get wrong. If a key spans a chunk boundary and the tail isn’t held back, the rows with equal keys are emitted in two separate sorted runs and the final order among them is wrong. The binary search that finds the split point is O(log n) per chunk, which is cheap — but the reason it exists is not obvious from the code, which is why the comment is there.

Wrapping up

The real win: the threshold filter. ORDER BY ... LIMIT N in ClickHouse isn’t “sort everything, take N”. It’s “sort a chunk, keep N, discard the rest, and tighten the cutoff so later chunks mostly die at the door”. The query gets faster the longer it runs.

What to do about it: if your ORDER BY ... LIMIT query is slow, check whether the sort key matches the table’s primary key. If it does, you get the “finish sorting” path — a merge of already-sorted runs, which is close to free. If it doesn’t, you’re paying for a full sort plus a threshold filter that has to reject everything by comparison.

And if you’re spilling to disk, watch the external_merge_event profile event and max_bytes_before_external_sort. That’s the knob that decides how much of your sort stays in RAM.

ClickhouseInternalsSortingPerformance