Opensource internals

ClickHouse Batches Distributed Inserts into Files on Disk

16 min read Jay Hubbert

A Distributed table INSERT is turned into Native-format .bin files on the local disk, queued by a background thread, and shipped to shards in batches that survive crashes and retries.

TL;DR: A Distributed table does not send rows over the network inline. It writes them to the local disk as Native-format .bin files, and a background thread ships those files to the shards. The filesystem is the queue: it survives a crash, it is what the “batch” is made of, and re-sending a batch after a failure is bit-for-bit identical, which is what makes ReplicatedMergeTree deduplication work.

Every Distributed table has the same shape: rows arrive at one node, and the node must put each row on the shard that owns it. The naive implementation — open a connection per shard, stream rows, done — has two problems that only show up under load.

The first is latency. An INSERT of a thousand rows to a five-shard cluster is five thousand network round trips if you are not careful, and the user is waiting for all of them.

The second is durability, and it is the one that shapes the whole design. If the initiator crashes halfway through an insert, what happens to the rows it had already handed to the network stack? Either they are lost, or they are somewhere and you do not know where. And if the initiator retries the insert, the rows that did make it are now duplicated. Retrying an insert is not an unusual thing to do — it happens when a client times out, when a coordinator restarts, when a user presses the button twice.

So the insert must be made durable before it is sent, and it must be idempotent when resent. The trick is that ClickHouse does not treat “the insert” as an in-memory buffer at all. It treats it as a file.

text 
   INSERT INTO dist
         |
         v
   +-------------------+   split by sharding key   +---------------+
   |  DistributedSink  | ------------------------> |  shard 0      |
   +-------------------+                          +---------------+
         |                                          +---------------+
         |  write ONE .bin file per shard            |  shard 1      |
         v                                          +---------------+
   disk: .../distributed/<dir>/<addr>/<n>.bin
         |
         v
   +-----------------------------+   pop files, group by header,
   | DistributedAsyncInsert      |   write current_batch.txt,
   | DirectoryQueue (background) |   send, delete on success
   +-----------------------------+

The file on disk is the pending insert. If the server dies, the file survives and the next start picks it up. If the send fails halfway, the file is still there and is sent again — the same bytes, so the receiving ReplicatedMergeTree recognizes it as a duplicate and drops it.

This article follows the whole path: how the sink writes the file, what is in it, how the background thread batches files, how a failed batch is split and retried, and where the data ends up when it is hopeless.

The file is the unit of work

DistributedSink has two modes, and the default is the file-based one: distributed_foreground_insert is false, so the rows go to the disk and the INSERT returns as soon as the file is written. Setting it to 1 switches to the inline mode, where the rows are pushed to the shards right away, in the calling thread, with one connection per shard and a RemoteInserter per connection.

The decision is made once, when the sink is created:

cpp 
// src/Storages/StorageDistributed.cpp:1288
    bool insert_sync = settings[Setting::distributed_foreground_insert] || settings[Setting::insert_shard_id] || relative_data_path.empty();

The two extra conditions are worth noting: an insert aimed at a specific shard (insert_shard_id) is always synchronous, and so is an insert into a table with no local data path — there is nowhere to put a file.

The file format is written by DistributedSink::writeToShard. The interesting part is not the data, it is the header — everything the sending side needs to reconstruct the insert on the other end, stored inside the file so that the file is self-contained and can be shipped by a thread that knows nothing about the query that produced it:

cpp 
// src/Storages/Distributed/DistributedSink.cpp:1031
            writeVarUInt(DBMS_DISTRIBUTED_SIGNATURE_HEADER, out);
            writeStringBinary(header, out);
            writePODBinary(CityHash_v1_0_2::CityHash128(header.data(), header.size()), out);

The DBMS_DISTRIBUTED_SIGNATURE_HEADER magic plus a CityHash-128 of the header lets the reader detect a truncated or corrupt file before it starts pushing rows into a remote table — which matters, because a partially-applied insert is much worse than a rejected one.

The header itself carries the query, the settings, the client info, the row and byte counts, and the shard number:

cpp 
// src/Storages/Distributed/DistributedSink.cpp:996
            writeVarUInt(block.rows(), header_buf);
            writeVarUInt(block.bytes(), header_buf);
            writeStringBinary(block.cloneEmpty().dumpStructure(), header_buf); /// obsolete
            /// Write block header separately in the batch header.
            /// It is required for checking does conversion is required or not.
            {
                NativeWriter header_stream{header_buf, DBMS_TCP_PROTOCOL_VERSION, std::make_shared<const Block>(block.cloneEmpty())};
                header_stream.write(block.cloneEmpty());
            }

            writeVarUInt(shard_info.shard_num, header_buf);
            writeStringBinary(storage.cluster_name, header_buf);
            writeStringBinary(storage.getStorageID().getFullNameNotQuoted(), header_buf);
            writeStringBinary(storage.getRemoteDatabaseName() + "." + storage.getRemoteTableName(), header_buf);

Note the /// obsolete marker on dumpStructure(): the file format has been extended over the years and the old field is still written so that old servers can still read new files. The block header is written separately, in Native format, precisely so that the sender can check “does the remote table have the same columns as me?” before sending any data — if not, it falls back to a per-block conversion path.

Forward compatibility by trailing fields

A file written by a new server may be read by an old server (during a rolling upgrade) or by a new server (after a downgrade). The format handles this by appending fields at the end of the header and never reordering:

cpp 
// src/Storages/Distributed/DistributedSink.cpp:1025
            /// Add new fields here, for example:
            /// writeVarUInt(my_new_data, header_buf);
            /// And note that it is safe, because we have checksum and size for header.

The ClientInfo written into the header deliberately omits client_agent and the other new fields, which are appended separately as trailing fields:

cpp 
// src/Storages/Distributed/DistributedSink.cpp:1011
            /// Trailing field: the detected AI coding agent of the initiating client.
            /// Kept here (rather than inside the embedded `ClientInfo` above) so older binaries can
            /// safely ignore it when draining queue files written by a newer binary.
            writeStringBinary(context->getClientInfo().client_agent, header_buf);

The reason is that ClientInfo::write has its own on-the-wire layout, and older readers parse it strictly. Adding a field inside ClientInfo would break them; adding it after the ClientInfo blob does not, because an old reader stops reading where the old structure ended.

The settings stored in the header are also filtered. The initiator-only settings (query shaping, result serialization, database) are stripped, because the replayed INSERT runs on a connection to the remote shard and must resolve the remote table against that server’s default database:

cpp 
// src/Storages/Distributed/DistributedSink.cpp:976
                Settings insert_settings = context->getSettingsCopy();
                ClusterProxy::stripInitiatorOnlySettings(insert_settings);
                insert_settings.write(header_buf);

A shard may have several replicas, and each replica needs its own copy of the data. Writing the payload once per replica would multiply the disk I/O by the replica count.

ClickHouse writes it once into a tmp directory and then hardlinks it into each replica’s directory:

cpp 
// src/Storages/Distributed/DistributedSink.cpp:916
    /// tmp directory is used to ensure atomicity of transactions
    /// and keep directory queue thread out from reading incomplete data
    std::string first_file_tmp_path;

The tmp subdirectory has a second job: the queue thread scans the parent directory for *.bin files, so a half-written file must never appear there. The sequence is: write to tmp/<n>.bin, fsync it, then hardlink it into the replica’s directory as <n>.bin, then hand the path to the queue.

cpp 
// src/Storages/Distributed/DistributedSink.cpp:1049
            createHardLink(first_file_tmp_path, bin_file);
            auto dir_sync_guard = make_directory_sync_guard(*it);
        }
        directory_queue->addFileAndSchedule(bin_file, file_size, sleep_ms);

and for the remaining replicas:

cpp 
// src/Storages/Distributed/DistributedSink.cpp:1065
            createHardLink(first_file_tmp_path, bin_file);
            auto dir_sync_guard = make_directory_sync_guard(*it);
        }
        directory_queue->addFileAndSchedule(bin_file, file_size, sleep_ms);

