ClickHouse's Query Pipeline: Ports, Status Codes and the Executing Graph

TL;DR: ClickHouse doesn’t call operators in a fixed order. Every processor reports a status — “I need data”, “my output is full”, “call my work()” — and the executor walks the graph, re-preparing only the neighbours that actually changed.

Every query engine eventually has to answer the same question: given a graph of operators — read these files, filter these rows, group them, sort them, write them out — in what order do you actually run them?

The textbook answer is the Volcano iterator model. Every operator exposes a next() that returns one tuple at a time, and the executor is a nest of while (child.next()) loops. It’s beautifully simple, and it’s also unbelievably slow: every tuple crosses every operator through a virtual call, and nobody ever gets to see a whole batch.

ClickHouse’s answer is the opposite extreme. Operators exchange blocks — tens of thousands of rows at a time — through ports, and the executor never calls operators in a fixed order. Instead every processor is asked, in effect, “given the state of your neighbours, what should happen next?”, and it answers with one of five status codes.

The executor is a scheduler that walks the graph, and the graph tells it where to go.

This is a tour of that machinery: the IProcessor interface, the one-slot mailbox that is a Port, the five status codes, the version counters that let a processor notice that a neighbour changed, and the ExecutingGraph that ties them together.

The building block: `IProcessor`

A processor is a node with zero or more input ports and zero or more output ports. The taxonomy in the header comment of src/Processors/IProcessor.h still describes every processor in the tree — sources, sinks, simple transformations, squashing transformations (DISTINCT, WHERE), accumulating transformations (ORDER BY, GROUP BY), and limiting transformations.

The crucial design decision is that there is no next(). A processor never asks its input for data; it declares whether it needs data, and the executor arranges for it to arrive. And a processor never decides what runs next — it returns a status, and the executor interprets it.

Here’s the entire status vocabulary:

cpp
// src/Processors/IProcessor.h:150
    enum class Status : uint8_t
    {
        /// Processor needs some data at its inputs to proceed.
        /// You need to run another processor to generate required input and then call 'prepare' again.
        NeedData,

        /// Processor cannot proceed because output port is full or not isNeeded().
        /// You need to transfer data from output port to the input port of another processor and then call 'prepare' again.
        PortFull,

        /// All work is done (all data is processed or all output are closed), nothing more to do.
        Finished,

        /// You may call 'work' method and processor will do some work synchronously.
        Ready,

        /// You may call 'schedule' method and processor will return a descriptor.
        /// You need to poll this descriptor and call work() afterwards.
        Async,

        /// Processor wants to add new processors and/or remove finished neighbours.
        /// Update must be obtained by updatePipeline() call.
        UpdatePipeline,
    };

NeedData and PortFull are the two “I’m blocked” answers, and they’re the whole point. They tell the executor which neighbour has to run next: NeedData means “run my upstream”, PortFull means “run my downstream”. Ready means “I have work, call work() on me”. Async means “I have a file descriptor, poll it”. UpdatePipeline means “I changed the graph”.

The contract for prepare() is spelled out carefully, because it’s where most of the subtlety lives:

cpp
// src/Processors/IProcessor.h:177
    /** Method 'prepare' is responsible for all cheap ("instantaneous": O(1) of data volume, no wait) calculations.
      *
      * It may access input and output ports,
      *  indicate the need for work by another processor by returning NeedData or PortFull,
      *  or indicate that processing has finished by returning `Finished`,
      *  it may pull data from input ports and push data to output ports.
      *
      * The method is not thread-safe and must be called from a single thread in one moment of time,
      *  even for different connected processors.
      *
      * Instead of all long work (CPU calculations or waiting) it should just prepare all required data and return Ready or Async.
      *
      * Thread safety and parallel execution:
      * - no methods (prepare, work, schedule) of single object can be executed in parallel;
      * - method 'work' can be executed in parallel for different objects, even for connected processors;
      * - method 'prepare' cannot be executed in parallel even for different objects,
      *   if they are connected (including indirectly) to each other by their ports;
      */
    virtual Status prepare();

