ClickHouse Buffers I/O with a pos()/end() Invariant

TL;DR: Every byte ClickHouse reads or writes passes through a ReadBuffer or a WriteBuffer: a cursor pos into a working_buffer that ends at end(). The buffer is drained in one memcpy per call and refilled by exactly one virtual function, nextImpl(). On top of that sits PeekableReadBuffer, which can checkpoint the cursor and roll back to it by copying the consumed bytes into private memory.

Every I/O-heavy program eventually grows a buffering layer, and every one of them reinvents the same handful of ideas. ClickHouse’s is worth reading because it is unusually explicit about what the invariant is, and because the same three-line invariant is reused for files, sockets, S3 objects, decompression streams, and format parsers.

The starting point is a complaint about std::istream that has not changed in fifteen years:

cpp
// src/IO/BufferBase.h:14
  * ReadBuffer and WriteBuffer are similar to istream and ostream, respectively.
  * They have to be used, because using iostreams it is impossible to effectively implement some operations.
  * For example, using istream, you can not quickly read string values from a tab-separated file,
  *  so that after reading, the position remains immediately after the read value.
  * (The only option is to call the std::istream::get() function on each byte, but this slows down due to several virtual calls.)

That is the whole motivation. istream::get() is a virtual call per byte. If you want to parse a CSV file, you want to look at the bytes in the buffer and decide where a field ends — not call get() 400 million times and hope the optimizer saves you. So the design goal is: give the caller direct access to the raw bytes, and make the common operation (copy a run of bytes out of the buffer) a memcpy.

The invariant

Everything is built on BufferBase, which holds three things and one cursor:

cpp
// src/IO/BufferBase.h:59
    BufferBase(Position ptr, size_t size, size_t offset)
        : pos(ptr + offset), working_buffer(ptr, ptr + size), internal_buffer(ptr, ptr + size) {}

Two buffers, not one. internal_buffer is the memory you own; working_buffer is the part of it that currently holds valid data. The distinction matters because a read from a file can return 10 bytes into a 1 MiB allocation, and code that scans for a delimiter must not run off the end into garbage.

The cursor and the two derived quantities:

cpp
// src/IO/BufferBase.h:81
    /// offset in bytes of the cursor from the beginning of the buffer
    size_t offset() const { return size_t(pos - working_buffer.begin()); }

    /// How many bytes are available for read/write
    size_t available() const { return size_t(working_buffer.end() - pos); }

cpp
// src/IO/BufferBase.h:94
    /** How many bytes have been read/written, counting those that are still in the buffer. */
    size_t count() const { return bytes + offset(); }

    /** Check that there is more bytes in buffer after cursor. */
    bool ALWAYS_INLINE hasPendingData() const { return available() > 0; }

So the entire contract of every buffer in ClickHouse is:

text
  internal_buffer  [ begin() ................. end() )   <- memory you own (1 MiB)
                    ^
                    |
  working_buffer   [ begin() .................. end() )   <- the valid window
                                pos
                                |
                                v  available() = end() - pos   bytes still to read

pos always lies in [working_buffer.begin(), working_buffer.end()]. That is asserted on every refill:

