How the ClickHouse Buffer Engine Holds Writes in RAM and Flushes Them in the Background

TL;DR: Buffer keeps its rows in RAM, spread over num_layers independent shards, each shard guarded by its own mutex. A write either goes straight to the destination (if the block is already bigger than the max_* thresholds) or is appended to the least busy shard. A background task wakes up when the soonest threshold is due and flushes each shard by swapping its block out first — so if the destination is down, the block is swapped back in and retried later instead of being lost.

The classic ClickHouse deployment pattern is “a fleet of application servers inserts into ClickHouse”. Each server, in a burst, inserts a few hundred rows. A MergeTree does not care how many rows an INSERT has; it cares that each INSERT produces a part, and a part is a set of files that must be merged later. Ten thousand servers inserting one row a second produce ten thousand tiny parts a second, and the merge queue falls over.

The naive fix is to batch in the application. The ClickHouse-native fix is the Buffer engine: a table that lives in RAM, absorbs small INSERTs, and periodically writes them to a real table in big chunks. The interesting part is the flush policy — three families of thresholds, a scheduler that computes its own next wake-up time, and a failure mode that must be survivable.

The shape of the thing

The class comment in the header is the design document. The buffer is a set of num_shards blocks, and the rule for a busy shard is “try the next block in a round-robin fashion, and so no more than num_shards times (then wait for lock)”:

cpp
// src/Storages/StorageBuffer.h:28
  * When using a block, it is locked by some mutex. If during write the corresponding block is already occupied
  *  - try to lock the next block in a round-robin fashion, and so no more than `num_shards` times (then wait for lock).
  * Thresholds are checked on insertion, and, periodically, in the background thread (to implement time thresholds).
  * Thresholds act independently for each shard. Each shard can be flushed independently of the others.
  * If a block is inserted into the table, which itself exceeds the max-thresholds, it is written directly to the subordinate table without buffering.

Five facts fall out of that comment: the buffer is a set of shards, each with its own mutex; thresholds are per shard; a block bigger than the max thresholds bypasses the buffer entirely; there are three threshold families (min, max, flush); and the data is not durable — “with a rough restart of the server, the data is lost”.

A shard is remarkably small:

cpp
// src/Storages/StorageBuffer.h:155
    struct Buffer
    {
        time_t first_write_time = 0;
        Block data;

plus an int32_t metadata_version, which exists because an ALTER changes the column set and blocks written before and after it cannot be concatenated. The chosen policy is “skip stale buffers rather than convert them” — a data-visibility bug in exchange for not writing a schema migrator.

The engine arguments decode like this:

sql
CREATE TABLE merge.hits_buffer AS merge.hits
ENGINE = Buffer(merge, hits, 1, 10, 100, 10000, 1000000, 10000000, 100000000);

text
Buffer(merge, hits,
  num_layers = 1,                                    -- independent shards
  min_time  = 10,  max_time  = 100,                  -- seconds since the first write
  min_rows  = 10000, max_rows = 1000000,             -- rows in the shard
  min_bytes = 10000000, max_bytes = 100000000)       -- bytes in the shard

num_layers multiplies the RAM ceiling: with 16 layers and max_bytes = 100 000 000, the worst case is 1.6 GB. This is the single most common misconfiguration of the Buffer engine.

The write path

BufferSink::consume is where the policy lives. First, the bypass.

cpp
// src/Storages/StorageBuffer.cpp:897
        /// If the block already exceeds the maximum limit, then we skip the buffer.
        if (rows > storage.max_thresholds.rows || bytes > storage.max_thresholds.bytes)
        {
            if (destination)
            {
                LOG_DEBUG(storage.log, "Writing block with {} rows, {} bytes directly.", rows, bytes);
                storage.writeBlockToDestination(block, destination);
            }
            return;
        }

This is what makes Buffer safe to point at a bulk loader: a 10-million-row INSERT does not sit in RAM waiting for a threshold that will never be reached. The cost is that the destination sees one huge INSERT.

Then the shard selection. The obvious implementation is “hash the thread id to a shard”; the code instead tries every shard with a non-blocking lock and picks the one with the fewest rows:

cpp
// src/Storages/StorageBuffer.cpp:908
        /// We distribute the load on the shards by the stream number.
        const auto start_shard_num = getThreadId() % storage.num_shards;

        /// We loop through the buffers, trying to lock mutex. No more than one lap.
        auto shard_num = start_shard_num;

        StorageBuffer::Buffer * least_busy_buffer = nullptr;
        std::unique_lock<std::mutex> least_busy_lock;
        size_t least_busy_shard_rows = 0;

        for (size_t try_no = 0; try_no < storage.num_shards; ++try_no)
        {
            std::unique_lock lock(storage.buffers[shard_num].tryLock());

            if (lock.owns_lock())
            {
                size_t num_rows = storage.buffers[shard_num].data.rows();
                if (!least_busy_buffer || num_rows < least_busy_shard_rows)
                {
                    least_busy_buffer = &storage.buffers[shard_num];
                    least_busy_lock = std::move(lock);
                    least_busy_shard_rows = num_rows;
                }
            }

            shard_num = (shard_num + 1) % storage.num_shards;
        }

Three properties fall out of this loop: the starting shard is thread-dependent, so two simultaneous writers begin their scan at different shards; the lock is held for the whole scan, so the “least busy” measurement is not a race — the writer that picks a shard also owns it; and the scan is exactly one lap.

If every shard is locked (all of them are being flushed at once), the writer blocks on its own starting shard, appends, unlocks, and calls storage.reschedule(0). Note that last call: the background task does not run on a fixed timer — it is re-armed by the writes themselves, and the “0” means “schedule me as soon as possible if a threshold is due”.

The append itself takes column checkpoints first, because appending a block to a block is a per-column insertRangeFrom, and a MEMORY_LIMIT_EXCEEDED in the middle of column 7 of 10 would leave a corrupt buffer. On failure, each column is rolled back to its checkpoint. The block’s columns are also sorted before being appended, so that concatenating two buffered blocks is a cheap insertRangeFrom rather than a column-by-column reshuffle.

The threshold algebra

Here is the part people get wrong when they read the docs. There are three families of thresholds, and they mean different things.

cpp
// src/Storages/StorageBuffer.cpp:1152
bool StorageBuffer::checkThresholdsImpl(bool direct, size_t rows, size_t bytes, time_t time_passed) const
{
    if (time_passed > min_thresholds.time && rows > min_thresholds.rows && bytes > min_thresholds.bytes)
    {
        ProfileEvents::increment(ProfileEvents::StorageBufferPassedAllMinThresholds);
        return true;
    }

    if (time_passed > max_thresholds.time)
    {
        ProfileEvents::increment(ProfileEvents::StorageBufferPassedTimeMaxThreshold);
        return true;
    }

    if (rows > max_thresholds.rows)
    {
        ProfileEvents::increment(ProfileEvents::StorageBufferPassedRowsMaxThreshold);
        return true;
    }

    if (bytes > max_thresholds.bytes)
    {
        ProfileEvents::increment(ProfileEvents::StorageBufferPassedBytesMaxThreshold);
        return true;
    }

Read the boolean structure:

min_* is an AND of all three. Time, rows and bytes must all be exceeded: the “the buffer is full” condition, which only fires when the shard is genuinely full on every axis.
max_* is an OR of any of the three. Time, rows or bytes being exceeded is enough: the “I cannot let this grow any further” safety valve.
flush_* is checked only for non-direct flushes (the !direct guard, plus “the threshold is non-zero”, so an omitted flush_* never fires). This is the “background worker, wake up” condition, and it exists so that the INSERT path does not block on a destination write.

That last distinction is why flush_* exists. With max_rows = 1e6 and flush_rows = 500e3, the shard is only forced to flush at 1e6 rows, but the background worker will opportunistically flush at 500e3 rows — and because that happens on another thread, the INSERT does not pay the latency of the destination write.

The direct flag is true on the insert path, where the flush threshold is deliberately ignored:

cpp
// src/Storages/StorageBuffer.cpp:958
        if (storage.checkThresholds(buffer, /* direct= */true, current_time, sorted_block.rows(), sorted_block.bytes()) ||
            buffer.metadata_version != metadata_version)

and false on the background worker:

cpp
// src/Storages/StorageBuffer.cpp:1245
        if (!checkThresholdsImpl(/* direct= */false, rows, bytes, time_passed))
            return false;

One more subtlety: checkThresholds is called before the new block is appended, with the incoming row and byte counts added in. The check is a prediction — “would the shard exceed the threshold if I add these rows?” — not an observation, and the threshold can still be exceeded: if max_rows = 1 000 000 and the buffer already had 500 000 rows, adding 800 000 rows leaves 1 300 000 in the buffer. The engine does not split the incoming block to respect the threshold; it accepts the overshoot and then flushes.

The background worker

The background flush is a BackgroundSchedulePool task:

cpp
// src/Storages/StorageBuffer.cpp:196
    flush_handle = bg_pool->createTask(getStorageID(), log->name() + "/Bg", [this]{ backgroundFlush(); });

Its body: flush everything, swallow any exception, re-arm itself.

cpp
// src/Storages/StorageBuffer.cpp:1383
void StorageBuffer::backgroundFlush()
{
    try
    {
        LOG_DEBUG(log, "Running background flush of buffers.");
        flushAllBuffers(true);
    }
    catch (...)
    {
        tryLogCurrentException(__PRETTY_FUNCTION__);
    }

The catch (...) is the “the destination is down” behaviour. What happens to the data is in flushBuffer, not here.

Computing the next wake-up

The scheduler does not poll every second: it computes the earliest moment at which any threshold could possibly fire, and sleeps until then:

The loop over the shards uses tryLock() rather than a blocking lock — “to avoid waiting for other layers flushing to be finished, since the buffer table may push to a Distributed table, that may take too much time” — and collects two things: the oldest first_write_time among the non-empty shards, and the total row count.

The oldest non-empty shard sets the clock: time_passed = now - min_first_write_time is how long the oldest data has been waiting, and the three candidate wake-up times are “threshold minus time already passed”:

cpp
// src/Storages/StorageBuffer.cpp:1434
    time_t current_time = time(nullptr);
    time_t time_passed = current_time - min_first_write_time;

    /// checkThresholdsImpl() uses strict comparison (> not >=), so we need to add offset to original time values
    static constexpr time_t THRESHOLD_COMPARISON_OFFSET = 1;

    /// For minimal threshold min_delay is ignored, since otherwise it will be triggered too frequently, once it is reached
    /// (while the Buffer cannot be flushed due to other min thresholds).
    size_t min = std::max<ssize_t>(min_thresholds.time + THRESHOLD_COMPARISON_OFFSET - time_passed, 1);
    size_t max = std::max<ssize_t>(max_thresholds.time + THRESHOLD_COMPARISON_OFFSET - time_passed, min_delay);
    size_t reschedule_sec = 0;
    if (flush_thresholds.time)
    {
        size_t flush = std::max<ssize_t>(flush_thresholds.time + THRESHOLD_COMPARISON_OFFSET - time_passed, min_delay);
        reschedule_sec = std::min({min, max, flush});
    }

Three things to notice:

The +1 offset exists because the comparison in checkThresholdsImpl is strict (> not >=). Without it, a 100-second threshold would be considered reached at 99 seconds, and the worker would wake up one second early, every time.
The min threshold deliberately ignores min_delay, because the min_* condition is an AND: waking up when only the time part of it is satisfied accomplishes nothing.
The wake-up is the minimum of the three candidates, so whichever threshold fires first wins.

And if there is nothing buffered at all, no reschedule happens at all: an idle Buffer table costs zero scheduler wake-ups, and the task is only re-armed by an INSERT.

The failure mode: the destination is down

This is the interesting part: a two-line trick with a large consequence.

flushBuffer does not write in place. It swaps the buffer’s block out into a local variable, and only then writes it:

cpp
// src/Storages/StorageBuffer.cpp:1249
    buffer.data.swap(block_to_write);
    buffer.first_write_time = 0;

swap is O(1) — it exchanges the contents of two Blocks. After the swap the shard is empty and the data lives in the local variable. The shard stays locked for the duration of the write, which the code justifies:

cpp
// src/Storages/StorageBuffer.cpp:1270
    /** For simplicity, buffer is locked during write.
        * We could unlock buffer temporary, but it would lead to too many difficulties:
        * - data, that is written, will not be visible for SELECTs;
        * - new data could be appended to buffer, and in case of exception, we must merge it with old data, that has not been written;
        * - this could lead to infinite memory growth.
        */

And if the write throws — the destination table is gone, the disk is full, the network is down — the block is swapped back into the shard and the exception is rethrown:

cpp
// src/Storages/StorageBuffer.cpp:1282
    catch (...)
    {
        ProfileEvents::increment(ProfileEvents::StorageBufferErrorOnFlush);

        /// Return the block to its place in the buffer.

        CurrentMetrics::add(CurrentMetrics::StorageBufferRows, block_to_write.rows());
        CurrentMetrics::add(CurrentMetrics::StorageBufferBytes, block_to_write.bytes());

        buffer.data.swap(block_to_write);

        if (!buffer.first_write_time)
            buffer.first_write_time = current_time;

        /// After a while, the next write attempt will happen.
        throw;
    }

The first_write_time is restored too, so the age of the data — which is what the time thresholds measure — survives the failed attempt. The data is not lost and the next background tick will try again.

Combined with the catch (...) in backgroundFlush, the behaviour is:

text
   INSERT -> shard buffer (RAM)
                 |  thresholds: min (AND) | max (OR) | flush (background only)
                 v
        backgroundFlush()  --(exception swallowed, rescheduled)
                 v
        flushBuffer():  swap out --> write to destination
                            ^                 |
                            +--- swap back -- (exception)

One caveat that the code itself flags: the swap-back assumes the destination write failed before writing anything. If the destination wrote part of the block and then failed, the data is written twice — once in the destination, once in the buffer. There is no transaction spanning the two, which is why the engine’s documentation warns that “it is not possible to have a reliable ’exactly once’ write to replicated tables.”

Trade-offs

What the trick buys: a MergeTree is fed large parts instead of a hundred thousand tiny ones, and the INSERT path never blocks on the destination. The threshold algebra gives three separate knobs — “flush when full”, “never let this grow past that”, “and do it in the background so my INSERT stays fast”.

What it costs:

The data is not durable. It is in RAM, in a process, and “with a rough restart of the server, the data is lost”. A Buffer table is a performance device in front of a durable table, not a queue; the engine’s documentation now recommends asynchronous inserts instead.
The order is destroyed. Rows arrive in the destination in whatever order the shards flushed in, and in whatever block sizes the shards happened to have. This breaks CollapsingMergeTree (hence the documentation’s advice to use num_layers = 1) and it breaks deduplication on replicated tables.
A failed flush can duplicate. The swap-back is only safe if the destination failed atomically; there is no 2-phase commit between the buffer and the destination.
The thresholds are per shard. max_bytes = 100 000 000 with 16 layers is 1.6 GB, not 100 MB. And because min_* is an AND of three axes, a shard that never reaches the row count is only ever flushed by a max_* or flush_* threshold, or by the server shutting down.

Wrapping up

The one thing to remember: the Buffer engine is a RAM buffer with a three-tier threshold algebra — min_* (all of time, rows, bytes), max_* (any of them), flush_* (background only) — evaluated per shard.

The part people get wrong: it is not durable and it is not ordered. If you need at-least-once delivery from a queue, use Kafka; if you need to avoid tiny parts, use Buffer or asynchronous inserts.

What to go do about it: insert one row at a time into a Buffer table and watch which threshold is actually doing the work in your deployment.

sql
SELECT event, count
FROM system.events
WHERE event LIKE 'StorageBuffer%'
ORDER BY event;