Read that last bullet twice. It’s the fundamental law of the model: work() may run in parallel with anything; prepare() may not run in parallel with anything it’s connected to.

The port connections define the happens-before edges of the whole execution, and the executor honours them by construction.

The port: a mailbox with the flags in the pointer

A port is where two processors meet. It’s shared state, and it must be readable and writable from two different threads (the producer and the consumer), so it’s an atomic pointer to a Data slot — plus three flag bits that say what the slot currently holds.

cpp
// src/Processors/Port.h:75
        /// Flags for Port state.
        /// Will store them in least pointer bits.

        /// Port was set finished or closed.
        static constexpr std::uintptr_t IS_FINISHED = 1;
        /// Block is not needed right now, but may be will be needed later.
        /// This allows to pause calculations if we are not sure that we need more data.
        static constexpr std::uintptr_t IS_NEEDED = 2;
        /// Check if port has data.
        static constexpr std::uintptr_t HAS_DATA = 4;

        static constexpr std::uintptr_t FLAGS_MASK = IS_FINISHED | IS_NEEDED | HAS_DATA;
        static constexpr std::uintptr_t PTR_MASK = ~FLAGS_MASK;

Packing the flags into the low bits of the pointer itself is the trick. It means the entire port state — the slot pointer plus the three flags — is a single atomic word. So a push and a “set not needed” can be one CAS operation, and there’s never a window in which the flags disagree with the pointer.

The little DataPtr class exists to make that safe. The swap() method is the whole implementation of a push or a pull:

cpp
// src/Processors/Port.h:117
            uintptr_t ALWAYS_INLINE swap(std::atomic<Data *> & value, std::uintptr_t flags, std::uintptr_t mask) /// NOLINT
            {
                Data * expected = nullptr;
                Data * desired = getPtr(flags | getUInt(data));

                while (!value.compare_exchange_weak(expected, desired))
                    desired = getPtr((getUInt(expected) & FLAGS_MASK & (~mask)) | flags | getUInt(data));

                /// It's not very safe. In case of exception after exchange and before assignment we will get leak.
                /// Don't know how to make it better.
                data = getPtr(getUInt(expected) & PTR_MASK);

                return getUInt(expected) & FLAGS_MASK;
            }

The mask argument says which flags the caller intends to change and which ones it wants to preserve from whatever is currently in the word. The CAS loop re-computes desired from expected on every failure, so concurrent flag updates by the two sides are merged rather than lost.

push() and pull() are then thin wrappers:

cpp
// src/Processors/Port.h:154
        void ALWAYS_INLINE push(DataPtr & data_, std::uintptr_t & flags)
        {
            flags = data_.swap(data, HAS_DATA, HAS_DATA);

            /// It's possible to push data into finished port. Will just ignore it.
            /// if (flags & IS_FINISHED)
            ///    throw Exception(ErrorCodes::LOGICAL_ERROR, "Cannot push block to finished port.");

            /// It's possible to push data into port which is not needed now.
            /// if ((flags & IS_NEEDED) == 0)
            ///    throw Exception(ErrorCodes::LOGICAL_ERROR, "Cannot push block to port which is not needed.");

            if (unlikely(flags & HAS_DATA))
                throw Exception(ErrorCodes::LOGICAL_ERROR, "Cannot push block to port which already has data");
        }

        void ALWAYS_INLINE pull(DataPtr & data_, std::uintptr_t & flags, bool set_not_needed = false)
        {
            uintptr_t mask = HAS_DATA;

            if (set_not_needed)
                mask |= IS_NEEDED;

            flags = data_.swap(data, 0, mask);

            /// It's ok to check because this flag can be changed only by pulling thread.
            if (unlikely((flags & IS_NEEDED) == 0) && !set_not_needed)
                throw Exception(ErrorCodes::LOGICAL_ERROR, "Cannot pull block from port which is not needed");

            if (unlikely((flags & HAS_DATA) == 0))
                throw Exception(ErrorCodes::LOGICAL_ERROR, "Cannot pull block from port which has no data");
        }

