Opensource internals

ClickHouse's Native Protocol Is a Half-Duplex State Machine

20 min read Jay Hubbert

ClickHouse's TCP handler is a half-duplex state machine: one thread pulls result blocks, checks for a Cancel packet on a timer, and piggybacks logs, progress and profile events onto the same output stream.

TL;DR: The native protocol handler is not a request/response loop — it is a half-duplex state machine. One thread owns the socket; it pulls a result block, and at most every interactive_delay (100 ms by default) it peeks at the input for a Cancel packet, then flushes progress, logs and profile events into the same output stream before the next block goes out.

There is a shape of server that most database wire protocols fall into: the client sends a command, the server computes, the server replies, and the client decides what to send next. MySQL works this way. PostgreSQL works this way. HTTP works this way (each request is a fresh transaction on the connection).

ClickHouse’s native protocol does not work this way, and the difference is the source of a lot of its behaviour — including the fact that a query can be cancelled mid-flight, that clickhouse-client shows a live progress bar while the server is still computing, and that internal server logs appear interleaved with result rows in the terminal.

The reason is a single design decision: the connection thread never blocks for long without checking the input socket. It cannot, because the only way to learn that the user pressed Ctrl-C is to read the socket, and the only way to show a live progress bar is to write to the socket while the pipeline is still running.

This article walks the whole machine: the packet vocabulary, the four receive states, the pull loop that fuses “compute” with “listen”, the piggybacking trick that makes progress/logs free, and the cancellation machinery that hangs off it.

The vocabulary

Both sides of the wire speak in VarUInt packet types. The server’s set:

cpp 
// src/Core/Protocol.h:77
        enum Enum
        {
            Hello = 0,                      /// Name, version, revision.
            Data = 1,                       /// A block of data (compressed or not).
            Exception = 2,                  /// The exception during query execution.
            Progress = 3,                   /// Query execution progress: rows read, bytes read.
            Pong = 4,                       /// Ping response
            EndOfStream = 5,                /// All packets were transmitted
            ProfileInfo = 6,                /// Packet with profiling info.
            Totals = 7,                     /// A block with totals (compressed or not).
            Extremes = 8,                   /// A block with minimums and maximums (compressed or not).
            TablesStatusResponse = 9,       /// A response to TablesStatus request.
            Log = 10,                       /// System logs of the query execution
            TableColumns = 11,              /// Columns' description for default values calculation
            PartUUIDs = 12,                 /// Obsolete: was used for query deduplication based on part UUIDs (allow_experimental_query_deduplication). Kept for protocol compatibility.
            ReadTaskRequest = 13,           /// String (UUID) describes a request for which next task is needed
                                            /// This is such an inverted logic, where server sends requests
                                            /// And client returns back response
            ProfileEvents = 14,             /// Packet with profile events from server.
            MergeTreeAllRangesAnnouncement = 15,
            MergeTreeReadTaskRequest = 16,  /// Request from a MergeTree replica to a coordinator
            TimezoneUpdate = 17,            /// Receive server's (session-wide) default timezone
            SSHChallenge = 18,              /// Return challenge for SSH signature signing
            MAX = SSHChallenge,

        };

And the client’s:

cpp 
// src/Core/Protocol.h:117
        enum Enum
        {
            Hello = 0,                      /// Name, version, revision, default DB
            Query = 1,                      /// Query id, query settings, stage up to which the query must be executed,
                                            /// whether the compression must be used,
                                            /// query text (without data for INSERTs).
            Data = 2,                       /// A block of data (compressed or not).
            Cancel = 3,                     /// Cancel the query execution.
            Ping = 4,                       /// Check that connection to the server is alive.
            TablesStatusRequest = 5,        /// Check status of tables on the server.
            KeepAlive = 6,                  /// Keep the connection alive
            Scalar = 7,                     /// A block of data (compressed or not).
            IgnoredPartUUIDs = 8,           /// Obsolete: was used for query deduplication based on part UUIDs (allow_experimental_query_deduplication). Kept for protocol compatibility.
            ReadTaskResponse = 9,           /// A filename to read from s3 (used in s3Cluster)
            MergeTreeReadTaskResponse = 10, /// Coordinator's decision with a modified set of mark ranges allowed to read

            SSHChallengeRequest = 11,       /// Request SSH signature challenge
            SSHChallengeResponse = 12,      /// Reply to SSH signature challenge

            QueryPlan = 13,                 /// Query plan

            MergeTreeAllRangesAnnouncementResponse = 14,
                                            /// Initiator's reply to a follower's announcement,
                                            /// carrying the authoritative parts list for the stream.

            MAX = MergeTreeAllRangesAnnouncementResponse,
        };

