Opensource internals

ClickHouse Computes Window Frames Incrementally, One Row at a Time

16 min read Jay Hubbert

ClickHouse's window transform keeps the partition sorted, advances a moving frame bound by bound, and updates the aggregate state incrementally instead of recomputing it per row.

TL;DR: A window function is an aggregate over a moving range of rows. The naive implementation recomputes the aggregate for every row, which is quadratic. ClickHouse sorts the input by PARTITION BY then ORDER BY, keeps three nested bounds — partition, peer group, frame — and for each row advances those bounds by a delta, adding only the rows that entered the frame. When the frame does not move at all, the previous result is copied forward instead of being finalised again.

sql 
SELECT
    team,
    player,
    salary,
    avg(salary) OVER (PARTITION BY team ORDER BY salary ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS rolling
FROM salaries;

The temptation is to implement this the obvious way: for each row, gather the rows in its frame and aggregate them. That is O(n · frame) per partition, and with the default frame (RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) it is O(n²). For a million-row partition that is 10^12 row visits.

The trick that makes it linear is to notice that consecutive rows have almost the same frame. The frame for row i+1 is the frame for row i plus a few rows at the end and minus a few at the start. If you keep the aggregate’s state alive between rows, you only have to feed it the difference.

That is the entire design of ClickHouse’s window transform, and the rest of this article is about how it manages the fact that the frame’s boundaries are not known until you have seen the data.

text 
   after the sort (PARTITION BY team, ORDER BY salary)

   team A                     team B
   +----------------------+   +----------------------+
   | 10  20  20  30  40  |   | 50  50  60           |
   +----------------------+   +----------------------+
     ^partition start        ^partition start
     |                        |
   peer group:  [20 20]       peer group: [50 50]
   frame (ROWS 2 PRECEDING):  frame (RANGE UNBOUNDED PRECEDING):
     [10 20 20]                 [50]
     [20 20 20]                 [50 50]
     [20 20 30]                 [50 50 60]   <- the start never moves
     [20 30 40]                 ...

   for each row: advance the frame, feed the aggregate only the rows
                  that entered it, finalise, write the row

The plan: sort, then transform

The window step adds its own sort. PARTITION BY columns come first, then ORDER BY columns, and the result is one stream per partition group:

cpp 
// src/Processors/QueryPlan/WindowStep.cpp:85
    // This resize is needed for cases such as `over ()` when we don't have a
    // sort node, and the input might have multiple streams. The sort node would
    // have resized it.
    if (window_description.full_sort_description.empty())
        pipeline.resize(1);

    /// The sorting step before the window sorts only the main stream, but `WindowTransform` also processes the totals,
    /// which an `arrayJoin` can expand into several unsorted rows.
    if (!window_description.full_sort_description.empty())
        SortingStep::fullSortStreams(
            pipeline, sort_settings, window_description.full_sort_description, /*limit_=*/0,
            /*skip_partial_sort=*/false, /*threshold_tracker=*/nullptr, /*sort_totals=*/true);

    pipeline.addSimpleTransform(
        [&](const SharedHeader & /*header*/)
        {
            return std::make_shared<WindowTransform>(
                input_headers.front(), output_header, window_description, window_functions);
        });

The sort is not optional and it is not free. This is the price of the incremental frame: to advance a frame forward you need the rows in order.

Three nested bounds

The transform maintains three independent notions of “where are we”:

  • the partition — the range of rows with the same PARTITION BY key;
  • the peer group — the range of rows with the same ORDER BY key, which is what RANGE and GROUPS frames are built from;
  • the frame — the actual window, expressed in terms of ROWS, RANGE or GROUPS boundaries.

Each is a small class with an advance method, and each is expressed as a RowPoint: a block number, a row within the block, a row index within the partition, and a peer-group index within the partition.

The partition is the outermost one, and it is discovered by scanning forward for the next row whose partition key differs:

cpp 
// src/Processors/Transforms/Window/Partition.cpp:20
void Partition::advance(const SlidingBlocks & blocks)
{
    RowPoint & end = partition_bounds.end;
    while (!partition_bounds.fully_visible && end.location < blocks.end())
    {
        const SlidingBlock & block = blocks.blockAt(end.location.block);

        int64_t next_start = end.location.row;
        while (next_start < block.rows_count && !block.index.partition_starts[next_start])
            ++next_start;

The partition_starts bitmap is computed once per block when the block arrives, and it is a bitmap, not a set of row indices — one bit per row, so a block of 65 536 rows costs 8 KB:

cpp 
// src/Processors/Transforms/Window/SlidingIndexes.cpp:24
std::vector<bool> markPartitionStarts(const Columns & columns, size_t rows_count, const std::vector<size_t> & partition_by_indices, const std::optional<Columns> & previous_partition_key)
{
    std::vector<bool> starts(rows_count, false);
    starts[0] = !previous_partition_key || !haveSameKeys(*previous_partition_key, 0, columns, 0, partition_by_indices);

    size_t next_start = getEqualRangeEndAssumeSorted(columns, partition_by_indices, 0, rows_count, /*nan_direction_hint=*/1);
    while (next_start < rows_count)
    {
        starts[next_start] = true;
        next_start = getEqualRangeEndAssumeSorted(columns, partition_by_indices, next_start, rows_count, /*nan_direction_hint=*/1);
    }

    return starts;
}

Note the previous_partition_key parameter: a partition can straddle a block boundary, so the last row of the previous block has to be compared with the first row of this one. The code keeps the last row’s key columns alive for exactly this:

cpp 
// src/Processors/Transforms/Window/SlidingIndexes.cpp:82
Columns cutLastKey(const Columns & columns, size_t rows_count, const std::vector<size_t> & key_indices)
{
    Columns last_row(columns.size());
    for (const size_t key : key_indices)
        last_row[key] = columns[key]->cut(rows_count - 1, 1);

    return last_row;
}

Peer groups get an index per row rather than a bitmap, because the frame machinery needs to know, for an arbitrary row, which peer group it is in:

cpp 
// src/Processors/Transforms/Window/SlidingIndexes.cpp:39
std::vector<int64_t> assignPeerGroupIndexes(
    const Columns & columns,
    size_t rows_count,
    const WindowTransformParams & params,
    const std::vector<bool> & partition_starts,
    const std::optional<Columns> & previous_order_key,
    int64_t last_peer_group_index)
{
    std::vector<int64_t> peer_group_indexes(rows_count);

There is one subtlety here: under a ROWS frame every row is its own peer group, so the whole index is just a counter:

cpp 
// src/Processors/Transforms/Window/SlidingIndexes.cpp:49
    const bool every_row_is_a_group = params.window_description.frame.type == WindowFrame::FrameType::ROWS;
    if (!every_row_is_a_group && order_by_indices.empty())
        return peer_group_indexes;

Advancing the frame

The frame is the interesting one. Its two boundaries are computed by advanceFrameStart and advanceFrameEnd, and the frame type decides which of three mechanisms is used.

ROWS is trivial — the boundary is a row index:

cpp 
// src/Processors/Transforms/Window/Frame.cpp:118
        case WindowFrame::BoundaryType::Offset:
            switch (frame.type)
            {
                case WindowFrame::FrameType::ROWS:
                    return advanceToRow(blocks, partition, current.row_index_in_partition + signedOffset(frame.begin_offset, frame.begin_preceding), start);
                case WindowFrame::FrameType::RANGE:
                    return advanceToRangeBound(params, blocks, partition, current, frame.begin_offset, frame.begin_preceding, false, start);
                case WindowFrame::FrameType::GROUPS:
                    return advanceToPeerGroup(blocks, partition, current.peer_group_index_in_partition + signedOffset(frame.begin_offset, frame.begin_preceding), start);
            }

GROUPS is a peer-group index, and finding the first row of a given peer group means walking the per-row index forward:

cpp 
// src/Processors/Transforms/Window/Frame.cpp:56
    while (bound.location != partition_end)
    {
        const SlidingBlock & block = blocks.blockAt(bound.location.block);
        const std::vector<int64_t> & groups = block.index.peer_group_indexes;
        const int64_t rows_end = partition_end.block == bound.location.block ? partition_end.row : block.rows_count;
        const int64_t row = groups[bound.location.row] >= group
            ? bound.location.row
            : static_cast<int64_t>(findEqualRangeEndAssumeSorted(bound.location.row, rows_end, 16, [&](size_t i) { return groups[i] < group; }));

RANGE is the hardest: the boundary is a value, not a position. RANGE BETWEEN 3 PRECEDING AND CURRENT ROW means “every row whose ORDER BY value is at least current_value - 3”. The search is a linear scan over the sorted key column with a comparator that shifts the current value by the offset:

cpp 
// src/Processors/Transforms/Window/Frame.cpp:85
    while (bound.location != partition_end)
    {
        const SlidingBlock & block = blocks.blockAt(bound.location.block);
        const IColumn * key_column = block.materialized_columns[key_position].get();
        const int64_t rows_end = partition_end.block == bound.location.block ? partition_end.row : block.rows_count;
        int64_t row = bound.location.row;
        while (row < rows_end && params.range_offset_comparator(key_column, row, current_key_column, current.location.row, offset, offset_is_preceding) * direction < past)
            ++row;

The comparator is chosen once, at construction time, and specialised on the key’s type — there is no per-row type dispatch:

cpp 
// src/Processors/Transforms/Window/WindowTransformParams.cpp:131
/// Null for every frame but RANGE OFFSET, which has exactly one ORDER BY key.
RangeOffsetComparator chooseRangeOffsetComparator(const Block & header, const WindowFrame & frame, const SortDescription & order_by, const std::vector<size_t> & order_by_indices)
{
    if (!isRangeOffsetFrame(frame))
        return nullptr;

    chassert(order_by_indices.size() == 1);
    const IColumn & column = *header.getByPosition(order_by_indices[0]).column;
    const bool missing_is_greater = order_by[0].nulls_direction > 0;

The incremental update

Here is the heart of it. After the frame bounds have moved, the aggregate state is updated — but only by the difference:

cpp 
// src/Processors/Transforms/WindowTransform.cpp:169
    // We might have to reset aggregation state and/or add some rows to it.
    // Figure out what to do.
    bool reset_aggregation = false;
    RowNumber rows_to_add_start;
    RowNumber rows_to_add_end;
    if (current_frame.start.row_index_in_partition == prev_frame.start.row_index_in_partition)
    {
        // The frame start didn't change, add the tail rows.
        reset_aggregation = false;
        rows_to_add_start = prev_frame.end.location;
        rows_to_add_end = current_frame.end.location;
    }
    else
    {
        // The frame start changed, reset the state and aggregate over the
        // entire frame. This can be made per-function after we learn to
        // subtract rows from some types of aggregation states, but for now we
        // always have to reset when the frame start changes.
        reset_aggregation = true;
        rows_to_add_start = current_frame.start.location;
        rows_to_add_end = current_frame.end.location;
    }

The comment is honest about the limitation: ClickHouse’s aggregate states cannot have rows removed from them, so a frame whose start moved backwards (or forwards) has to be recomputed from scratch. But the common case — the default frame RANGE UNBOUNDED PRECEDING TO CURRENT ROW — has a fixed start, so only the tail is added, every time.

The rows are then fed to the aggregate in a batch, block by block:

cpp 
// src/Processors/Transforms/WindowTransform.cpp:217
        for (auto block_number = rows_to_add_start.block;
             block_number < past_the_end_block;
             ++block_number)
        {
            const auto & block = blocks.blockAt(block_number);

            if (ws.cached_block_number != block_number)
            {
                for (size_t i = 0; i < ws.argument_column_indices.size(); ++i)
                {
                    ws.argument_columns[i] = block.materialized_columns[
                        ws.argument_column_indices[i]].get();
                }
                ws.cached_block_number = block_number;
            }

The first and last block may be partial, the middle ones are whole, and the aggregate is fed a row range rather than a list of rows:

cpp 
// src/Processors/Transforms/WindowTransform.cpp:233
            // First and last blocks may be processed partially, and other blocks
            // are processed in full.
            const auto first_row = block_number == rows_to_add_start.block
                ? rows_to_add_start.row : 0;
            const auto past_the_end_row = block_number == rows_to_add_end.block
                ? rows_to_add_end.row : block.rows_count;

The frame that did not move

There is one more optimisation, and it is a big one for the default frame. With RANGE UNBOUNDED PRECEDING TO CURRENT ROW and an ORDER BY on a column with many duplicates, many consecutive rows share the same frame. The result for all of them is identical.

So the transform checks whether the frame moved, and if it did not, copies the previous row’s result instead of finalising the aggregate again:

cpp 
// src/Processors/Transforms/WindowTransform.cpp:275
        if (frame_unchanged && !ws.is_aggregate_function_state && peers.current().location.row > 0)
        {
            // Same frame as the previous row -> same result. When that row is in this same block its
            // result is already in result_column one position back, so copy it instead of
            // re-finalizing. We copy the column into itself with insertRangeFrom (not insertFrom):
            // insertRangeFrom appends via resize + memcpy from a disjoint source range, which is
            // self-safe even if the append reallocates and even for nested columns (Array, Variant,
            // Dynamic, JSON) whose sub-columns are not covered by the top-level reserve.
            chassert(std::cmp_equal(result_column->size(), peers.current().location.row));
            result_column->insertRangeFrom(*result_column, peers.current().location.row - 1, 1);
        }

The condition for “the frame did not move” is that both the start and the end are the same as the previous row’s:

cpp 
// src/Processors/Transforms/WindowTransform.cpp:257
    const FrameBounds & current_frame = frame.bounds();
    const bool frame_unchanged = peers.current().row_index_in_partition > 0 && current_frame.start.row_index_in_partition == prev_frame.start.row_index_in_partition && current_frame.end.row_index_in_partition == prev_frame.end.row_index_in_partition;

Note that copying the column into itself has to be done carefully: insertRangeFrom appends via resize + memcpy from a disjoint source range, so it is safe even if the append reallocates the buffer. A naive insertFrom on the same column would be reading from memory that the resize may have just moved.

True window functions do not use the aggregate at all

row_number, rank, dense_rank, lag, lead and friends are not aggregates at all — they are computed from the position and the peer bounds. They implement a separate interface:

cpp 
// src/WindowFunctions/IWindowFunction.h:23
class IWindowFunction
{
public:
    virtual ~IWindowFunction() = default;

    // Must insert the result for current_row.
    virtual void windowInsertResultInto(const WindowTransform * transform, size_t function_index) const = 0;

    virtual std::optional<WindowFrame> getDefaultFrame() const { return {}; }

    /// Is the frame type supported by this function.
    virtual bool checkWindowFrameType(const WindowTransform * /*transform*/) const { return true; }
};

rank is then a one-liner that reads the start of the peer group:

cpp 
// src/WindowFunctions/WindowFunctionRank.cpp:33
    void windowInsertResultInto(const WindowTransform * transform,
        size_t function_index) const override
    {
        IColumn & to = *transform->blocks.blockAt(transform->peers.current().location.block)
            .result_columns[function_index];
        assert_cast<ColumnUInt64 &>(to).getData().push_back(
            transform->peers.bounds().start.row_index_in_partition + 1);
    }

and row_number reads the current position:

cpp 
// src/WindowFunctions/WindowFunctionRowNumber.cpp:22
    void windowInsertResultInto(const WindowTransform * transform,
        size_t function_index) const override
    {
        IColumn & to = *transform->blocks.blockAt(transform->peers.current().location.block)
            .result_columns[function_index];
        assert_cast<ColumnUInt64 &>(to).getData().push_back(
            transform->peers.current().row_index_in_partition + 1);
    }

The difference between the two is exactly the difference between the “with gaps” and “without gaps” semantics: rank is the first row of the peer group, row_number is the row itself.

The streaming loop

Now the part that makes this hard: the transform is a streaming processor, but the frame’s end may not be knowable yet.

The frame’s end for RANGE CURRENT ROW depends on where the next different ORDER BY value starts, and that may be in a block that has not arrived yet. So the frame carries a fully_visible flag, and if it is not set, the transform waits for more input:

cpp 
// src/Processors/Transforms/WindowTransform.cpp:339
            if (!frame.bounds().fully_visible)
            {
                // Wait for more input data to find the frame.
                chassert(!input_is_finished);
                chassert(!partition.bounds().fully_visible);
                return;
            }

And the same applies to the partition’s end: the transform cannot start the next partition until it knows where the current one ends.

cpp 
// src/Processors/Transforms/WindowTransform.cpp:315
void WindowTransform::computeReadyRows()
{
    for (;;)
    {
        // Either we ran out of data or we found the end of partition (maybe
        // both, but this only happens at the total end of data).
        const RowNumber partition_end = partition.bounds().end.location;
        chassert(partition.bounds().fully_visible || partition_end == blocks.end());

The main loop is: advance the peer bounds, advance the frame, update the aggregate, write the row, move to the next row.

cpp 
// src/Processors/Transforms/WindowTransform.cpp:331
        while (peers.current().location < partition_end)
        {
            peers.advancePeerGroupBounds(blocks, partition.bounds());
            checkInvariants(blocks, next_output_block_number, partition.bounds(), peers, frame.bounds(), prev_frame);

            frame.advance(blocks, peers.current(), partition.bounds());
            checkInvariants(blocks, next_output_block_number, partition.bounds(), peers, frame.bounds(), prev_frame);

The checkInvariants calls are no-ops in release builds, and they are quite specific — they assert, among other things, that the frame never moves backwards and that the previous frame’s start is never after the current one:

cpp 
// src/Processors/Transforms/WindowTransform.cpp:81
    chassert(partition.start.location <= prev_frame.start.location);
    chassert(prev_frame.start.location <= prev_frame.end.location);
    chassert(prev_frame.end.location <= partition.end.location);
    chassert(prev_frame.start.row_index_in_partition <= prev_frame.end.row_index_in_partition);
    chassert(prev_frame.end.row_index_in_partition <= partition.end.row_index_in_partition);
    if (frame.fully_visible)
    {
        chassert(prev_frame.start.location <= frame.start.location);
        chassert(frame.start.location <= frame.end.location);
        chassert(prev_frame.end.location <= frame.end.location);

Releasing memory

Because the frame moves forward, the blocks behind it are dead weight. The transform drops them:

cpp 
// src/Processors/Transforms/WindowTransform.cpp:560
void WindowTransform::releaseUnusedBlocks()
{
    // We don't really have to keep the entire partition, and it can be big, so
    // we want to drop the starting blocks to save memory. We can drop the old
    // blocks if we already returned them as output, and the frame and the
    // current row are already past them. The previous frame start is never
    // after the current frame start, so we don't have to check the latter. Note
    // that the frame start can be further than current row for some frame specs
    // (e.g. EXCLUDE CURRENT ROW), so we have to check both.
    const auto first_used_block = std::min({next_output_block_number, prev_frame.start.location.block, peers.current().location.block});
    while (blocks.begin().block < first_used_block)
        blocks.pop();
}

The three-way minimum is worth reading twice. A block can be released when it has been output (next_output_block_number), when the frame has moved past it (prev_frame.start), and when the current row has moved past it (peers.current). The comment notes the EXCLUDE CURRENT ROW case, where the frame start can be after the current row — a frame that excludes the current row would otherwise pin a block in memory.

The arena is reset per partition, so a partition’s aggregate states do not accumulate memory across partitions:

cpp 
// src/Processors/Transforms/WindowTransform.cpp:429
    // Replace the arena so that it does not grow across partitions. All states
    // were destroyed above and no result lives in it, see the field comment.
    if (arena)
    {
        arena = std::make_unique<Arena>();
    }

What the aggregate state can and cannot do

The incremental trick has a hard limit that the code states plainly: an aggregate state cannot have a row removed from it. So the frame is recomputed from scratch whenever its start moves:

cpp 
// src/Processors/Transforms/WindowTransform.cpp:183
        // The frame start changed, reset the state and aggregate over the
        // entire frame. This can be made per-function after we learn to
        // subtract rows from some types of aggregation states, but for now we
        // always have to reset when the frame start changes.

In practice this matters for frames like ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING: the start moves on every row, so every row re-aggregates the whole frame — which is fine, because the frame is small. It would matter for ROWS BETWEEN 100 PRECEDING AND CURRENT ROW: the start moves on every row and the frame is large, so that frame is quadratic again.

sum() and count() could support subtraction; avg() could be derived from them; min()/max() cannot (you would need a multiset). The comment says this could be made per-function, and is not.

Trade-offs

What the design buys:

  • Linear time for the common frames. With a fixed frame start, each row is aggregated exactly once, ever.
  • A copy instead of a recomputation when the frame does not move. For ORDER BY columns with many duplicates, most rows are free.
  • Streaming memory. The transform keeps only the blocks the frame still needs, and drops the rest as it goes.
  • True window functions are nearly free. rank and row_number read the peer bounds directly; there is no aggregate state at all.

What it costs:

  • A full sort of the input. PARTITION BY + ORDER BY has to be materialised, which for a large input means a disk-backed sort.
  • Frames with a moving start are quadratic. ROWS BETWEEN N PRECEDING AND CURRENT ROW re-aggregates the whole frame for every row, because the aggregate states cannot have rows removed.
  • RANGE with an offset is a linear scan. The boundary search walks the key column row by row rather than binary-searching, because the boundary is a value and the search predicate is a shifted comparison.
  • The frame’s end may not be knowable. For RANGE CURRENT ROW, the transform must wait for the next block before it can finalise a row, which adds latency to the first rows of every partition.
  • The peer-group index costs 8 bytes per row. int64_t per row, for every block, even when the frame is a ROWS frame that does not need it.

Wrapping up

The one thing to remember: the window transform keeps the aggregate state alive between rows and only feeds it the rows that entered the frame. That is the entire difference between O(n²) and O(n).

The part people get wrong: the incremental update only works when the frame’s start is fixed. The default frame (RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) has a fixed start and is linear. A frame like ROWS BETWEEN 100 PRECEDING AND CURRENT ROW has a moving start and is quadratic again, and no setting changes that.

What to go do about it: if a window query is slow, look at the frame first, then at the sort. A window over an already-sorted table (optimize_read_in_order) skips the sort entirely. And if the frame is ROWS BETWEEN N PRECEDING AND CURRENT ROW with a large N, consider whether RANGE on a quantised key would give the same answer with a fixed start.

ClickhousePerformanceInternalsWindow functions