Note the semantics of IS_NEEDED. It’s not “the consumer wants data”; it’s “the consumer might want data later”.

A producer may push into a port that’s currently IS_NEEDED == 0 — the data simply sits in the slot — but it may not pull from a port that’s not needed, because pulling implies “give me the next thing” and the consumer has explicitly said it doesn’t want more right now.

canPush() is the producer’s gate:

cpp
// src/Processors/Port.h:478
    bool ALWAYS_INLINE canPush() const
    {
        assumeConnected();
        auto flags = state->getFlags();
        return (flags & State::IS_NEEDED) && ((flags & State::HAS_DATA) == 0);
    }

Both conditions matter. If the consumer said “not needed”, the producer must wait; and if the slot already has data, the producer must wait, because the port holds exactly one chunk.

This “not needed” handshake is what makes the pull side of the model honest. The consumer says input.setNotNeeded() while it’s busy, the producer’s canPush() goes false, and the producer stops.

That’s how a LIMIT transform can stop a system.numbers_mt source from producing infinity. It closes the port.

sql
-- The Limit transform closes its input port once the limit is reached,
-- so the source stops generating rows instead of the query reading forever.
SELECT * FROM system.numbers_mt WHERE number = 1000000 LIMIT 1;

The shape of the resulting graph is visible with EXPLAIN PIPELINE, which prints the processors and the ports the executor will walk:

sql
EXPLAIN PIPELINE
SELECT number FROM numbers(10) WHERE number > 5 LIMIT 2;

`prepare()` and `work()`

The canonical implementation of the two-phase protocol is ISimpleTransform, the base class of nearly every operator in the tree (filters, expressions, DISTINCT, window functions, LIMIT BY…).

cpp
// src/Processors/ISimpleTransform.cpp:23
ISimpleTransform::Status ISimpleTransform::prepare()
{
    /// Check can output.

    if (output.isFinished())
    {
        input.close();
        onFinish();
        return Status::Finished;
    }

    if (!output.canPush())
    {
        input.setNotNeeded();
        return Status::PortFull;
    }

    /// Output if has data.
    if (has_output)
    {
        output.pushData(std::move(output_data));
        has_output = false;

        if (!no_more_data_needed)
            return Status::PortFull;

    }

    /// Stop if don't need more data.
    if (no_more_data_needed)
    {
        input.close();
        output.finish();
        return Status::Finished;
    }

    /// Check can input.
    if (!has_input)
    {
        if (input.isFinished())
        {
            output.finish();
            onFinish();
            return Status::Finished;
        }

        input.setNeeded();

        if (!input.hasData())
            return Status::NeedData;

        input_data = input.pullData(set_input_not_needed_after_read);
        has_input = true;

        if (input_data.exception)
            /// No more data needed. Exception will be thrown (or swallowed) later.
            input.setNotNeeded();
    }

    /// Now transform.
    return Status::Ready;
}

Every branch here is a decision about who runs next:

text
  output.isFinished()      -> input.close();  output.finish()  -> Finished
  !output.canPush()        -> input.setNotNeeded()              -> PortFull
  has_output               -> output.pushData()                 -> PortFull
  no_more_data_needed      -> input.close();   output.finish()  -> Finished
  input.isFinished()       -> output.finish()                   -> Finished
  !input.hasData()         -> input.setNeeded()                 -> NeedData
  otherwise                -> pullData();  transform()          -> Ready

Notice that prepare() does push: output.pushData(...) and input.pullData(...) both happen inside prepare(). The division of labour isn’t “prepare moves no data”, it’s “prepare does no expensive work”.