Two of these matter enormously for the state machine: Data/Scalar (the client streaming data into the server) and Cancel (the client trying to interrupt the server while it is streaming data out).

The shape of the machine

text 
                    +-----------------------------------+
                    |          runImpl() loop           |
                    |  (one query per iteration)        |
                    +-----------------------------------+
                                     |
        +----------------------------+----------------------------+
        |                            |                            |
        v                            v                            v
  expect-query                 expect-data                 expect-cancel
  (idle, waiting               (INSERT: client is          (SELECT: client is
   for a command)                streaming rows in)           idle, but we poll
        |                            |                            |  for Cancel)
        |                            |                            |
  +-----+-----+                 +-----+-----+               +-----+-----+
  | Hello | -> err             | Data | -> ok              | Cancel| -> throw
  | Query | -> execute         | Cancel-> throw             | (poll(0))
  | Ping  | -> Pong            | Query | -> err             |   |
  | Tables| -> status          | Ping  | -> Pong            |   v
  | Cancel| -> ignore          +-----+-----+            return; "nothing
  +-----+-----+                                                  to cancel"

Three of the four states share the same trick: they are not blocking reads. receivePacketsExpectData and receivePacketsExpectCancel both poll the socket with a timeout, and the “expect-query” state only ever does a blocking read because there is genuinely nothing to do but wait.

State 1: expecting a command

cpp 
// src/Server/TCPHandler.cpp:1290
bool TCPHandler::receivePacketsExpectQuery(std::shared_ptr<QueryState> & state)
{
    UInt64 packet_type = 0;
    readVarUInt(packet_type, *in);

    switch (packet_type)
    {
        case Protocol::Client::Hello:
            throwUnexpectedPacket(packet_type);

        case Protocol::Client::Data:
        case Protocol::Client::Scalar:
            /// The payload is deliberately left unread: nothing consumes it here (the connection
            /// closes with no reply), and in interserver mode the connection is not authenticated
            /// until the Query packet, so reading it would deserialize a peer-chosen type.
            ///
            /// Name that case for what it is: an unauthenticated peer, so that `runImpl` answers it
            /// the way it answers every other interserver authentication failure.
            if (is_interserver_mode && !is_interserver_authenticated)
                throw Exception(ErrorCodes::AUTHENTICATION_FAILED,
                    "Unexpected data packet received before interserver authentication");
            throwUnexpectedPacket(packet_type);

        case Protocol::Client::Ping:
            writeVarUInt(Protocol::Server::Pong, *out);
            out->finishChunk();
            out->sync();
            return false;

        case Protocol::Client::Cancel:
            return false;

        case Protocol::Client::TablesStatusRequest:
            processTablesStatusRequest();
            return false;

        case Protocol::Client::IgnoredPartUUIDs:
            processObsoleteIgnoredPartUUIDs();

        case Protocol::Client::Query:
            processQuery(state);
            return true;

        default:
            throw Exception(ErrorCodes::UNKNOWN_PACKET_FROM_CLIENT, "Unknown packet {} from client", toString(packet_type));
    }
}

Note the return convention: true means “a query arrived, start executing it”; false means “nothing to do, go back to the top of the loop”. Ping is answered inline and returns false. Cancel outside a query is simply swallowed — there is nothing running to cancel.

State 2: expecting data (the INSERT side)