cpp
// src/IO/ReadBuffer.cpp:103
bool ReadBuffer::next()
{
    chassert(!hasPendingData());
    chassert(position() <= working_buffer.end());
    chassert(!isCanceled(), "ReadBuffer is canceled. Can't read from it.");

    bytes += offset();
    bool res = false;
    try
    {
        res = nextImpl();
    }
    catch (...)
    {
        cancel();
        throw;
    }

Note bytes += offset(): the count of bytes consumed is separated from the count of bytes buffered. count() adds them back together, so it is the logical stream position even when the buffer holds 1 MiB of unread data.

Draining the buffer

A derived class implements exactly one thing: nextImpl(), “give me more data in working_buffer”. Here is the file implementation:

cpp
// src/IO/ReadBufferFromFileDescriptor.cpp:162
bool ReadBufferFromFileDescriptor::nextImpl()
{
    /// If internal_buffer size is empty, then read() cannot be distinguished from EOF
    chassert(!internal_buffer.empty());

    size_t bytes_read = readImpl(internal_buffer.begin(), 1, internal_buffer.size(), file_offset_of_buffer_end);

    file_offset_of_buffer_end += bytes_read;

    if (bytes_read)
    {
        working_buffer = internal_buffer;
        working_buffer.resize(bytes_read);
    }
    else
        return false;

    return true;
}

readImpl is a pread/read loop that may return fewer bytes than asked for; the working buffer is resized to exactly what arrived. Everything above this layer then sees a simple “N bytes available” world.

The read side is a loop that copies out of the window until it is empty or the request is satisfied:

cpp
// src/IO/ReadBuffer.h:165
    /** Reads as many as there are, no more than n bytes. */
    [[nodiscard]] size_t read(char * to, size_t n)
    {
        size_t bytes_copied = 0;

        while (bytes_copied < n && !eof())
        {
            size_t bytes_to_copy = std::min(static_cast<size_t>(working_buffer.end() - pos), n - bytes_copied);
            ::memcpy(to + bytes_copied, pos, bytes_to_copy);
            pos += bytes_to_copy;
            bytes_copied += bytes_to_copy;
        }

        return bytes_copied;
    }

The min with working_buffer.end() - pos is the invariant in action: you can never read past the valid window, and when the window runs dry the loop calls eof(), which refills.

`eof()` has a side effect, and that is the point

cpp
// src/IO/ReadBuffer.h:91
    bool ALWAYS_INLINE eof()
    {
        return !hasPendingData() && !next();
    }

eof() is not a pure query. If the buffer is drained, it refills and then answers. This is what lets the read loop above be written as “copy until eof() says stop”, and it means the check for end-of-stream is amortised: one next() (a syscall) per 1 MiB consumed, not one per byte.

The alternative — isEof() as a pure predicate plus an explicit next() — is what std::istream does, and it is why std::istream is slow: every caller must remember to check and to advance, and the two operations cannot be inlined into one branch.

Note also that the read side is pull-based with a pushback-free design: there is no unget. If you want to look ahead and then re-read, you need a different class. Which brings us to the peek buffer.

The write side

WriteBuffer is the mirror image, with one extra idea: a flush threshold. Data accumulates in the working buffer until next() is called, which pushes the accumulated bytes to the underlying sink and resets the cursor.

cpp
// src/IO/WriteBuffer.h:44
    void next()
    {
        if (canceled)
            return;

        if (!offset())
            return;

        auto bytes_in_buffer = offset();

        try
        {
            nextImpl();
            ++flush_count;
        }
        catch (CurrentBufferExhausted &)
        {
            pos = working_buffer.begin();
            bytes += bytes_in_buffer;
            throw;
        }
        catch (...)
        {
            /** If the nextImpl() call was unsuccessful, move the cursor to the beginning,
              * so that later (for example, when the stack was expanded) there was no second attempt to write data.
              */
            pos = working_buffer.begin();
            bytes += bytes_in_buffer;

            cancel();

            throw;
        }

        bytes += bytes_in_buffer;
        pos = working_buffer.begin() + nextimpl_working_buffer_offset;
        nextimpl_working_buffer_offset = 0;
    }

Two details worth stealing:

On failure, the cursor rewinds to the beginning of the working buffer before the exception propagates. The bytes are still there, bytes still counts them as pending, and the buffer is marked canceled so that no later code will try to write again. A partially-written file is detected rather than silently produced.
nextimpl_working_buffer_offset lets a derived class say “the data I just wrote occupies only the first N bytes of what I was given” — used by buffers that consume a prefix for their own purposes.

The hot path is the one every format writer uses — copy into the window, and only fall back to the slow path when the window is full:

cpp
// src/IO/WriteBuffer.h:95
    void write(const char * from, size_t n)
    {
        /// `closed` is checked first: after `finalize` some buffers reset their storage to null (`set(nullptr, 0)`),
        /// and the pointer arithmetic in `available` must not be evaluated then.
        /// The comparison is unsigned as in `available`: for a constant `n` it is a single `cmp`.
        if (!closed && n <= available()) [[likely]]
        {
            copyBytes(pos, from, n);
            pos += n;
            return;
        }
        writeSlow(from, n);
    }

And the copy itself is specialised at compile time when possible:

cpp
// src/IO/WriteBuffer.h:191
    /// A size known at compile time becomes plain moves. A run-time size is copied with `memcpy`, which keeps
    /// the inlined code small: a larger fast path changes how its callers are optimized (inlining, loop unswitching).
    /// Zero bytes, which is common for empty strings and empty delimiters, does not call `memcpy` at all.
    ALWAYS_INLINE static void copyBytes(char * __restrict to, const char * __restrict from, size_t n)
    {
        if (__builtin_constant_p(n))
            __builtin_memcpy(to, from, n);
        else if (n != 0)
            memcpy(to, from, n);
    }

writeSlow is where the flush happens — and it is a loop, because a single write() may span several buffer refills:

cpp
// src/IO/WriteBuffer.cpp:48
void WriteBuffer::writeSlow(const char * from, size_t n)
{
    if (finalized)
        throwWriteToFinalizedBuffer();

    if (canceled)
        throwWriteToCanceledBuffer(ErrorCodes::CANNOT_WRITE_AFTER_BUFFER_CANCELED);

    size_t bytes_copied = 0;

    /// Produces endless loop
    chassert(!working_buffer.empty());

    while (bytes_copied < n)
    {
        nextIfAtEnd();
        size_t bytes_to_copy = std::min(static_cast<size_t>(working_buffer.end() - pos), n - bytes_copied);
        memcpy(pos, from + bytes_copied, bytes_to_copy);
        pos += bytes_to_copy;
        bytes_copied += bytes_to_copy;
    }
}

nextIfAtEnd() is the flush trigger: if there is no room left, call next(), which pushes the accumulated bytes downstream. So the “flush threshold” is not a byte count — it is simply the buffer is full. The buffer size is the threshold.

Where the bytes actually go

For a file, nextImpl() is a write(2) loop that handles partial writes:

cpp
// src/IO/WriteBufferFromFileDescriptor.cpp:45
void WriteBufferFromFileDescriptor::nextImpl()
{
    if (!offset())
        return;

    Stopwatch watch;

    size_t bytes_written = 0;
    while (bytes_written != offset())
    {
        ProfileEvents::increment(ProfileEvents::WriteBufferFromFileDescriptorWrite);

        ssize_t res = 0;
        {
            CurrentMetrics::Increment metric_increment{CurrentMetrics::Write};
            res = ::write(fd, working_buffer.begin() + bytes_written, offset() - bytes_written);
        }

Note the loop condition: write(2) may write less than requested (a signal interrupts it, the pipe is full), so the loop keeps going until the whole window is out. And bytes_written is not reset, so the next call to nextImpl() picks up where it left off — but next() has already rewound pos to working_buffer.begin(), so the accounting is: offset() bytes are pending, bytes_written bytes of them are done.

Sizing the buffer

The default buffer is 1 MiB:

cpp
// src/Core/Defines.h:23
static constexpr auto DBMS_DEFAULT_BUFFER_SIZE = 1048576ULL;

That is the fallback. The size actually used for reading a file comes from a setting, and the local-filesystem case has its own smaller default:

cpp
// src/Core/Settings.cpp:445
    DECLARE(NonZeroUInt64, max_read_buffer_size, DBMS_DEFAULT_BUFFER_SIZE, R"(
The maximum size of the buffer to read from the filesystem. Values above 256 MiB are clamped to 256 MiB, as a read buffer never needs to be larger.
)", 0) \
    DECLARE(UInt64, max_read_buffer_size_local_fs, 128*1024, R"(
The maximum size of the buffer to read from local filesystem. If set to 0 then max_read_buffer_size will be used. Values above 256 MiB are clamped to 256 MiB, as a read buffer never needs to be larger.
)", 0) \

so a query reading a local file gets a 128 KiB buffer, and one reading from S3 gets 1 MiB, unless told otherwise:

sql
SELECT name, value FROM system.settings
WHERE name IN ('max_read_buffer_size', 'max_read_buffer_size_local_fs');

text
┌──────────────────────────────┬────────┐
│ name                         │ value  │
├──────────────────────────────┼────────┤
│ max_read_buffer_size         │ 1048576│
│ max_read_buffer_size_local_fs│ 131072 │
└──────────────────────────────┴────────┘

The “clamped to 256 MiB” note is a design opinion worth repeating: a read buffer larger than 256 MiB is never useful, because the point of the buffer is to amortise the syscall, and 256 MiB of amortisation is already absurd.

Every ReadBufferFromFile and WriteBufferFromFile constructor defaults to it. The allocation is padded for SIMD:

cpp
// src/IO/BufferWithOwnMemory.h:31
    static constexpr size_t pad_right = PADDING_FOR_SIMD - 1;

and PADDING_FOR_SIMD is 64, which means a parser may safely read 64 bytes past the end of the logical data using vector loads without a page fault. This is not a micro-optimisation; it is what makes the SSE2/NEON scanning loops in the CSV reader legal at all.

But 1 MiB per buffer is a lot when you have thousands of them. A wide table with 200 columns, each with its own column stream, means 200 write buffers. So the buffer can start small and grow:

cpp
// src/IO/BufferWithOwnMemory.h:128
    void enableAdaptiveBufferGrowth(bool use_adaptive, size_t max_size)
    {
        use_adaptive_buffer_size = use_adaptive;
        adaptive_buffer_max_size = max_size;
    }

    /// Call at the end of nextImpl; grows only when the flush used the whole buffer.
    void growAdaptiveBufferAfterFlush()
    {
        if (!Base::available() && use_adaptive_buffer_size && memory.size() < adaptive_buffer_max_size)
        {
            memory.resize(std::min(memory.size() * 2, adaptive_buffer_max_size));
            this->BufferBase::set(memory.data(), memory.size(), 0);
        }
    }

The growth condition is !Base::available() — the buffer was completely full at flush time. A buffer that is 90% full every time never grows, which is the right answer: it means 1 MiB is too big for this workload, so the initial size was wrong. A buffer that is 100% full every time doubles, until it hits the cap.

Binary helpers and the endianness assumption

The binary writers are almost embarrassingly simple, because the on-disk format is defined as native:

cpp
// src/IO/WriteHelpers.h:80
/// Write POD-type in native format. It's recommended to use only with packed (dense) data types.
template <typename T>
inline void writePODBinary(const T & x, WriteBuffer & buf)
{
    if constexpr (sizeof(T) <= MAX_INLINE_WRITE_SIZE)
        writeNBytes(reinterpret_cast<const char *>(&x), sizeof(x), buf); /// NOLINT
    else
        buf.write(reinterpret_cast<const char *>(&x), sizeof(x)); /// NOLINT
}

Native format. On every platform ClickHouse supports, that is little-endian. The explicit endian-converting variants exist for the formats that need them:

cpp
// src/IO/WriteHelpers.h:1719
template <typename T>
inline void writeBinaryBigEndian(T x, WriteBuffer & buf)
{
    writeBinaryEndian<std::endian::big>(x, buf);
}

and writeBinaryEndian byte-swaps in place before handing the value to the same native writer. So the default is “do nothing”, and the rare format that needs a different byte order pays for it explicitly.

Note MAX_INLINE_WRITE_SIZE:

cpp
// src/IO/WriteHelpers.h:76
/// Largest value for which __builtin_memcpy still beats a call. Above it clang emits a memcpy call
/// anyway, so the inline path would only add branches.
constexpr size_t MAX_INLINE_WRITE_SIZE = 64;

Below 64 bytes the compiler expands the copy inline; above it, it emits a call to memcpy anyway, so the inline path would only add branches.

Peeking: the checkpoint/rollback trick

Now the interesting part. Many parsers need to look at bytes before committing to a decision: is this a quoted field? Does this row end here? What is the header? ReadBuffer has no unget, because rewinding a stream is impossible.

PeekableReadBuffer solves this with a checkpoint. Set a checkpoint at the current position, read as far as you like, and roll back to the checkpoint later. The trick is that the bytes between the checkpoint and the current position are copied into the peek buffer’s own memory, so the rollback is a pointer move, not a re-read:

cpp
// src/IO/PeekableReadBuffer.h:32
    /// Sets checkpoint at current position
    ALWAYS_INLINE inline void setCheckpoint()
    {
        if (canceled)
            throw Exception(ErrorCodes::LOGICAL_ERROR, "Attempt to set a checkpoint on a canceled buffer");

        if (checkpoint)
        {
            /// Recursive checkpoints. We just remember offset from the
            /// first checkpoint to the current position.
            recursive_checkpoints_offsets.push(offsetFromCheckpoint());
            return;
        }

        checkpoint_in_own_memory = currentlyReadFromOwnMemory();
        if (!checkpoint_in_own_memory)
        {
            /// Don't need to store unread data anymore
            peeked_size = 0;
        }
        checkpoint.emplace(pos);
    }

The state machine has two modes: reading from the sub-buffer (the normal case, zero copy) and reading from own memory (the case where bytes between the checkpoint and the cursor had to be preserved). currentlyReadFromOwnMemory() is literally “is the working buffer’s begin different from the sub-buffer’s begin?”:

cpp
// src/IO/PeekableReadBuffer.h:99
    bool useSubbufferOnly() const { return !peeked_size; }
    bool currentlyReadFromOwnMemory() const { return working_buffer.begin() != sub_buf->buffer().begin(); }

The transition happens in peekNext(), which is called when the sub-buffer runs dry while a checkpoint is active. It copies the unread tail of the sub-buffer into own memory:

cpp
// src/IO/PeekableReadBuffer.cpp:50
    /// May throw an exception
    resizeOwnMemoryIfNecessary(bytes_to_copy);

    if (useSubbufferOnly())
    {
        sub_buf->position() = copy_from;
    }

    char * memory_data = getMemoryData();

    /// Save unread data from sub-buffer to own memory
    memcpy(memory_data + peeked_size, sub_buf->position(), bytes_to_copy);

and then the rollback is trivial:

cpp
// src/IO/PeekableReadBuffer.cpp:97
void PeekableReadBuffer::rollbackToCheckpoint(bool drop)
{
    checkStateCorrect();

    chassert(checkpoint);

    /// Reset canceled flag since the purpose of rollback is to retry reading from the beginning.
    /// This is important for schema detection where we try multiple formats and expect some to fail.
    canceled = false;

    if (recursive_checkpoints_offsets.empty())
    {
        if (checkpointInOwnMemory() == currentlyReadFromOwnMemory())
        {
            /// Both checkpoint and position are in the same buffer.
            pos = *checkpoint;
        }
        else
        {
            /// Checkpoint is in own memory and position is not.
            chassert(checkpointInOwnMemory());

            char * memory_data = getMemoryData();
            /// Switch to reading from own memory.
            BufferBase::set(memory_data, peeked_size, *checkpoint - memory_data);
        }
    }

Two cases: if the checkpoint and the cursor live in the same buffer, roll the cursor back; if the checkpoint lives in own memory but the cursor has moved on into the sub-buffer, switch the working buffer back to own memory and set the cursor to the checkpoint’s offset.

The canceled = false line is a lovely detail: schema detection tries several formats against the same buffer and expects some of them to fail. A failed parse cancels the buffer; a rollback un-cancels it, because “we are going to try again from here” is exactly what a rollback means.

Recursive checkpoints

Nested checkpoints are supported by storing the offset from the first checkpoint rather than a second pointer. That keeps the invariant simple: one pointer into one of two buffers, plus a stack of offsets.

cpp
// src/IO/PeekableReadBuffer.h:131
class PeekableReadBufferCheckpoint : boost::noncopyable
{
    PeekableReadBuffer & buf;
    bool auto_rollback;
public:
    explicit PeekableReadBufferCheckpoint(PeekableReadBuffer & buf_, bool auto_rollback_ = false)
                : buf(buf_), auto_rollback(auto_rollback_) { buf.setCheckpoint(); }
    ~PeekableReadBufferCheckpoint()
    {
        if (!buf.checkpoint)
            return;
        if (auto_rollback)
            buf.rollbackToCheckpoint();
        buf.dropCheckpoint();
    }

This RAII wrapper is the ergonomic front end: create it, do speculative reading, and either let it roll back automatically or commit by letting it destruct. The CSV reader uses it exactly like this, to test whether a multi-character custom delimiter really follows:

cpp
// src/IO/ReadHelpers.cpp:1111
                {
                    PeekableReadBufferCheckpoint checkpoint{*peekable_buf, true};
                    if (checkString(custom_delimiter, *peekable_buf))
                        return;
                }

The stack scratch buffer

One more piece of pragmatism: the peek buffer does not allocate on construction. It has a small stack buffer and only falls back to the heap if the checkpoint spans more than that:

cpp
// src/IO/PeekableReadBuffer.h:120
    /// To prevent expensive and in some cases unnecessary memory allocations on PeekableReadBuffer
    /// creation (for example if PeekableReadBuffer is often created or if we need to remember small amount of
    /// data after checkpoint), at the beginning we will use small amount of memory on stack and allocate
    /// larger buffer only if reserved memory is not enough.
    char stack_memory[PADDING_FOR_SIMD]; // NOLINT(cppcoreguidelines-pro-type-member-init,hicpp-member-init) - scratch buffer, written before read
    bool use_stack_memory = true;

PADDING_FOR_SIMD bytes — 64 — on the stack, and the promotion to heap memory happens lazily:

cpp
// src/IO/PeekableReadBuffer.cpp:258
    if (use_stack_memory)
    {
        /// If stack memory is still enough, do nothing.
        if (sizeof(stack_memory) >= new_size)
            return;

        /// Stack memory is not enough, allocate larger buffer.
        use_stack_memory = false;
        memory.resize(std::max(static_cast<size_t>(DBMS_DEFAULT_BUFFER_SIZE), new_size));
        memcpy(memory.data(), stack_memory, sizeof(stack_memory));

Most checkpoints in practice are short — you peek at a few bytes to see whether a quote follows — so the common case is zero allocations.

Trade-offs

What this design buys:

One virtual call per 1 MiB, not per byte. The nextImpl() boundary is the only dynamic dispatch in the whole stack.
A scanner can look at the bytes. buffer().begin() and buffer().end() are public, so a parser can memchr for a delimiter without going through the buffer API at all.
Failure is contained. A write that fails rewinds the cursor and marks the buffer canceled, so a partially written file is detectable.
Cheap speculation. Checkpoint/rollback turns “read ahead and decide” from a scary problem into a pointer assignment.

What it costs:

eof() mutates. The API looks like a predicate and is not. Code that calls eof() in a loop for its side effect is fine; code that assumes eof() is idempotent is not.
The invariant is manual. pos is a raw char * into a raw allocation. There is no bounds checking; chassert catches violations in debug builds only. Every derived class that implements nextImpl() must get working_buffer right.
No pushback. ReadBuffer cannot unread. Every parser that needs to look ahead must be wrapped in a PeekableReadBuffer, which is a second layer of state (two buffers, a checkpoint, a stack of recursive offsets) for what looks like a trivial feature.
Two copies for peeked data. When a checkpoint is active and the sub-buffer is exhausted, the remaining bytes are copied into own memory. For a large speculative read, that is a real memcpy of the whole region.

The design is not exotic; it is just very carefully pinned down. ClickHouse’s I/O layer is a good example of an API that is small enough to hold in your head and tight enough that the compiler can inline the hot parts — the trick is that it exposes the raw bytes instead of hiding them.