Pulling a chunk is O(1) — it’s a pointer swap. Actually computing on it is not, and that’s what work() is for:

cpp
// src/Processors/ISimpleTransform.cpp:86
void ISimpleTransform::work()
{
    if (input_data.exception)
    {
        transform(input_data.exception);

        output_data = std::move(input_data);
        has_input = false;
        has_output = true;
        return;
    }

    try
    {
        transform(input_data.chunk, output_data.chunk);
    }
    catch (DB::Exception &)
    {
        output_data.exception = std::current_exception();
        has_output = true;
        has_input = false;
        return;
    }

    has_input = !needInputData();

    if (!skip_empty_chunks || output_data.chunk)
        has_output = true;

    if (has_output && !output_data.chunk && !getOutputPort().getHeader().empty())
        /// Support invariant that chunks must have the same number of columns as header.
        output_data.chunk = Chunk(getOutputPort().getHeader().cloneEmpty().getColumns(), 0);
}

Three things are worth noting.

Exceptions are data. A failure in the upstream processor is pushed into the port as an exception_ptr alongside the chunk, not thrown across the thread boundary. work() picks it up, hands it to transform(), and pushes it on. The exception then travels downstream with the stream, which means the error surfaces in the right place and every processor upstream of the failure gets to clean up.

work() cannot block. That’s why the Status::Async case exists: a processor that needs to wait for a socket or a file descriptor returns Async from prepare() and hands out an fd from schedule().

The empty-chunk invariant. Every chunk that crosses a port must have exactly as many columns as the port’s header, even if it has zero rows. ISimpleTransform enforces it at the bottom of work() by synthesising an empty chunk with the right shape.

Version counters: how a processor learns that a neighbour changed

The executor must know which processors to wake up. Polling every processor in the graph after every chunk would be O(V) per chunk, which for a twenty-node graph and a million chunks is a lot of pointless work.

ClickHouse’s answer is a version counter per connection, with a twist: the counter lives in the UpdateInfo of the port, and it’s compared against the value the neighbour last saw.

cpp
// src/Processors/Port.h:34
    struct UpdateInfo
    {
        using UpdateList = VectorWithMemoryTracking<void *>;

        UpdateList * update_list = nullptr;
        void * id = nullptr;
        UInt64 version = 0;
        UInt64 prev_version = 0;

        void inline ALWAYS_INLINE update()
        {
            if (version == prev_version && update_list)
                update_list->push_back(id);

            ++version;
        }

        void inline ALWAYS_INLINE trigger() { prev_version = version; }
    };

update() is called whenever the port’s state changes (a push, a pull, a setNeeded). It bumps version, and — this is the interesting part — if version == prev_version, meaning nobody has consumed the change yet, it appends the port’s identity to a list of “ports that changed”. trigger() marks the change as seen by setting prev_version = version.

So the protocol is: a port that changes while nobody was looking records itself on a list; the executor drains that list and re-prepares exactly those neighbours. If a port changes twice before the neighbour is re-prepared, the neighbour is only told once. It doesn’t need to know how many times it changed, only that it did.

This is the mechanism that replaced the old “wake up the neighbour immediately” design, and it’s why prepare() can be called on a subset of the graph instead of on all of it.

The `ExecutingGraph`: nodes, edges, and who owns what

The executor’s view of the pipeline is ExecutingGraph, in src/Processors/Executors/Runtime/V1/. It holds one Node per processor and two kinds of Edge per connection: a direct edge from an output port to the consuming processor’s input port, and a backward edge from an input port back to the producing processor’s output port.