cpp 
// src/Server/TCPHandler.cpp:1345
bool TCPHandler::receivePacketsExpectData(QueryState & state)
{
    /// Poll interval should not be greater than receive_timeout
    constexpr UInt64 min_timeout_us = 5000; // 5 ms
    UInt64 timeout_us = std::max(
            min_timeout_us,
            std::min(
                poll_interval * 1000000,
                static_cast<UInt64>(receive_timeout.totalMicroseconds())));

    Stopwatch watch;

    while (tcp_server.isOpen() || !server.isCancelled())
    {
        while (!in->poll(timeout_us))
        {
            size_t elapsed = size_t(watch.elapsedSeconds());
            if (elapsed > size_t(receive_timeout.totalSeconds()))
            {
                throw NetException(ErrorCodes::SOCKET_TIMEOUT,
                                "Timeout exceeded while receiving data from client. Waited for {} seconds, timeout is {} seconds.",
                                elapsed, receive_timeout.totalSeconds());
            }
        }

        UInt64 packet_type = 0;
        /// The type is a varuint, so a client can leave the input mid-packet by truncating it.
        state.at_packet_boundary = false;
        readVarUInt(packet_type, *in);

        switch (packet_type)
        {
            case Protocol::Client::Query:
            case Protocol::Client::Hello:
            case Protocol::Client::TablesStatusRequest:
                throwUnexpectedPacket(packet_type);

            case Protocol::Client::IgnoredPartUUIDs:
                processObsoleteIgnoredPartUUIDs();

            case Protocol::Client::Data:
            case Protocol::Client::Scalar:
            {
                bool empty_block = false;
                if (state.skipping_data)
                    empty_block = !skipDataPacket(state);
                else
                    empty_block = !processData(state, packet_type == Protocol::Client::Scalar);
                if (empty_block)
                    state.read_all_data = true;
                return !empty_block;
            }

            case Protocol::Client::QueryPlan:
                return receiveQueryPlan(state);

            case Protocol::Client::Ping:
                state.at_packet_boundary = true;  /// no body
                writeVarUInt(Protocol::Server::Pong, *out);
                out->finishChunk();
                out->sync();
                continue;

            case Protocol::Client::Cancel:
                state.at_packet_boundary = true;  /// no body, and `processCancel` throws
                processCancel(state);
                return false; // We return false from this function as if no more data received

            default:
                throw Exception(ErrorCodes::UNKNOWN_PACKET_FROM_CLIENT, "Unknown packet {} from client", toString(packet_type));
        }
    }

    chassert(!tcp_server.isOpen() && server.isCancelled());
    throw Exception(ErrorCodes::ABORTED, "Server shutdown is called");
}

Two details worth pausing on.

First, the minimum timeout of 5 ms. Even if poll_interval is 1 second, the loop wakes at least every 5 ms so that a cancellation is noticed promptly. That is the price of responsiveness.

Second, the at_packet_boundary flag. The packet type is a VarUInt, so a client can hang up in the middle of writing one. The flag is cleared before the type is read and set again only once the whole packet has been consumed:

cpp 
// src/Server/TCPHandler.h:133
    /// Whether the input is positioned at the start of a packet. Cleared before the packet type is
    /// read and set again once the packet is fully consumed, so any throw in between leaves it
    /// cleared. Input starting mid-packet can only be discarded with the connection.
    bool at_packet_boundary = true;

So if anything throws halfway through a packet, the handler knows the stream is desynchronised and the only safe recovery is to drop the connection — not to try to resynchronise on a byte boundary.

State 3: expecting a cancel (the SELECT side)

This is the interesting one, because it runs while the query is executing.

