How ClickHouse Batches Async Inserts in a Per-Key Queue

TL;DR: Every async insert is hashed into a UInt128 key that covers the query AST, the user’s identity, the format and every non-default setting. Inserts with the same key land in the same buffer, one buffer per key per shard. The buffer is flushed when it is big enough, holds enough queries, or its (adaptive) deadline expires — and the flush is done by a background thread while the client blocks on a std::future.

Writing to a MergeTree table is expensive per statement: a part has to be created, its metadata written, its checksums computed, and (for a replicated table) a ZooKeeper transaction has to allocate a block number. A client that sends 10 000 single-row INSERT statements pays that cost 10 000 times.

The obvious fix — “batch client-side, send one big INSERT” — puts the responsibility on the client, which every client gets wrong at least once, and which the INSERT ... SELECT path cannot use at all.

ClickHouse’s answer is to move the batching into the server and make it transparent: the client still sends one statement per batch of rows, the server buffers it, and the server decides when to turn N statements into one part.

The problem is that “one batch” is not a well-defined thing

Before you can buffer inserts, you have to decide which inserts may share a buffer. Two inserts into the same table with the same format obviously can. But what about two inserts into the same table with different SETTINGS? One with insert_deduplication_token = 'a', another with insert_deduplication_token = 'b'? Or one from alice and one from bob?

The answer has to be: nothing but inserts that are semantically indistinguishable may share a buffer. And that set is large — it is a function of the query text, the user, the roles, the credentials, the settings, the format, and even the physical representation of the data.

The second problem is that a buffer is not just a std::vector<Block>. It has to:

know when to flush,
be visible to a background thread,
carry a deadline that a background thread can watch,
and let the client learn, later, whether its rows actually landed — and how many.

The key: a hash of everything that matters

AsynchronousInsertQueue keys each buffer by an InsertQuery object, and that object computes a UInt128 hash of everything that must not be shared:

cpp
// src/Interpreters/AsynchronousInsertQueue.cpp:162
    SipHash siphash;

    siphash.update(data_kind);
    query->updateTreeHash(siphash, /*ignore_aliases=*/ true);

    if (user_id)
    {
        siphash.update(*user_id);
        for (const auto & current_role : current_roles)
        {
            siphash.update(current_role);
        }
    }

The query is hashed structurally, through the AST (updateTreeHash), not textually. Two INSERT INTO t VALUES (1) statements with different whitespace produce the same key; two inserts with different column lists do not.

Then the credential limit and the credential expiry are folded in, because a deferred flush must not run with more rights than the session had, and must not run at all if the credential expired while the data sat in the buffer:

cpp
// src/Interpreters/AsynchronousInsertQueue.cpp:176
    /// Fold the credential grant limit into the key so inserts made under different limits are never
    /// coalesced into one flush (the queue is keyed by `hash` alone). A null limit (the common,
    /// unrestricted case) contributes nothing, so it does not change the hash of existing keys.
    /// A non-null limit is hashed by its textual form even when it is semantically empty (a deny-all
    /// token), so it never collides with the null case; the leading length prefix keeps it unambiguous.
    if (authentication_grants)
    {
        /// Use the precise serialization (never `toString`): under `enable_read_write_grants = 0` the
        /// backward-compatibility widening collapses distinct source limits such as `READ ON FILE` and
        /// `WRITE ON FILE` into one, which would let a read-only token piggyback into a write token's bucket.
        const auto grants_str = authentication_grants->toStringPrecise();
        siphash.update(grants_str.size());
        siphash.update(grants_str);
    }

and the identity strings are length-prefixed, because a SipHash::update(String) streams raw bytes and the queue is keyed by the hash alone:

cpp
// src/Interpreters/AsynchronousInsertQueue.cpp:198
    /// Length-prefix each field: update(String) streams only bytes and the queue is keyed
    /// by hash alone, so otherwise "a"/"a"/"aaa" and "aa"/"aa"/"a" would collide.
    for (const String & identity_field : {current_user, initial_user, authenticated_user, quota_key})
    {
        siphash.update(identity_field.size());
        siphash.update(identity_field);
    }

Finally, the settings. Every setting that differs from the default makes a different buffer, because a setting is semantics — max_insert_block_size changes the block shape, async_insert_deduplicate changes whether deduplication happens at all:

cpp
// src/Interpreters/AsynchronousInsertQueue.cpp:206
    setting_changes = settings->changes();
    for (auto it = setting_changes.begin(); it != setting_changes.end();)
    {
        if (settings_to_skip.contains(it->name))
        {
            it = setting_changes.erase(it);
        }
        else
        {
            siphash.update(it->name);
            applyVisitor(FieldVisitorHash(siphash), it->value);
            ++it;
        }
    }

    hash = siphash.get128();

Two settings are explicitly skipped, and the comment says why:

cpp
// src/Interpreters/AsynchronousInsertQueue.cpp:127
static const NameSet settings_to_skip
{
    /// We don't consider this setting because it is only for deduplication,
    /// which means we can put two inserts with different tokens in the same block safely.
    "insert_deduplication_token",
    "log_comment",
};

insert_deduplication_token is the interesting one. It looks like semantics — it names a deduplication token — but the token travels with the data, not with the buffer, so two inserts with different tokens can safely share a buffer. The token is carried per entry and re-attached at flush time (see below).

The queue: a multimap of deadlines plus a hash index

The data structure is a two-part affair. The primary container is ordered by time, because the background worker needs to find the earliest deadline:

cpp
// src/Interpreters/AsynchronousInsertQueue.h:270
    struct Container
    {
        InsertQuery key;
        InsertDataPtr data;
    };

    /// Ordered container
    /// Key is a timestamp of the first insert into batch.
    /// Used to detect for how long the batch is active, so we can dump it by timer.
    /// Must be a multimap: two queries with different keys may theoretically compute the same deadline.
    using Queue = std::multimap<std::chrono::steady_clock::time_point, Container>;
    using QueueIterator = Queue::iterator;
    using QueueIteratorByKey = std::unordered_map<UInt128, QueueIterator>;

The comment about multimap is a real subtlety: two different keys can legitimately compute the same deadline (they both hashed to different buckets but the adaptive timeout converged them), and a std::map would silently drop one of the buffers.

The secondary container is a UInt128 → iterator index, because the producer needs to find an existing buffer for its key in O(1) and append to it. The queue is also sharded, so that two producers never contend on the same mutex:

cpp
// src/Interpreters/AsynchronousInsertQueue.h:286
    struct QueueShard
    {
        mutable std::mutex mutex;
        mutable std::condition_variable are_tasks_available;
        /// Counts batches removed by producers or the deadline worker, including those
        /// still waiting for pool admission. Released when the batch is destroyed.
        std::atomic<size_t> in_flight_flushes{0};

        Queue queue TSA_GUARDED_BY(mutex);
        QueueIteratorByKey iterators TSA_GUARDED_BY(mutex);

        OptionalTimePoint last_insert_time TSA_GUARDED_BY(mutex);
        std::chrono::milliseconds busy_timeout_ms TSA_GUARDED_BY(mutex) {};
    };

The shard is picked by the key hash, so the same key always goes to the same shard and two keys with the same hash share a shard:

cpp
// src/Interpreters/AsynchronousInsertQueue.cpp:736
    size_t shard_num = static_cast<size_t>(key.hash % pool_size);
    auto & shard = queue_shards[shard_num];
    const auto flush_time_points = flush_time_history_per_queue_shard[shard_num].getRecentTimePoints();
    {
        std::lock_guard lock(shard.mutex);

and this is where the buffer is found or created:

cpp
// src/Interpreters/AsynchronousInsertQueue.cpp:760
        if (inserted)
        {
            try
            {
                it->second = shard.queue.emplace(now + timeout_ms, Container{key, std::make_unique<InsertData>(timeout_ms)});
            }
            catch (...)
            {
                shard.iterators.erase(it);
                throw;
            }
        }

it->second is the deadline of the buffer. Note that the deadline is computed at creation time and stored in the multimap’s key, so the deadline worker does not have to look at the buffer to know whether it is expired — it just looks at queue.begin().

The buffer: a list of entries, each with a promise

The buffer is a list of entries. Each entry is one client’s insert: the raw bytes (or a pre-parsed block), the query id, the dedup token, the format, and a std::promise that the flush will fulfil:

cpp
// src/Interpreters/AsynchronousInsertQueue.h:191
    struct InsertData
    {
        struct Entry
        {
        public:
            DataChunk chunk;
            const String query_id;
            const String async_dedup_token;
            const String format;
            /// Keeps the queued bytes charged to the user that pushed them until the flush frees them.
            const std::unique_ptr<MemoryTracker> queued_data_tracker;
            const std::chrono::time_point<std::chrono::system_clock> create_time;
            NameToNameMap query_parameters;

            Entry(
                DataChunk && chunk_,
                String && query_id_,
                const String & async_dedup_token_,
                const String & format_,
                std::unique_ptr<MemoryTracker> queued_data_tracker_);

            void resetChunk();
            void finish(ResultProgress result = {});
            void finish(std::exception_ptr exception_);

            std::future<ResultProgress> getFuture() { return promise.get_future(); }
            bool isFinished() const { return finished; }

        private:
            std::promise<ResultProgress> promise;
            std::atomic_bool finished = false;
        };

The promise is the whole client-facing contract. The producer takes the future before releasing the shard lock, and returns it to the client:

cpp
// src/Interpreters/AsynchronousInsertQueue.cpp:803
        data->size_in_bytes += entry_data_size;
        progress_future = entry->getFuture();

The finished flag is an atomic_bool with an exchange, which makes finish() idempotent — the flush path and the shutdown path can both try to finish the same entry:

cpp
// src/Interpreters/AsynchronousInsertQueue.cpp:327
void AsynchronousInsertQueue::InsertData::Entry::finish(ResultProgress result)
{
    if (finished.exchange(true))
        return;

    resetChunk();
    promise.set_value(result);
}

There is a second, subtler reason for the atomic_bool: the buffer’s destructor also finishes the entries, so an entry that was already flushed must not have its promise set twice.

When to flush: three triggers

A buffer is flushed when any of three conditions holds.

Size. The buffer accumulated at least async_insert_max_data_size bytes (10 MiB by default):

cpp
// src/Core/Settings.cpp:8526
    DECLARE(UInt64, async_insert_max_data_size, 10485760, R"(
Maximum size in bytes of unparsed data collected per query before being inserted

Cloud default value: `104857600` (100 MiB).
)", 0, \
        {"24.2", 1000000, 10485760, "The previous value appeared to be too small."}) \

Query count. The buffer holds at least async_insert_max_query_number (450) entries — but only when deduplication is on, because otherwise merging the rows into one block would change the deduplication semantics:

cpp
// src/Core/Settings.cpp:8532
    DECLARE(UInt64, async_insert_max_query_number, 450, R"(
Maximum number of insert queries before being inserted.
Only takes effect if setting [`async_insert_deduplicate`](#async_insert_deduplicate) is 1.
)", 0) \

Deadline. The buffer’s deadline has passed.

The three are evaluated together, and if any fires, the buffer is removed from the queue and handed to the thread pool:

cpp
// src/Interpreters/AsynchronousInsertQueue.cpp:811
        bool has_enough_bytes = data->size_in_bytes >= (*key.settings)[Setting::async_insert_max_data_size];
        bool has_enough_queries
            = data->entries.size() >= (*key.settings)[Setting::async_insert_max_query_number] && isDeduplicationEnabledForInsert(*key.settings);

cpp
// src/Interpreters/AsynchronousInsertQueue.cpp:828
        if (!flush_stopped && (has_enough_bytes || has_enough_queries || max_busy_timeout_exceeded()))
        {
            LOG_TRACE(log, "Scheduling async insert processing job because {}",
                      has_enough_bytes ? "enough bytes accumulated" :
                      has_enough_queries ? "enough queries accumulated" :
                      "maximum busy wait timeout exceeded");
            data->timeout_ms = Milliseconds::zero();
            data->trackFlush(shard.in_flight_flushes);
            data_to_process = std::move(data);

            NOEXCEPT_SCOPE({
                shard.iterators.erase(it);
                shard.queue.erase(queue_it);
            });
        }

Note the data_to_process variable: the flush is scheduled after the shard lock is released, so a slow pool admission does not block other producers appending to other buffers in the same shard.

The adaptive deadline

The deadline is not a constant. It adapts to the insert rate, per shard:

cpp
// src/Core/Settings.cpp:8540
    DECLARE(Bool, async_insert_use_adaptive_busy_timeout, true, R"(
If it is set to true, use adaptive busy timeout for asynchronous inserts
)", 0, \
        {"24.2", false, true, "Use adaptive asynchronous insert timeout"}) \
    DECLARE(Milliseconds, async_insert_busy_timeout_min_ms, 50, R"(
If auto-adjusting is enabled through async_insert_use_adaptive_busy_timeout, minimum time to wait before dumping collected data per query since the first data appeared. It also serves as the initial value for the adaptive algorithm
)", 0, \
        {"24.2", 50, 50, "The minimum value of the asynchronous insert timeout in milliseconds; it also serves as the initial value, which may be increased later by the adaptive algorithm"}) \
    DECLARE_WITH_ALIAS(Milliseconds, async_insert_busy_timeout_max_ms, 200, R"(
Maximum time to wait before dumping collected data per query since the first data appeared.

Cloud default value: `1000` (1s).
)", 0, async_insert_busy_timeout_ms, \
        {"24.2", 200, 200, "The minimum value of the asynchronous insert timeout in milliseconds; async_insert_busy_timeout_ms is aliased to async_insert_busy_timeout_max_ms"}) \

The default window is therefore 50 ms to 200 ms, with a 20 % growth and 20 % decay per step. The rule is simple: frequent inserts grow the window, idle inserts shrink it.

cpp
// src/Interpreters/AsynchronousInsertQueue.cpp:896
    const auto decreased_timeout_ms = std::min(
        std::chrono::duration_cast<Milliseconds>(shard.busy_timeout_ms / (1.0 + decrease_rate)), shard.busy_timeout_ms - Milliseconds(1));

    /// Increase the timeout for frequent inserts.
    if (last_insert_time + min_ms > now)
    {
        auto timeout_ms = std::max(
            std::chrono::duration_cast<Milliseconds>(shard.busy_timeout_ms * (1.0 + increase_rate)),
            shard.busy_timeout_ms + Milliseconds(1));

        return normalize(timeout_ms);
    }

The “frequent” test is last_insert_time + min_ms > now: an insert arrived less than min_ms (50 ms) after the previous one, so the producer is hot and the buffer is worth keeping open longer. The decay test additionally requires that the last two flushes were also far apart, so the timeout only shrinks when the shard is genuinely idle:

cpp
// src/Interpreters/AsynchronousInsertQueue.cpp:908
    /// Decrease the timeout if inserts are not frequent,
    /// that is, if the time since the last insert and the difference between the last two queue flushes were both
    /// long enough (exceeding the adjusted timeout).
    /// This ensures the timeout value converges to the minimum over time for non-frequent inserts.
    if (last_insert_time + decreased_timeout_ms < now && t1 + decreased_timeout_ms < t2)
        return normalize(decreased_timeout_ms);

Note that the timeout is per shard, not per key. That is a deliberate simplification: with 16 shards and hundreds of keys, a per-key timeout would mean hundreds of independent timers, and the “thundering herd” problem would be solved by the shard’s own mutex anyway. The per-shard value is what all the keys in that shard share.

The deadline worker

One thread per shard watches the deadlines:

cpp
// src/Interpreters/AsynchronousInsertQueue.cpp:1111
void AsynchronousInsertQueue::processBatchDeadlines(size_t shard_num) TSA_NO_THREAD_SAFETY_ANALYSIS
{
    auto & shard = queue_shards[shard_num];

    while (!shutdown)
    {
        std::vector<Container> entries_to_flush;
        {
            std::unique_lock lock(shard.mutex);

            const auto rel_time
                = std::min(shard.busy_timeout_ms, saturatedMilliseconds(getContext()->getSettingsRef()[Setting::async_insert_poll_timeout_ms].totalMilliseconds()));
            shard.are_tasks_available.wait_for(
                lock,
                rel_time,
                [&shard, this] TSA_NO_THREAD_SAFETY_ANALYSIS
                {
                    if (shutdown)
                        return true;

                    if (!shard.queue.empty() && shard.queue.begin()->first < std::chrono::steady_clock::now())
                        return true;

                    return false;
                });

The wait is a predicate wait, not a plain sleep: the worker wakes up early if the front of the queue has expired, and it is also woken explicitly by a producer (shard.are_tasks_available.notify_one()) when a new buffer is created. The poll interval is bounded by async_insert_poll_timeout_ms, which defaults to 10 ms:

cpp
// src/Core/Settings.cpp:8536
    DECLARE(Milliseconds, async_insert_poll_timeout_ms, 10, R"(
Timeout for polling data from asynchronous insert queue
)", 0, \
        {"24.2", 10, 10, "Timeout in milliseconds for polling data from asynchronous insert queue"}) \

so the worst case extra delay is the poll interval, not the window. The window is the deadline; the poll interval is how often the worker checks it. So a buffer with a 200 ms deadline is flushed between 200 ms and 210 ms after it was created.

Then the worker pops all expired buffers (the queue is ordered by deadline, so it is a prefix) and schedules them:

cpp
// src/Interpreters/AsynchronousInsertQueue.cpp:1145
            while (true)
            {
                if (shard.queue.empty() || shard.queue.begin()->first > now)
                    break;

                auto it = shard.queue.begin();

                NOEXCEPT_SCOPE({
                    /// The only exception that is possible here is MEMORY_LIMIT_EXCEEDED, by blocking them it is highly unlikely that we will fail here.
                    /// Besides it is not possible to write exception safe code here even with proper rollback, since the object can be moved out already (we can do copy, but this is more costly)
                    MemoryTrackerBlockerInThread lock_memory_tracker;

                    shard.iterators.erase(it->second.key.hash);

                    it->second.data->trackFlush(shard.in_flight_flushes);
                    entries_to_flush.emplace_back(std::move(it->second));

                    shard.queue.erase(it);
                });
            }
        }

        for (auto & entry : entries_to_flush)
            scheduleDataProcessingJob(entry.key, std::move(entry.data), getContext(), shard_num);

The flush: a whole new query

The flush is not a “write the buffer to the table” function. It builds a complete new query — a new context, a new query id, a new process list entry, a new pipeline — and runs it. That is the only way to get all the downstream machinery (materialized views, quotas, query_log) to behave exactly as if the client had sent one big insert.

cpp
// src/Interpreters/AsynchronousInsertQueue.cpp:1261
    /// Declared before `insert_context` so it's freed before the detach below reparents the tracker to
    /// `total_memory_tracker`; it accumulates access info charged to the inserting user while attached.
    DB::QueryScope query_scope;

    auto insert_context = Context::createCopy(global_context);
    insert_context->makeQueryContext();

The context is a copy of the global context, not of the client’s context — because the client’s context is gone by the time the flush runs. The identity has to be replayed by hand:

cpp
// src/Interpreters/AsynchronousInsertQueue.cpp:1275
    /// Access rights must be checked for the user who executed the initial INSERT query.
    if (key.user_id)
    {
        /// Replay the whole originating identity in one call: the external (pushed) roles, the
        /// credential grant limit and its expiry are restored together with the user, so a limited
        /// credential does not regain the full user's rights and a pushed-role session does not fail
        /// role revalidation.
        insert_context->setUser(*key.user_id, key.external_roles, key.authentication_grants, key.authentication_valid_until);
        /// `current_roles` are the session's *effective* current roles, which already include the
        /// external roles restored above. Re-apply them without the grant check: external roles are not
        /// locally granted, so a checked re-apply would throw `SET_NON_GRANTED_ROLE`; the locally
        /// granted current roles are kept and the external ones come from `setUser`.
        insert_context->setCurrentRoles(key.current_roles, /*check_grants=*/ false);
    }

and the settings have to be restored too, because the flush is a different query with its own settings:

cpp
// src/Interpreters/AsynchronousInsertQueue.cpp:1300
    insert_context->setSettings(*key.settings);

The credential expiry is re-checked at flush time, which is a genuinely nice property — the check the session would have done per query is done once, at the moment the data is actually written:

cpp
// src/Interpreters/AsynchronousInsertQueue.cpp:1237
    /// Fail closed if the authentication method that queued this insert has expired between enqueue
    /// and flush. The synchronous path re-checks per query in `Session::checkIfUserIsStillValid`; the
    /// deferred flush has no session, so without this a token could enqueue data just before expiry
    /// and have the server flush it afterwards. The per-method expiry is part of the batching key, so
    /// every entry in this batch shares it and they all fail closed together. 0 means no expiry.
    if (key.authentication_valid_until != 0)
    {
        const time_t now = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
        if (now > key.authentication_valid_until)
            throw Exception(ErrorCodes::USER_EXPIRED, "Authentication method used to submit the deferred insert has expired");
    }

Then the interpreter is built and run, exactly as for a synchronous insert:

cpp
// src/Interpreters/AsynchronousInsertQueue.cpp:1443
    try
    {
        interpreter = std::make_unique<InterpreterInsertQuery>(
            key.query,
            insert_context,
            (*key.settings)[Setting::insert_allow_materialized_columns],
            false,
            false,
            true);

        pipeline = interpreter->execute().pipeline;
        chassert(pipeline.pushing());

The rows: parse everything, insert once

The flush then parses every entry into a single chunk and pushes that one chunk through the pipeline. This is where the per-entry dedup token is re-attached — the token travels with the data, not with the buffer:

cpp
// src/Interpreters/AsynchronousInsertQueue.cpp:1618
    StreamingFormatExecutor executor(
        header,
        format,
        std::move(on_error),
        data->size_in_bytes,
        data->entries.size(),
        std::move(adding_defaults_transform),
        [query_status = insert_context->getProcessListElement()] { return query_status && query_status->isKilled(); });

    auto deduplication_info = DeduplicationInfo::create(/*async_insert=*/true);

    for (const auto & entry : data->entries)
    {
        current_entry = entry;

        const auto * bytes = entry->chunk.asString();
        if (!bytes)
            throw Exception(ErrorCodes::LOGICAL_ERROR,
                "Expected entry with data kind Parsed. Got: {}", entry->chunk.getDataKind());

        auto buffer = std::make_unique<ReadBufferFromString>(*bytes);
        executor.setQueryParameters(entry->query_parameters);

        size_t num_bytes = bytes->size();
        size_t num_rows = executor.execute(*buffer, num_bytes);

        total_rows += num_rows;

        /// it is ok if total_rows is 0 here or async_dedup_token is empty
        deduplication_info->setUserToken(entry->async_dedup_token, num_rows);

        add_to_async_insert_log(entry, current_exception, num_rows, num_bytes);
        current_exception.clear();
        entry->resetChunk();
    }

A StreamingFormatExecutor is used rather than a per-entry InputFormat: the executor keeps the parser state alive across entries, so 450 CSV payloads are parsed as one continuous stream. The on_error callback is what makes a single malformed entry fail only itself:

cpp
// src/Interpreters/AsynchronousInsertQueue.cpp:1605
    auto on_error = [&](const MutableColumns & result_columns, const ColumnCheckpoints & checkpoints, Exception & e)
    {
        current_exception = e.displayText();
        LOG_ERROR(logger, "Failed parsing for insert query id {}. {}",
            current_entry->query_id, current_exception);

        for (size_t i = 0; i < result_columns.size(); ++i)
            result_columns[i]->rollback(*checkpoints[i]);

        current_entry->finish(std::current_exception());
        return 0;
    };

rollback(*checkpoints[i]) unwinds the partially-parsed columns so the bad row does not poison the good rows that follow it in the same buffer. The bad entry’s promise is fulfilled with the exception, so its client gets an error, while the other 449 clients get their rows.

Note the two data kinds. If the client sent a pre-parsed block (the native protocol path), the data is already a Block and the entries are simply concatenated after a header conversion:

cpp
// src/Interpreters/AsynchronousInsertQueue.cpp:1690
            if (!isCompatibleHeader(block_to_insert, header))
                convertBlockToHeader(block_to_insert, header, context_);

The client side: two different contracts

There are two entry points into the queue, and they answer the client differently.

Inline data (INSERT INTO t VALUES (...) over HTTP, or the data part of a native-protocol query) goes through executeQuery:

cpp
// src/Interpreters/executeQuery.cpp:3123
            auto result = queue->pushQueryWithInlinedData(out_ast, context);

            if (result.status == AsynchronousInsertQueue::PushResult::OK)
            {
                // Increment InsertQuery for async insert with inline data
                ProfileEvents::increment(ProfileEvents::InsertQuery);

                if (settings[Setting::wait_for_async_insert])
                {
                    auto timeout = saturatedMilliseconds(settings[Setting::wait_for_async_insert_timeout].totalMilliseconds()).count();
                    auto source = std::make_shared<WaitForAsyncInsertSource>(
                        std::move(result.future),
                        timeout,
                        context->getProcessListElement(),
                        context->getProgressCallback(),
                        /* report_read_progress */ true);
                    res.pipeline = QueryPipeline(Pipe(std::move(source)));
                    res.pipeline.complete(std::make_shared<NullOutputFormat>(std::make_shared<const Block>(Block())));
                }

Block data (the native protocol, where the client sends Blocks) goes through TCPHandler:

cpp
// src/Server/TCPHandler.cpp:1595
    if (insert_queue && async_insert_enabled && !insert_query.select)
    {
        auto result = processAsyncInsertQuery(state, *insert_queue);
        if (result.status == AsynchronousInsertQueue::PushResult::OK)
        {
            /// Reset pipeline because it may hold write lock for some storages.
            state.io.resetPipeline(/*cancel=*/true);
            if (settings[Setting::wait_for_async_insert])
            {
                auto timeout = saturatedMilliseconds(settings[Setting::wait_for_async_insert_timeout].totalMilliseconds());
                size_t timeout_ms = timeout.count();
                auto wait_status = result.future.wait_for(timeout);

                if (wait_status == std::future_status::deferred)
                    throw Exception(ErrorCodes::LOGICAL_ERROR, "Got future in deferred state");

                if (wait_status == std::future_status::timeout)
                    throw Exception(ErrorCodes::TIMEOUT_EXCEEDED, "Wait for async insert timeout ({} ms) exceeded", timeout_ms);

                auto progress_result = result.future.get();

The two paths differ in how they wait (a source processor that participates in the query pipeline vs. a blocking future::wait_for), but the semantics are the same.

wait_for_async_insert defaults to true, so the default behaviour is “the client is told the truth”:

cpp
// src/Core/Settings.cpp:8520
    DECLARE(Bool, wait_for_async_insert, true, R"(
If true wait for processing of asynchronous insertion.
)", 0) \

With wait_for_async_insert = 0, the query returns as soon as the data is in the buffer, and the guarantee is only “the server has taken responsibility for these bytes”. The query_log documentation spells out the difference:

cpp
// src/Interpreters/QueryLog.cpp:80
        {"written_rows", std::make_shared<DataTypeUInt64>(), "The number of rows written by the query, including any rows written by downstream inserts triggered by the pipeline, such as attached materialized views. For a synchronous insert these downstream rows are recorded on the `query_kind` = `Insert` entry; for an asynchronous insert they are recorded on the `query_kind` = `AsyncInsertFlush` entry, while the client-facing `Insert` entry records only the rows accepted from the client, and only when `wait_for_async_insert = 1` makes it wait for that flush result; with `wait_for_async_insert = 0` the client-facing `Insert` entry returns before the flush and this field stays 0. For queries that do not write rows, it is 0."},

So there are two query_log rows for one async insert: the client-facing Insert row and the server-side AsyncInsertFlush row.

The escape hatch: too much data

If a single insert is larger than async_insert_max_data_size, buffering it would be silly — the server would hold 100 MiB of client data in RAM and then write it. So the producer reads at most the limit, and if there is more left over, it hands the rest back to the caller, which falls back to a synchronous insert:

cpp
// src/Interpreters/AsynchronousInsertQueue.cpp:638
        if (!read_buf->eof())
        {
            /// Concat read buffer with already extracted from insert
            /// query data and with the rest data from insert query.
            ConcatReadBuffer::Buffers buffers;
            buffers.emplace_back(std::make_unique<ReadBufferFromOwnString>(bytes));
            buffers.emplace_back(std::move(read_buf));

            return PushResult
            {
                .status = PushResult::TOO_MUCH_DATA,
                .future = {},
                .insert_data_buffer = std::make_unique<ConcatReadBuffer>(std::move(buffers)),
            };
        }

The PushResult::TOO_MUCH_DATA status is handled identically in both entry points, and the “rest” of the data is not lost: the caller re-attaches it to the query and runs the whole thing synchronously.

Things that are refused

Some combinations simply do not make sense, and the queue rejects them at the door rather than at flush time.

Quorum inserts. A non-parallel quorum insert allows exactly one in-flight quorum part per table, which a flush (which writes the data of several queries at once) cannot honour:

cpp
// src/Interpreters/AsynchronousInsertQueue.cpp:686
    auto quorum_is_enabled = settings[Setting::insert_quorum].valueOr(0) > 1 || settings[Setting::insert_quorum].is_auto;
    if (quorum_is_enabled && !settings[Setting::insert_quorum_parallel])
        throw Exception(
            ErrorCodes::UNSUPPORTED_PARAMETER,
            "Async inserts with quorum only make sense with enabled insert_quorum_parallel setting, either disable quorum "
            "or set insert_quorum_parallel=1 or do not use async inserts");

Transactions. Async inserts inside a transaction are rejected, because the flush happens after the transaction would have committed (or rolled back).

INSERT ... SELECT with views. The INSERT ... SELECT route to the queue is only taken when the destination has no dependent materialized views, because a diverted block leaves the views’ sink chain built but unfed.

Trade-offs

What it buys. One part per 450 inserts instead of one part per insert. The part is the expensive object: metadata, checksums, ZooKeeper transaction, merges. Amortising it over hundreds of statements is a 10-100× reduction in per-statement overhead. The number of round trips does not change — the client still sends one statement per batch — but the number of parts the server has to create, merge and later compact does.

What it costs.

Latency. The data is invisible to SELECT until the flush. With the default 50-200 ms adaptive window, a freshly inserted row may not be queryable for up to ~200 ms. This is the classic “read-your-writes” problem, and it is why wait_for_async_insert defaults to true — the client waits for the flush, so the client knows the data is visible, but other clients still have to wait for the window.

Memory. The buffer holds up to async_insert_max_data_size bytes per key per shard. With many distinct keys (many clients, many settings) this multiplies. The MemoryTracker per entry keeps the charge on the user that pushed the data, not on the server, so a runaway client is stopped by its own quota:

cpp
// src/Interpreters/AsynchronousInsertQueue.cpp:310
void AsynchronousInsertQueue::InsertData::Entry::resetChunk()
{
    /// Released against the user that pushed the data, never the flush, whose user is whoever's insert it
    /// happens to be flushing. The parameters were charged the same way and go with it.
    if (queued_data_tracker)
    {
        MemoryTrackerSwitcher switcher(queued_data_tracker.get());
        chunk = {};
        query_parameters.clear();
        return;
    }

Error attribution. A batch of 450 inserts is one part. If the part cannot be written, all 450 clients get the error — which is correct, but it means a single bad row in one entry can fail the whole batch if the parse error is not isolated. The on_error rollback above is what prevents that.
The key is a hash. Two inserts whose keys differ only in a way the hash cannot see would be merged. The hash covers the AST structurally, the user, the roles, the credential limit and expiry, the identity strings, and every non-default setting — that is a lot of coverage, but it is coverage by construction, not by proof.

The bottom line

The async insert queue is a content-addressed batcher. The key is a UInt128 SipHash over everything that could make two inserts behave differently; the value is a list of entries, each carrying its own promise. Three independent triggers — bytes, entry count, deadline — decide when a buffer is handed to a thread pool, and the flush replays the whole query machinery (context, identity, settings, process list, query_log) so that a deferred insert is indistinguishable from a synchronous one, except for its query id.

The most interesting design decision is not the batching but the replay: the flush does not write blocks to a table, it runs a query. That is what makes materialized views, quotas, row policies and query_log keep working on the deferred path — and it is also why the queue has to carry so much identity in its key.

The thing people get wrong

async_insert = 1 does not mean “faster”. It means “the server will batch this for you, and the data will not be visible for a short while”. If your workload needs read-your-writes, the answer is wait_for_async_insert = 1 (turn the waiting on) or async_insert = 0 (turn the batching off) — not wait_for_async_insert = 0, which just means “I do not want to know whether it worked”.

And if you are seeing rows appear in batches of 450 rather than batches of 10 000, check async_insert_max_data_size (10 MiB) and async_insert_busy_timeout_max_ms (200 ms): one of the two is your throughput limit, and the other is your latency limit.