cpp
// src/Processors/Executors/Runtime/V1/ExecutingGraph.h:31
    /// Edge represents connection between OutputPort and InputPort.
    /// For every connection, two edges are created: direct and backward (it is specified by backward flag).
    struct Edge
    {
        Edge(Node * to_, bool backward_,
             InputPort * input_port_, OutputPort * output_port_,
             VectorWithMemoryTracking<void *> * update_list)
            : to(to_), backward(backward_)
            , input_port(input_port_), output_port(output_port_)
        {
            update_info.update_list = update_list;
            update_info.id = this;
        }

        /// Node this edge points to.
        /// It is the node with output_port for direct edge or the node with input_port for backward edge.
        Node * to = nullptr;
        bool backward;
        /// Ports of this connection. Both pointers are set regardless of direction:
        /// for a direct edge (out -> in) input_port belongs to `to`; for a back edge the reverse.
        InputPort * input_port = nullptr;
        OutputPort * output_port = nullptr;

        /// Edge version is increased when port's state is changed (e.g. when data is pushed). See Port.h for details.
        /// To compare version with prev_version we can decide if neighbour processor need to be prepared.
        Port::UpdateInfo update_info;
    };

The Node carries a status with a very specific ownership protocol:

cpp
// src/Processors/Executors/Runtime/V1/ExecutingGraph.h:62
    /// Status for processor.
    /// Can be owning or not. Owning means that executor who set this status can change node's data and nobody else can.
    enum class ExecStatus : uint8_t
    {
        Idle,  /// prepare returned NeedData or PortFull. Non-owning.
        Preparing,  /// some executor is preparing processor, or processor is in task_queue. Owning.
        Executing,  /// prepare returned Ready or Async and task is executing. Owning.
        Finished,  /// prepare returned Finished. Non-owning.
    };

“Owning” is the key word. Two threads may never call prepare() on the same processor, and may never call prepare() on connected processors at the same time. The status is how the graph enforces that: a node in Preparing or Executing is owned by exactly one thread, and the per-node std::mutex status_mutex is taken when ownership is transferred.

The traversal itself is updateNode(). It starts from one processor and spreads along the edges, re-preparing whoever was affected:

cpp
// src/Processors/Executors/Runtime/V1/ExecutingGraph.cpp:524
ExecutingGraph::UpdateNodeStatus ExecutingGraph::updateNode(IProcessor & initial, Queue & queue, Queue & async_queue, UpdateNodeBuffers & buffers)
{
    Processors delayed_destruction;
    auto & updated_edges = buffers.updated_edges;
    auto & updated_processors = buffers.updated_processors;
    auto & pending_expansion = buffers.pending_expansion;

    /// The buffers may be left non-empty by a previous call that exited early or threw.
    buffers.reset();

    std::shared_lock read_lock(nodes_mutex);

    /// Traversal starts from the node related to updated processor
    updated_processors.push_back(processors_map.at(&initial));

    while (!updated_processors.empty() || !updated_edges.empty())
    {
        std::optional<std::unique_lock<std::mutex>> stack_top_lock;

        if (updated_processors.empty())
        {
            auto * edge = updated_edges.front();
            updated_edges.pop_front();

            /// Here we have ownership on edge, but node can be concurrently accessed.

            auto & node = *edge->to;

            std::unique_lock lock(node.status_mutex);

            ExecutingGraph::ExecStatus status = node.status;

            if (status != ExecutingGraph::ExecStatus::Finished)
            {
                if (edge->backward)
                    node.updated_output_ports.push_back(edge->output_port);
                else
                    node.updated_input_ports.push_back(edge->input_port);

                if (status == ExecutingGraph::ExecStatus::Idle)
                {
                    node.status = ExecutingGraph::ExecStatus::Preparing;
                    updated_processors.push_front(edge->to);
                    stack_top_lock = std::move(lock);
                }
                else
                    edge->to->processor()->onUpdatePorts();
            }
        }

Two work lists are at work here, and the distinction is worth spelling out. updated_processors is a queue of nodes to call prepare() on. updated_edges is a queue of connections whose port state changed and whose far side therefore needs to know.

The loop interleaves them. An edge whose far side is Idle promotes that node to Preparing and puts it on the processor queue. If the far side is already Preparing or Executing, the edge is simply recorded in that node’s updated_input_ports / updated_output_ports list, and the node will pick it up on its next prepare().

That’s the “which processor am I at” trick in its modern form: there is no cursor. There’s no “current processor” that remembers where it was. Instead, each processor holds the set of ports that changed since it was last prepared, and the graph walks edges until it reaches a fixed point.

The core of the traversal is the prepare() call and the dispatch on its result:

cpp
// src/Processors/Executors/Runtime/V1/ExecutingGraph.cpp:621
                switch (*node.last_processor_status)
                {
                    case IProcessor::Status::NeedData:
                    case IProcessor::Status::PortFull:
                    {
                        node.status = ExecutingGraph::ExecStatus::Idle;
                        if (spill_state.memory_reservation)
                        {
                            /// A request may have arrived while this node was owned by normal execution.
                            for (auto * spillable : processor.getSpillables())
                                if (spill_state.scheduleSpill(node, spillable, queue))
                                    break;
                        }
                        break;
                    }
                    case IProcessor::Status::Finished:
                    {
                        node.status = ExecutingGraph::ExecStatus::Finished;
                        if (last_status != IProcessor::Status::Finished)
                            spill_state.unregisterSpillables(node);
                        accountFinishedProcessorInGroup(*node.processor_iter);
                        break;
                    }
                    case IProcessor::Status::Ready:
                    {
                        node.status = ExecutingGraph::ExecStatus::Executing;
                        queue.push(node.processor());
                        break;
                    }
                    case IProcessor::Status::Async:
                    {
                        node.status = ExecutingGraph::ExecStatus::Executing;
                        async_queue.push(node.processor());
                        break;
                    }
                    case IProcessor::Status::UpdatePipeline:
                    {
                        need_update_pipeline = true;
                        pending_expansion.push_back(current);
                        break;
                    }
                }

Ready puts the processor on the work queue, to be picked up by a worker thread. NeedData/PortFull put the node back to Idle, meaning “somebody else must move first”. Finished retires the node for good. And then the edges that this processor disturbed get re-armed:

cpp
// src/Processors/Executors/Runtime/V1/ExecutingGraph.cpp:666
                    /// If you wonder why edges are pushed in reverse order,
                    /// it is because updated_edges is a stack, and we prefer to get from stack
                    /// input ports firstly, and then outputs, both in-order.
                    ///
                    /// Actually, there should be no difference in which order we process edges.
                    /// However, some tests are sensitive to it (e.g. something like SELECT 1 UNION ALL 2).
                    /// Let's not break this behaviour so far.

                    for (auto it = node.post_updated_output_ports.rbegin(); it != node.post_updated_output_ports.rend(); ++it)
                    {
                        auto * edge = static_cast<ExecutingGraph::Edge *>(*it);
                        updated_edges.push_front(edge);
                        edge->update_info.trigger();
                    }

edge->update_info.trigger() is the prev_version = version from Port.h, and it’s called after the edge is queued. The ordering matters: trigger() says “the neighbour has now seen this change”, and the queue entry is what makes it true.

The graph can change shape at runtime

The last status, UpdatePipeline, is unusual: a processor can, while running, decide that it needs more processors.

The classic case is external sort. MergeSortingTransform starts as a single-input, single-output processor, and when it decides to spill to disk it must insert a BufferingToFileSink and a BufferingFromFileSource into the middle of its own pipeline.

The mechanism is updatePipeline(), which returns a PipelineUpdate:

cpp
// src/Processors/IProcessor.h:262
    struct PipelineUpdate
    {
        Processors to_add;
        Processors to_remove;
    };
    virtual PipelineUpdate updatePipeline();

The graph handles the insertion by rebuilding the edges of every node and scheduling the affected ones for re-preparation. The interesting part is the error handling, which encodes a hard-won lesson about allocation failures:

cpp
// src/Processors/Executors/Runtime/V1/ExecutingGraph.cpp:324
    try
    {
        return updatePipelineImpl(stack, cur_node, update);
    }
    catch (...)
    {
        expansion_failed = true;
        /// The ports of the pipeline's processors point at the new processors, and the pipeline is read after the graph is gone.
        std::lock_guard guard(processors_mutex);
        for (auto it = update.to_add.begin(); it != update.to_add.end();)
        {
            auto next = std::next(it);
            if (std::ranges::find(*processors, *it) == processors->end())
                processors->splice(processors->end(), update.to_add, it);
            it = next;
        }
        throw;
    }

The processor has already connected its ports to the new processors before updatePipeline() is even called. So if the graph then fails to record them — say, addNode throws MEMORY_LIMIT_EXCEEDED — the ports point at processors the graph doesn’t know about, and the graph is permanently inconsistent.

The handler keeps the new processors alive (so the ports stay valid) and sets expansion_failed, which makes every later expansion bail out instead of walking into the inconsistency.

Trade-offs

What this buys you.

The model is fully data-parallel by construction. Any node of the graph may be scheduled on any thread, because the only shared state between processors is a port, and a port is a single atomic word. There’s no notion of “this operator belongs to this thread”, which is what lets ClickHouse run a SELECT over a MergeTree with 16 streams on 16 cores without any of the 16 streams knowing about the other 15.

The graph also reshapes itself. External sort, GLOBAL JOIN, subquery materialisation, and LIMIT pushdown all insert or remove processors while the query is running, and none of them require a separate “compile the plan” phase.

Finally, the split between prepare() (cheap, may not run concurrently with connected peers) and work() (expensive, may run concurrently with everything) is precisely the split that lets the executor overlap computation without any locking on the hot path.

What it costs.

prepare() cannot run in parallel with connected processors. For a wide graph — say a UNION ALL of twenty subqueries, each with a dozen operators — this means the scheduling phase is effectively serialised per connection. That’s why the pipeline shape matters so much: every extra operator in a chain adds a serialisation point between the parallel parts.

It’s also why Pipe::resize() has a comment about splitting the resize processor when multiple threads are allocated:

cpp
// src/QueryPipeline/Pipe.cpp:781
    /// Performance bottleneck identified: Severe lock contention for ExecutingGraph::Node::status_mutex.
    /// Issues observed:
    /// 1. Increased latency of ExecutingGraph::updateNode, lengthening overall query latency.
    /// 2. Unnecessary CPU cycle consumption in native_queued_spin_lock_slowpath (kernel).
    /// 3. Decreased CPU utilization.
    ///
    /// Proposed solution: Split ResizeProcessor when multiple threads are allocated to execute a query.

The second cost is the per-node status_mutex. Every edge traversal takes it. On a hot path with many small chunks this is measurable, and the comment above is the record of someone measuring it.

The third cost is that the model is hard to reason about in the presence of prepare() side effects. ISimpleTransform::prepare() pushes data. LimitTransform::prepare() closes ports. SortingTransform::prepare() can return UpdatePipeline and rebuild the graph. There’s no single “execute this operator” entry point — which is exactly why ClickHouse’s test infrastructure is built around assembling a whole Pipe.

Wrapping up

The real win: ClickHouse’s pipeline doesn’t have a “current operator”. It has a set of statuses and a graph that knows how to resolve them. That single design choice buys data parallelism for free, because the only thing two processors share is an atomic word.

Two things to internalize:

If you’re debugging a stuck ClickHouse query, look at the graph, not the SQL. EXPLAIN PIPELINE shows you the processors and the ports; the status codes tell you who’s blocked and on whom.
The model is deliberately push-driven at the ports. IS_NEEDED and HAS_DATA are the only two things that matter, and every producer must respect them or you get “Cannot push block to port which already has data”.

So next time you add a custom operator, remember: your prepare() must be O(1), your work() must not block, and your exceptions must travel as data — not as thrown errors.

The building block: IProcessor