cpp 
// src/Server/TCPHandler.cpp:3170
void TCPHandler::receivePacketsExpectCancel(QueryState & state, bool force)
{
    if (!force && after_check_cancelled.elapsed() / 1000 < interactive_delay)
        return;

    after_check_cancelled.restart();

    /// During request execution the only packet that can come from the client is stopping the query.
    if (in->poll(0))
    {
        try
        {
            if (in->isCanceled() || in->eof())
                throw NetException(ErrorCodes::ABORTED, "Client has dropped the connection, cancel the query.");

            UInt64 packet_type = 0;
            state.at_packet_boundary = false;
            readVarUInt(packet_type, *in);

            switch (packet_type)
            {
                case Protocol::Client::Cancel:
                    state.at_packet_boundary = true;  /// no body, and `processCancel` throws
                    processCancel(state);
                    break;

                case Protocol::Client::IgnoredPartUUIDs:
                    processObsoleteIgnoredPartUUIDs();

                default:
                    throw NetException(ErrorCodes::UNKNOWN_PACKET_FROM_CLIENT, "Unknown packet from client {}", toString(packet_type));
            }
        }
        catch (...)
        {
            /// Mark the query as stopped so that other callbacks (e.g. ClusterFunctionReadTaskCallback)
            /// that are waiting on callback_mutex will see the cancellation via checkIfQueryCanceled()
            /// before attempting to read from the now-broken connection.
            /// Without this, a race condition exists: when this function throws while holding callback_mutex,
            /// the mutex is released during stack unwinding, and a pipeline worker thread can acquire it
            /// and attempt to read from the canceled ReadBuffer before the executor is canceled.
            state.stop_query = true;
            throw;
        }
    }
}

in->poll(0) — a zero-timeout poll. It never blocks; it just asks “is there a byte waiting?” And the rate limit is interactive_delay:

cpp 
// src/Core/Settings.cpp:467
    DECLARE(UInt64, interactive_delay, 100000, R"(
The interval in microseconds for checking whether request execution has been canceled and sending the progress.
)", 0) \

100 milliseconds. So the server polls the input socket for a Cancel at most ten times a second while a query is running. That is the entire cancellation latency budget for a running SELECT.

Fusing “compute” with “listen”

Now the interesting part: how does the connection thread both wait for query output and poll the socket? The answer is PullingAsyncPipelineExecutor, which runs the pipeline on a separate thread and hands finished blocks to the connection thread through a LazyOutputFormat with a timed getChunk().

cpp 
// src/Processors/Executors/PullingAsyncPipelineExecutor.cpp:116
bool PullingAsyncPipelineExecutor::pull(Chunk & chunk, uint64_t milliseconds)
{
    if (!data)
    {
        data = std::make_unique<Data>();
        data->executor = createExecutor(pipeline.processors, pipeline.process_list_element);
        data->executor->setReadProgressCallback(pipeline.getReadProgressCallback());
        data->executor->setStepProfiler(pipeline.getStepProfiler());
        data->lazy_format = lazy_format.get();

        auto func = [&, thread_group = CurrentThread::getGroup()]()
        {
            threadFunction(*data, thread_group, pipeline.getNumThreads(), pipeline.getConcurrencyControl());
        };

        data->thread = ThreadFromGlobalPool(std::move(func));

        /// Simulates a consumer that is slow to pull the first chunks, so that the time limit is exceeded while
        /// the result is already sitting in the lazy format. Only a query whose id starts with the prefix is
        /// delayed, so that the failpoint does not affect the queries of concurrently running tests.
        fiu_do_on(FailPoints::pulling_async_pipeline_executor_delay_first_pull,
        {
            if (CurrentThread::getQueryId().starts_with(pulling_async_pipeline_executor_delay_first_pull_query_id_prefix))
                sleepForMilliseconds(2000);
        });
    }

    data->rethrowExceptionIfHas();

    /// Throws when the time limit is exceeded with `timeout_overflow_mode = 'throw'`. With 'break' the partial result
    /// is returned as a success: the execution is cancelled with `CancelledByTimeout`, so the pipeline finishes on
    /// its own, the chunks that are already in the lazy format are pulled as usual, and the format is finalized by
    /// the executor - only then is the end of the data reported.
    if (pipeline.process_list_element && !pipeline.process_list_element->checkTimeLimitSoft())
    {
        data->executor->cancel(IProcessor::CancelReason::CancelledByTimeout);
        pipeline.process_list_element->checkTimeLimit();
    }

    const bool execution_finished = lazy_format ? lazy_format->isFinished() : data->is_finished.load();
    if (execution_finished)
    {
        /// If lazy format is finished, we don't cancel pipeline but wait for main thread to be finished.
        data->is_finished = true;
        /// Wait thread and rethrow exception if any.
        cancel();
        return false;
    }

    /// When a cancel callback is set and no explicit timeout was requested, use the interactive timeout
    /// to periodically poll the callback (e.g. to check for Cancel packets during scalar subquery execution).
    uint64_t effective_timeout = milliseconds;
    if (cancel_callback && milliseconds == 0)
        effective_timeout = interactive_timeout_ms;

    chunk = lazy_format->getChunk(effective_timeout);

    if (cancel_callback)
        cancel_callback();

    data->rethrowExceptionIfHas();
    return true;
}