The hardlink is what makes the file name unique across replicas: the same file name <n>.bin (from storage.file_names_increment) is used for every replica of a shard, and only the containing directory differs. The queue for each directory is then independent — one replica’s queue being stuck does not block the others.

The temporary file is removed at the very end, after all hardlinks are made:

cpp 
// src/Storages/Distributed/DistributedSink.cpp:1071
    /// remove the temporary file, enabling the OS to reclaim inode after all threads
    /// have removed their corresponding files
    fs::remove(first_file_tmp_path);

The inode survives until the last hardlink goes away, which is exactly the semantics you want: the data is not deleted out from under a queue thread that is still reading it.

The queue

Each (disk, replica directory) pair gets its own DistributedAsyncInsertDirectoryQueue. The queue holds an in-memory list of pending file paths and a background task that drains it:

cpp 
// src/Storages/Distributed/DistributedAsyncInsertDirectoryQueue.h:154
    ConcurrentBoundedQueue<std::string> pending_files;

The in-memory list is a cache, not the source of truth. On startup the queue scans the directory and rebuilds it:

cpp 
// src/Storages/Distributed/DistributedAsyncInsertDirectoryQueue.cpp:362
        for (fs::directory_iterator it{path}; it != end; ++it)
        {
            const auto & file_path = it->path();
            const auto & base_name = file_path.stem().string();
            if (!it->is_directory() && startsWith(fs::path(file_path).extension(), ".bin") && parse<UInt64>(base_name))
            {
                if (!pending_files.push(fs::absolute(file_path).string()))
                    throw Exception(ErrorCodes::LOGICAL_ERROR, "Cannot schedule a file '{}'", file_path.string());
                bytes_count += fs::file_size(file_path);
            }

The file names are <counter>.bin, so parse<UInt64>(base_name) is a cheap validity test: anything that is not a number is not a queue file and is logged as unexpected.

Batching: group files that can be sent together

With distributed_background_insert_batch = 1, the queue does not send one file at a time. It groups consecutive files that have the same header — same query, same settings, same client info kind, same column structure — and sends them in a single INSERT to the shard.

The grouping key is a small struct with an operator== and a hash:

cpp 
// src/Storages/Distributed/DistributedAsyncInsertDirectoryQueue.cpp:526
    bool operator==(const BatchHeader & other) const
    {
        return std::tie(settings, query, client_info.query_kind, client_info.client_agent, client_info.is_internal) ==
               std::tie(other.settings, other.query, other.client_info.query_kind, other.client_info.client_agent, other.client_info.is_internal) &&
               blocksHaveEqualStructure(header, other.header);
    }

Note that settings is compared by value: Settings has an operator== that compares the actual values, so two files with the same settings text but different max_threads land in different batches. That is the correct behaviour — they cannot be merged into one remote INSERT without changing what the remote server does.

A batch is sent when it is big enough:

cpp 
// src/Storages/Distributed/DistributedAsyncInsertBatch.cpp:78
bool DistributedAsyncInsertBatch::isEnoughSize() const
{
    return (!parent.min_batched_block_size_rows && !parent.min_batched_block_size_bytes)
        || (parent.min_batched_block_size_rows && total_rows >= parent.min_batched_block_size_rows)
        || (parent.min_batched_block_size_bytes && total_bytes >= parent.min_batched_block_size_bytes);
}

The thresholds are min_insert_block_size_rows and min_insert_block_size_bytes — the insert block size settings, not a separate “batch size” setting. With both at zero (the defaults), any non-empty batch is sent immediately, so the batching is really about coalescing whatever is already pending rather than about waiting for a quota.

current_batch.txt: the pending-files trick

Here is the trick that makes crash recovery and exactly-once semantics work.

Before sending a batch, the queue writes the list of files in that batch to a sidecar file called current_batch.txt. The comment says why:

cpp 
// src/Storages/Distributed/DistributedAsyncInsertBatch.cpp:96
        /// For deduplication in Replicated tables to work, in case of error
        /// we must try to re-send exactly the same batches.
        /// So we save contents of the current batch into the current_batch_file_path file
        /// and truncate it afterwards if all went well.
        serialize();

The write is atomic — write to a temp file, then rename:

cpp 
// src/Storages/Distributed/DistributedAsyncInsertBatch.cpp:168
    /// Temporary file is required for atomicity.
    String tmp_file{parent.current_batch_file_path + ".tmp"};

and the file contains the indices of the files, not their paths, because the paths are reconstructed from the queue’s own directory:

cpp 
// src/Storages/Distributed/DistributedAsyncInsertBatch.cpp:176
        WriteBufferFromFile out{tmp_file, O_WRONLY | O_TRUNC | O_CREAT};
        for (const auto & file : files)
        {
            UInt64 file_index = parse<UInt64>(std::filesystem::path(file).stem());
            out << file_index << '\n';
        }

        out.finalize();
        if (fsync)
            out.sync();

On the next start, the queue checks whether current_batch.txt exists; if it does, the previous run died mid-batch:

cpp 
// src/Storages/Distributed/DistributedAsyncInsertDirectoryQueue.cpp:575
    /// Possibly, we failed to send a batch on the previous iteration. Try to send exactly the same batch.
    if (fs::exists(current_batch_file_path))
    {
        LOG_DEBUG(log, "Restoring the batch");

        DistributedAsyncInsertBatch batch(*this);
        if (batch.recoverBatch())
            batch.send(settings_changes, /*update_current_batch=*/ false);

recoverBatch() reconstructs the file list and, importantly, refuses to guess:

cpp 
// src/Storages/Distributed/DistributedAsyncInsertBatch.cpp:204
    /// Files are removed in order, so a missing prefix was already processed
    /// before an abnormal shutdown. Keep the surviving suffix in its persisted order.
    ///
    /// A quarantined twin in the broken directory does not prove that the files listed
    /// before it were sent: older servers skipped files that failed with a transient
    /// error and went on with the next one, so a batch written by such a server can
    /// hold an unsent file in front of a quarantined one. Never finalize a file that
    /// still exists here. Resending it may duplicate rows, deleting it loses them.
    auto first_existing_file = files.begin();
    while (first_existing_file != files.end() && !fs::exists(*first_existing_file))
    {
        LOG_WARNING(parent.log, "File {} does not exist, likely due abnormal shutdown", *first_existing_file);
        ++first_existing_file;
    }
    files.erase(files.begin(), first_existing_file);

This is a nice example of a design decision that is conservative on purpose: the code knows it cannot distinguish “sent” from “not sent” for a file that is missing, so it keeps the whole surviving suffix rather than deleting anything it is not sure about. Duplicates are recoverable (the receiving ReplicatedMergeTree deduplicates them); lost rows are not.

The rest of recoverBatch re-reads each file’s header to recompute the batch’s row and byte counts, and gives up entirely if any file is corrupt:

cpp 
// src/Storages/Distributed/DistributedAsyncInsertBatch.cpp:220
    /// A missing file inside the surviving suffix cannot be recovered safely.
    for (const auto & file : files)
    {
        if (!fs::exists(file))
        {
            LOG_WARNING(parent.log, "File {} does not exist, likely due abnormal shutdown", file);
            return false;
        }

Splitting a batch that fails

A batch is sent as one remote INSERT. If that INSERT fails, the whole batch fails. But the failure may be caused by one of the files — for example a MATERIALIZED VIEW with a GROUP BY that exceeds max_bytes_to_transform because of the accumulated rows in that one file.

Retrying the same batch would fail forever. Splitting it into individual files usually succeeds. That is exactly what distributed_background_insert_split_batch_on_failure does, and the list of “splittable” error codes is explicit:

cpp 
// src/Storages/Distributed/DistributedAsyncInsertBatch.cpp:57
static bool isSplittableErrorCode(int code, bool remote)
{
    return code == ErrorCodes::MEMORY_LIMIT_EXCEEDED
        /// FunctionRange::max_elements and similar
        || code == ErrorCodes::ARGUMENT_OUT_OF_BOUND
        || code == ErrorCodes::TOO_MANY_PARTS
        || code == ErrorCodes::TOO_MANY_BYTES
        || code == ErrorCodes::TOO_MANY_ROWS_OR_BYTES
        || code == ErrorCodes::TOO_MANY_PARTITIONS
        || code == ErrorCodes::DISTRIBUTED_TOO_MANY_PENDING_BYTES
        || code == ErrorCodes::DISTRIBUTED_BROKEN_BATCH_INFO
        || isDistributedSendBroken(code, remote)
    ;
}

The split is not a retry loop; it is a different send path. sendSeparateFiles opens a fresh connection per file, and quarantines the individual file that actually failed:

cpp 
// src/Storages/Distributed/DistributedAsyncInsertBatch.cpp:382
                parent.markAsBroken(file);
                ++broken_files;
                ++processed_files;

Note the asymmetry: the file that failed is quarantined, not retried forever. That is the escape hatch, and it is the subject of the next section.

The broken directory

Every queue has a sibling broken/ directory. A file that cannot be sent — a corrupt file, a file whose remote INSERT fails with an unrecoverable error — is renamed into it:

cpp 
// src/Storages/Distributed/DistributedAsyncInsertDirectoryQueue.cpp:697
void DistributedAsyncInsertDirectoryQueue::markAsBroken(const std::string & file_path)
{
    const String & broken_file_path = fs::path(broken_path) / fs::path(file_path).filename();

A rename is atomic and cheap, and it takes the file out of the queue’s view without deleting it. The operator can look at system.distribution_queue to see the counts:

cpp 
// src/Storages/Distributed/DistributedAsyncInsertDirectoryQueue.cpp:719
    fs::rename(file_path, broken_file_path);
    LOG_ERROR(log, "Renamed `{}` to `{}`", file_path, broken_file_path);

This is the difference between “the data is lost” and “the data is parked where a human can find it”. A distributed insert that hits a broken shard ends up in broken/, visible in system.distribution_queue under broken_data_files / broken_data_compressed_bytes, and the server keeps running.

Backoff

A queue that cannot reach its shard must not hammer it. The sleep time grows exponentially with the error count:

cpp 
// src/Storages/Distributed/DistributedAsyncInsertDirectoryQueue.cpp:200
std::chrono::milliseconds DistributedAsyncInsertDirectoryQueue::calculateSleepTime(
    std::chrono::milliseconds default_sleep_time,
    std::chrono::milliseconds max_sleep_time,
    size_t error_count)
{
    if (default_sleep_time.count() <= 0)
        return std::min(default_sleep_time, max_sleep_time);

    const UInt64 q = doubleToUInt64(std::exp2(error_count));
    /// Clamp before multiplying: default_sleep_time * q can overflow for large error_count.
    if (q > static_cast<UInt64>(max_sleep_time.count() / default_sleep_time.count()))
        return max_sleep_time;

    return std::min(std::chrono::milliseconds(default_sleep_time.count() * static_cast<Int64>(q)), max_sleep_time);
}

2^error_count times the base interval, clamped to max_sleep_time. The base is distributed_background_insert_sleep_time_ms (100 ms by default) and the cap is distributed_background_insert_max_sleep_time_ms (30 s).

The error count decays, so a transient outage does not leave the queue sleeping for 30 seconds forever:

cpp 
// src/Storages/Distributed/DistributedAsyncInsertDirectoryQueue.cpp:265
    /// Decay error_count on a non-failing, unblocked run so the backoff relaxes after recovery.
    /// Skip on error (keep growing during an outage) and while sends are stopped (preserve backoff).
    if (!had_error && !monitor_blocker.isCancelled())
    {
        const auto now = std::chrono::system_clock::now();
        std::lock_guard status_lock(status_mutex);
        if (now - last_decrease_time > decrease_error_count_period)
        {
            status.error_count /= 2;
            last_decrease_time = now;
        }
    }

Halving every five minutes rather than resetting to zero means a queue that fails once an hour still keeps a little hysteresis, but a queue that has been healthy for a while converges back to the base interval.

The whole picture

text 
INSERT INTO dist (a, b) VALUES ...
        |
        |  DistributedSink::consume -> writeAsync -> writeAsyncImpl
        |  (split by sharding key, one block per shard)
        v
  writeToShard(shard):
        |
        |  write <tmp>/<n>.bin     (header + Native block, compressed)
        |  fsync (optional)
        |  createHardLink -> <replica_dir>/<n>.bin
        |  queue->addFileAndSchedule(<replica_dir>/<n>.bin)
        v
  DirectoryQueue::pending_files  <---- also rebuilt from disk at startup
        |
        |  background task, sleep = 100ms * 2^errors (max 30s)
        v
  processFilesWithBatching:
        |  group by (query, settings, client_info, structure)
        |  if big enough: write current_batch.txt, send
        v
  sendBatch:
        |  one connection, one RemoteInserter, N files streamed
        |  on failure -> isSplittableErrorCode? -> sendSeparateFiles
        v
  success: markAsSend (unlink)        failure: markAsBroken (rename to broken/)

Using it

sql 
CREATE TABLE events_distributed
(
    `user_id` UInt64,
    `event_time` DateTime,
    `payload`  String
)
ENGINE = Distributed('my_cluster', 'db', 'events_local', cityHash64(user_id));

-- The background mode: the INSERT returns as soon as the file is on disk.
INSERT INTO events_distributed (user_id, event_time, payload)
VALUES (42, now(), 'hello');

-- What is pending, and what is hopeless?
SELECT database, table, data_path, data_files, data_compressed_bytes,
       broken_data_files, broken_data_compressed_bytes, error_count
FROM system.distribution_queue
WHERE is_blocked = 0;

The settings that matter:

sql 
-- Batch the files into one remote INSERT, and split the batch if it fails.
SET distributed_background_insert_batch = 1;
SET distributed_background_insert_split_batch_on_failure = 1;

-- The default: the INSERT returns once the file is on disk.
SET distributed_foreground_insert = 0;

-- The synchronous mode: the INSERT returns only after every shard has the data.
SET distributed_foreground_insert = 1;

distributed_foreground_insert = 0 is the switch that turns on the whole file-based path, and it is the default. With it on, DistributedSink::writeAsync writes the file and returns; with it off, the rows are pushed inline to the remote shards instead.

Trade-offs

What the file-based design buys you:

  • Durability. The pending insert is on disk before the client is told anything. A crash of the initiator loses nothing.
  • Idempotence. The bytes are identical on every retry, so the receiving ReplicatedMergeTree can recognize the duplicate and drop it. This is the whole reason current_batch.txt records the exact file list.
  • Backpressure for free. The queue is the buffer; there is no separate “queue full” error path. Slow shards just accumulate files.
  • A uniform failure taxonomy. A file either goes or it is quarantined in broken/. There is no third state.

What it costs:

  • Disk I/O and space. Every row is written locally and read back. For a high-throughput insert this is the dominant cost, and it is why distributed_foreground_insert defaults to on.
  • Extra latency. The data is written, then read, then sent. The queue sleeps up to distributed_background_insert_sleep_time_ms between rounds, so the tail latency of an insert is at least one sleep interval plus one round trip.
  • A file per (shard, replica) pair. A five-shard, three-replica cluster writes fifteen copies of every block, though hardlinks make the extra copies free in terms of bytes — only inodes.
  • The broken/ directory is a data-loss-shaped hole. Rows that end up there are not in the table, and nothing will retry them. Monitoring system.distribution_queue is not optional in production.

The part people get wrong: the “exactly once” guarantee is not the distributed table’s, it is the receiving table’s. The Distributed engine guarantees at-least-once delivery of the same bytes. It is ReplicatedMergeTree’s content-hash deduplication that turns that into exactly-once. If the remote table is a plain MergeTree, a retried batch is a duplicated batch.

The part people get wrong, part two: the batch grouping key includes Settings by value. If your inserts differ in any setting — even max_threads — they will not be coalesced, and you will get one remote INSERT per file, which is exactly the latency profile you were trying to avoid.

ClickhouseDistributedInternalsReliabilityPerformance