getChunk(effective_timeout) returns an empty chunk if nothing arrived within the timeout. That empty chunk is the signal that the connection thread has stopped waiting and must do its periodic chores.

And here is the loop that uses it:

cpp 
// src/Server/TCPHandler.cpp:1656
void TCPHandler::processOrdinaryQuery(QueryState & state)
{
    auto & pipeline = state.io.pipeline;

    const bool discard_query_data = state.query_context->getSettingsRef()[Setting::discard_query_data]
        && state.query_context->getClientInfo().query_kind == ClientInfo::QueryKind::INITIAL_QUERY;

    /// Send header-block, to allow client to prepare output format for data to send.
    {
        const auto & header = pipeline.getHeader();

        if (!header.empty())
        {
            sendData(state, header);
            out->sync();
        }
    }

    {
        PullingAsyncPipelineExecutor executor(pipeline);
        pipeline.setConcurrencyControl(state.query_context->getSettingsRef()[Setting::use_concurrency_control]);
        CurrentMetrics::Increment query_thread_metric_increment{CurrentMetrics::QueryThread};

        try
        {
            Block block;
            while (executor.pull(block, interactive_delay / 1000))
            {
                bool stop_read_return_partial_result = false;
                {
                    std::lock_guard lock(*callback_mutex);
                    receivePacketsExpectCancel(state);
                    stop_read_return_partial_result = state.stop_read_return_partial_result;
                }

                if (stop_read_return_partial_result)
                {
                    executor.cancelReading();
                }

                {
                    std::lock_guard lock(*callback_mutex);

                    if (after_send_progress.elapsed() / 1000 >= interactive_delay)
                    {
                        /// Some time passed and there is a progress.
                        after_send_progress.restart();
                        sendInteractiveUpdates(state);
                    }
                    else
                        sendLogs(state);

                    // Block might be empty in case of timeout, i.e. there is no data to process
                    if (!block.empty() && !state.io.null_format && !discard_query_data)
                        sendData(state, block);

                    out->sync();
                }
            }
        }
        catch (...)
        {
            executor.cancel();
            throw;
        }

Read that loop carefully. It is the whole trick:

  1. executor.pull(block, interactive_delay / 1000) blocks for at most 100 ms waiting for a result block. If nothing arrives, it returns true with an empty block — a timeout, not an end-of-stream.
  2. Under callback_mutex, receivePacketsExpectCancel(state) polls the input socket for a Cancel packet.
  3. If at least interactive_delay has passed since the last progress report, sendInteractiveUpdates fires (progress + profile events + logs). Otherwise only the logs are flushed.
  4. If the block is non-empty, it goes out as a Data packet.
  5. out->sync() pushes the whole thing to the socket.

So the connection thread wakes up at most every 100 ms whether or not there is data, and on every wake it does four things: check for cancellation, maybe report progress, maybe flush logs, maybe send data.

The mutex is the keystone. It is declared with an explicit contract:

cpp 
// src/Server/TCPHandler.h:269
    /// `callback_mutex` protects using `out` (WriteBuffer), `in` (ReadBuffer) and other members concurrent inside callbacks.
    /// All the methods which are run inside callbacks are marked with TSA_REQUIRES.
    std::shared_ptr<std::mutex> callback_mutex = std::make_shared<std::mutex>();

The TSA_REQUIRES(callback_mutex) annotations on receivePacketsExpectData, receivePacketsExpectCancel, sendLogs, sendInteractiveUpdates and friends are enforced by the thread-safety-analysis compiler pass. The mutex exists because pipeline worker threads can call back into the handler: to ask for a new read task in distributed mode, to exchange MergeTree read tasks, to send a read-task response. All of those need the socket, and the socket is a single object with a single cursor.

Piggybacking: the “free” progress bar

Here is the trick that makes the whole design affordable. Progress, logs and profile events are not separate packets sent on their own schedule. They are appended to the same output stream, between data packets, at a moment when the server was already going to write something.

cpp 
// src/Server/TCPHandler.cpp:3439
void TCPHandler::sendInteractiveUpdates(QueryState & state)
{
    sendProgress(state);
    sendSelectProfileEvents(state);
    sendLogs(state);
    out->sync();
}

Three logical messages, one sync(). And progress itself is a delta, not a cumulative value, which is what makes it cheap to interleave:

cpp 
// src/Server/TCPHandler.cpp:3426
void TCPHandler::sendProgress(QueryState & state)
{
    writeVarUInt(Protocol::Server::Progress, *out);
    auto increment = state.progress.fetchValuesAndResetPiecewiseAtomically();
    UInt64 current_elapsed_ns = state.watch.elapsedNanoseconds();
    increment.elapsed_ns = current_elapsed_ns - state.prev_elapsed_ns;
    state.prev_elapsed_ns = current_elapsed_ns;
    increment.write(*out, client_tcp_protocol_version);

    out->finishChunk();
}

fetchValuesAndResetPiecewiseAtomically is a set of fetch_and(0) calls on std::atomic counters, one per field, so the value the client sees is “what happened since the last progress packet”:

cpp 
// src/IO/Progress.cpp:182
ProgressValues Progress::fetchValuesAndResetPiecewiseAtomically()
{
    ProgressValues res;

    res.read_rows = read_rows.fetch_and(0);
    res.read_bytes = read_bytes.fetch_and(0);

    res.total_rows_to_read = total_rows_to_read.fetch_and(0);
    res.total_bytes_to_read = total_bytes_to_read.fetch_and(0);

    res.written_rows = written_rows.fetch_and(0);
    res.written_bytes = written_bytes.fetch_and(0);

    res.result_rows = result_rows.fetch_and(0);
    res.result_bytes = result_bytes.fetch_and(0);

    res.elapsed_ns = elapsed_ns.fetch_and(0);

    res.memory_usage = memory_usage.fetch_and(0);

    return res;
}

“Piecewise” is the honest word: each field is atomically reset independently, so a progress packet is not a consistent snapshot. For a progress bar, that is fine — and it means the counters can be updated by any number of pipeline threads without a lock:

cpp 
// src/Server/TCPHandler.cpp:3420
void TCPHandler::updateProgress(QueryState & state, const Progress & value)
{
    state.progress.incrementPiecewiseAtomically(value);
}

The logs go through the same channel. They are collected into a queue by the logging thread and drained in batches:

cpp 
// src/Server/TCPHandler.cpp:3453
void TCPHandler::sendLogs(QueryState & state, std::shared_ptr<TCPHandlerPocoChunkedWriter> out, UInt32 client_tcp_protocol_version)
{
    if (!state.logs_queue)
        return;

    MutableColumns logs_columns;
    MutableColumns curr_logs_columns;
    size_t rows = 0;

    for (; state.logs_queue->tryPop(curr_logs_columns); ++rows)
    {
        if (rows == 0)
        {
            logs_columns = std::move(curr_logs_columns);
        }
        else
        {
            for (size_t j = 0; j < logs_columns.size(); ++j)
                logs_columns[j]->insertRangeFrom(*curr_logs_columns[j], 0, curr_logs_columns[j]->size());
        }
    }

    if (rows > 0)
    {
        Block block = InternalTextLogsQueue::getSampleBlock();
        block.setColumns(std::move(logs_columns));
        sendLogData(state, block, out, client_tcp_protocol_version);
    }
}

The queue is only created at all if the client can receive it and has not turned it off:

cpp 
// src/Server/TCPHandler.cpp:697
            /// Should we send internal logs to client?
            const auto client_logs_level = query_state->query_context->getSettingsRef()[Setting::send_logs_level];
            if (client_tcp_protocol_version >= DBMS_MIN_REVISION_WITH_SERVER_LOGS
                && client_logs_level != LogsLevel::none)
            {
                query_state->logs_queue = std::make_shared<InternalTextLogsQueue>();
                query_state->logs_queue->max_priority = Poco::Logger::parseLevel(client_logs_level.toString());
                query_state->logs_queue->setSourceRegexp(query_state->query_context->getSettingsRef()[Setting::send_logs_source_regexp]);
                CurrentThread::attachInternalTextLogsQueue(query_state->logs_queue, client_logs_level);
            }

So the cost of “no live logs, no live progress events” is one pointer check per iteration, and the cost of “yes, live logs” is amortised over the whole batch.

The end of a query

When the pipeline finally runs dry, the handler emits a fixed, ordered sequence of packets:

cpp 
// src/Server/TCPHandler.cpp:1731
        std::lock_guard lock(*callback_mutex);

        receivePacketsExpectCancel(state);

        if (!discard_query_data)
        {
            sendTotals(state, executor.getTotalsBlock());
            sendExtremes(state, executor.getExtremesBlock());
        }
        sendProfileInfo(state, executor.getProfileInfo());
        sendProgress(state);
        sendLogs(state);
        sendSelectProfileEvents(state);

        sendData(state, {});

        sendProgress(state);
    }

The ordering is not arbitrary. Totals and Extremes can only be sent once the pipeline has finished, because they are computed by the aggregation transform at the very end. The comment above explains why they are not sent on the cancellation path:

cpp 
// src/Server/TCPHandler.cpp:1722
        /** If data has run out, we will send the profiling data and total values to
          * the last zero block to be able to use
          * this information in the suffix output of stream.
          * If the request was interrupted, then `sendTotals` and other methods could not be called,
          *  because we have not read all the data yet,
          *  and there could be ongoing calculations in other threads at the same time.
          */

The empty sendData(state, {}) is the zero block the comment refers to: it carries the “suffix” information. And a final progress packet is sent after it, so the client’s progress bar lands on 100%.

Note the ordering asymmetry: Totals/Extremes come before the empty data block, and the final Progress comes after it. A client that only reads Data packets still gets a clean end-of-stream.

Cancellation

The cancel path is deliberately split in two, because “stop now” and “stop and give me what you have” are different requests.

cpp 
// src/Server/TCPHandler.cpp:3155
void TCPHandler::processCancel(QueryState & state)
{
    if (state.allow_partial_result_on_first_cancel && !state.stop_read_return_partial_result)
    {
        state.stop_read_return_partial_result = true;
        LOG_INFO(log, "Received 'Cancel' packet from the client, returning partial result.");
            return;
    }

    state.read_all_data = true;
    state.stop_query = true;

    throw Exception(ErrorCodes::QUERY_WAS_CANCELLED_BY_CLIENT, "Received 'Cancel' packet from the client, canceling the query.");
}

allow_partial_result_on_first_cancel is the partial_result_on_first_cancel setting. When it is on, the first Cancel is downgraded to “stop reading more data but let the pipeline finish and send what you have”; the second one is fatal.

sql 
-- Ctrl-C in clickhouse-client sends a native `Cancel` packet.
-- With this setting, the first Ctrl-C returns what has been computed so far
-- instead of throwing the query away; a second Ctrl-C kills it for good.
SELECT count()
FROM numbers(10000000000)
SETTINGS partial_result_on_first_cancel = 1;

The SQL-level equivalent, from a different connection, is KILL QUERY:

sql 
-- session A
SELECT sum(number) FROM numbers_mt(100000000000);

-- session B, while A is still running
KILL QUERY WHERE query_id = '2f0a...';

Note that KILL QUERY and the native Cancel packet are different paths: the packet is handled by the connection thread inside the half-duplex loop described above, whereas KILL QUERY goes through the process list and cancels the query from the outside. The process-list path is what max_execution_time and the KILL QUERY statement use.

The flag that other threads check is stop_query:

cpp 
// src/Server/TCPHandler.cpp:3149
void TCPHandler::checkIfQueryCanceled(QueryState & state)
{
    if (state.stop_query)
        throw Exception(ErrorCodes::QUERY_WAS_CANCELLED_BY_CLIENT, "Packet 'Cancel' has been received from the client, canceling the query.");
}

Every callback that might block on the socket checks it first — the ClusterFunctionReadTaskCallback, the MergeTreeAllRangesCallback, the external-tables initializer, the input-blocks reader. That is how a cancel arriving while a worker thread is blocked waiting for a distributed read task gets noticed promptly.

The rollback trick

One more detail worth stealing, because it is the kind of thing that only shows up after a year of production.

sendData writes a packet header, then a block. If the write fails half-way — the client vanished, the socket errored — the handler rewinds the write buffers to where they were before the attempt:

cpp 
// src/Server/TCPHandler.cpp:3266
    catch (...)
    {
        tryLogCurrentException(__PRETTY_FUNCTION__);

        /// In case of unsuccessful write, if the buffer with written data was not flushed,
        ///  we will rollback write to avoid breaking the protocol.
        /// (otherwise the client will not be able to receive exception after unfinished data
        ///  as it will expect the continuation of the data).
        /// It looks like hangs on client side or a message like "Data compressed with different methods".

        if (state.compression == Protocol::Compression::Enable)
        {
            auto extra_bytes_written_compressed = state.maybe_compressed_out->count() - prev_bytes_written_compressed_out;
            if (state.maybe_compressed_out->offset() >= extra_bytes_written_compressed)
                state.maybe_compressed_out->position() -= extra_bytes_written_compressed;
        }

        auto extra_bytes_written_out = out->count() - prev_bytes_written_out;
        if (out->offset() >= extra_bytes_written_out)
            out->position() -= extra_bytes_written_out;

        throw;
    }

The point: a half-written Data packet makes the client desynchronise permanently — it will treat the following bytes as block data. If the server then tries to send an Exception packet, the client either hangs or reports “Data compressed with different methods”. Rewinding the buffer by the number of bytes written but not yet flushed restores the stream to a packet boundary, so the Exception can be delivered cleanly.

What this design buys, and what it costs

Buys:

  • Cancellation with a bounded latency. Ten polls a second, one poll(0) each — the cost of a cancel check is one syscall per 100 ms per connection, amortised to nothing.
  • A live progress bar with no extra protocol traffic. Progress rides along with data packets and with the periodic sendInteractiveUpdates tick.
  • Server logs visible in the client terminal, interleaved with results, because the log queue is drained into the same output stream.
  • No separate thread per connection state machine. The connection thread is the only writer to the socket; everything else goes through callback_mutex.

Costs:

  • The 100 ms granularity. A cancel is not instant. On a query that finishes in 5 ms, a cancel arriving at t=1 ms may not be seen at all — which is fine, because there is nothing to cancel.
  • One thread per connection, mostly asleep. The connection thread wakes ten times a second per active query. At ten thousand concurrent queries that is 100 000 wakeups a second, and it is why the pipeline runs on a separate thread (PullingAsyncPipelineExecutor uses a ThreadFromGlobalPool) rather than on the connection thread.
  • The half-duplex constraint. The client cannot send a new command while the server is streaming results for the previous one — the protocol has no pipelining. The Query packet is rejected with throwUnexpectedPacket while data is expected.
  • A mutex on the hot path. Every sendData and every poll on the input happens under callback_mutex, so a distributed query with many read-task round-trips serialises the connection thread against its own workers.

The design is a deliberate trade: ClickHouse gave up command pipelining and per-connection responsiveness in exchange for the ability to interrupt a running query, stream progress, and surface internal logs, all over a single socket, without a second protocol.

ClickhouseInternalsNetworkingProtocolConcurrency