ClickHouse Runs a Query Pipeline on N Threads: Task Queues, Work Stealing and a Pool That Grows on Demand

TL;DR: ClickHouse doesn’t hand you max_threads threads. It starts with one, measures how much parallel work the pipeline actually creates, and grows the pool only when idle threads genuinely can’t absorb it — then shrinks again if the server preempts you.

A query plan in ClickHouse is not one pipeline. It’s a stack of pipelines: read from storage, filter, project, aggregate, sort, format. Each stage produces many parallel streams, and consecutive stages are connected by ResizeProcessors that shuffle chunks between them.

Running that means answering three questions at once:

Which processor should run next? (the graph — see the previous article on ExecutingGraph)
Which thread should run it?
How many threads should exist at all?

The third question is the interesting one. A query that reads one small table does not want sixteen threads. A query that scans a hundred partitions does. And a query that has been running for thirty seconds may need fewer threads than it started with, because the server is under load from other queries.

ClickHouse’s answer, in src/Processors/Executors/Runtime/V1/, is a system that starts with one thread, measures demand as the pipeline runs, and grows the pool only when there is genuinely parallel work that idle threads cannot absorb.

Two kinds of parallelism

Before the machinery, let’s be precise about what “parallel” means here, because ClickHouse does two different things and they get conflated a lot.

Data parallelism is the obvious one: SELECT count() FROM t reads from N partitions, each partition is its own stream, and N streams are processed at once. This is a property of the plan — the number of streams is fixed when the plan is built.

Pipeline parallelism is the subtler one. Even in a pipeline with a single stream, the stage reading from disk isn’t blocked while the stage formatting the output is busy. A chunk is handed from one processor to the next, and while the consumer works on chunk k, the producer is already building chunk k+1.

The executor supports both, and the interesting bit is how it decides how many threads to use:

sql
-- Sixteen streams of data parallelism; the number of threads that actually
-- run them is decided at execution time, not by the plan.
SELECT count() FROM numbers_mt;

sql
-- The ceiling. The pool starts at one thread and grows only if there is
-- parallel work that idle threads cannot absorb.
SETTINGS max_threads = 16;
SELECT count() FROM numbers_mt;

Note that the two are not the same thing. A pipeline with 16 streams needs 16 threads to be fully parallel, but it will still make progress — just less of it — on one thread.

The shape of the executor

PipelineExecutor owns three things:

cpp
// src/Processors/Executors/Runtime/V1/PipelineExecutor.h:49
private:
    /// Outlives graph registrations, including when construction fails.
    QueryStatusPtr process_list_element;
    ExecutingGraphPtr graph;

    ExecutorTasks tasks;

    /// Concurrency control related
    SlotAllocationPtr cpu_slots;
    AcquiredSlotPtr single_thread_cpu_slot; // cpu slot for single-thread mode to work using executeUntil()
    std::unique_ptr<ThreadPool> pool;
    std::mutex spawn_mutex;

graph is the processor graph (the subject of the previous article). tasks is the scheduler: the queues, the per-thread contexts, the counters. cpu_slots and pool are the resources: how many threads are allowed to exist, and the threads themselves.

The entry point is execute(), which is unremarkable apart from its comment about output formats:

cpp
// src/Processors/Executors/Runtime/V1/PipelineExecutor.cpp:231
void PipelineExecutor::execute(size_t num_threads, bool concurrency_control)
{
    /// `QueryStatus` may release its reservation before this executor is destroyed.
    SCOPE_EXIT({ detachSpillables(); });
    if (process_list_element && !process_list_element->checkTimeLimit())
        cancel(IProcessor::CancelReason::CancelledByTimeout);

    num_threads = std::max<size_t>(num_threads, 1);

and executeImpl() is where the threads actually appear:

cpp
// src/Processors/Executors/Runtime/V1/PipelineExecutor.cpp:783
void PipelineExecutor::executeImpl(size_t num_threads, bool concurrency_control)
{
    initializeExecution(num_threads, concurrency_control);

    try
    {
        if (num_threads > 1)
        {
            {
                std::lock_guard lock(spawn_mutex);
                // Start at least one thread, could block to acquire the first CPU slot
                spawnThreads(cpu_slots->acquire());
            }
            tasks.processAsyncTasks();
            pool->wait();
        }
        else
        {
            auto slot = cpu_slots->acquire();
            tasks.upscale(slot->slot_id);
            executeSingleThread(slot->slot_id, WorkloadResources(slot.get(), process_list_element));
        }
    }

Notice pool->wait(). The master thread does not execute pipeline work in the multi-threaded case. It spawns worker threads and then parks in processAsyncTasks(), which is the epoll loop for processors that returned Status::Async.

The master thread is a scheduler, not a worker — and that distinction matters for deadlock avoidance, which is why the master is never downscaled:

cpp
// src/Processors/Executors/Runtime/V1/PipelineExecutor.cpp:591
    // The first thread is called master thread.
    // It is NOT the thread that handles async tasks (unless query has max_threads=1).
    // Master thread is different from other threads due to special role in scheduling:
    //  1. During query start, master thread spawns worker threads, so starting it fast is important.
    //  2. It should never be downscaled to avoid query deadlock (0 threads to process tasks), although it can be preempted.
    //  3. When using ConcurrencyControl, master thread is granted immediately and may not request CPU slot (see 'fair_round_robin').
    //  4. When using resource scheduler, master thread may use different resource link than worker threads.
    const auto master_threads = 1uz;

Task queues with work stealing

The scheduler’s central data structure is TaskQueue, and it’s per-thread-first:

cpp
// src/Processors/Executors/Runtime/V1/TasksQueue.h:17
/// A set of task queues for multithreaded processing.
/// Every threads has its dedicated queue and uses it preferably.
/// When there are no tasks left in dedicated queue it steals tasks from other threads
template <typename Task>
class TaskQueue
{
public:
    void init(size_t num_threads) { queues.resize(num_threads); }

    /// Push a task into thread's dedicated queue
    void push(Task * task, size_t thread_num)
    {
        queues[thread_num].push(task);
        use_queues = std::max(use_queues, thread_num + 1);
        ++num_tasks;
    }

    /// Returns thread number to pop task from.
    /// First it check dedicated queue, and only if it is empty, it steal from other threads
    size_t getAnyThreadWithTasks(size_t from_thread)
    {
        if (num_tasks == 0)
            throw Exception(ErrorCodes::LOGICAL_ERROR, "TaskQueue is empty");

        if (from_thread >= use_queues)
            from_thread = 0;

        for (size_t i = 0, max_iterations = use_queues; i < max_iterations; ++i)
        {
            if (!queues[from_thread].empty())
                return from_thread;

            ++from_thread;
            if (from_thread == use_queues)
            {
                use_queues--; // lazy update: less expensive than updating on every pop
                from_thread = 0;
            }
        }

        throw Exception(ErrorCodes::LOGICAL_ERROR, "TaskQueue is empty");
    }

Each thread has a private FIFO. A thread with work prefers its own queue; only when it’s empty does it walk the other threads’ queues looking for something to steal.

use_queues is a lazily-maintained upper bound on how many queues are non-empty, so the scan is O(number of active threads) and not O(total threads). That “lazy update” comment is doing real work — updating the bound on every pop would cost more than it saves.

The per-thread affinity matters for cache reasons: a task that runs on thread 3 tends to produce data that thread 3 will consume next, and the data stays in that core’s L1/L2.

There are three queues, not one:

cpp
// src/Processors/Executors/Runtime/V1/ExecutorTasks.h:43
    /// Queue with pointers to tasks. Each thread will concurrently read from it until finished flag is set.
    /// Stores processors need to be prepared. Preparing status is already set for them.
    TaskQueue<IProcessor> task_queue;

    /// Async tasks should be processed with higher priority, but also require task stealing logic.
    /// So we have a separate queue specifically for them.
    TaskQueue<IProcessor> fast_task_queue;
    std::atomic_bool has_fast_tasks = false; // Required only to enable local task optimization
    std::atomic_bool spill_requested = false;

    /// Queue which stores tasks where processors returned Async status after prepare.
    /// If multiple threads are used, main thread will wait for async tasks.
    /// For single thread, will wait for async tasks only when task_queue is empty.
    PollingQueue async_task_queue;

fast_task_queue holds processors that came back from an asynchronous operation — a socket that became readable, a file that finished being written. They get priority because they’re latency-sensitive: the data is already there and the client is waiting.

async_task_queue is different in kind. It’s not a task queue at all, it’s an epoll wrapper that owns the file descriptors.

The local-task optimization

Here’s the single most important performance trick in the executor, and it’s small.

When a processor finishes, the next processor to run is often known immediately — it’s the downstream consumer of the chunk that was just produced. Rather than pushing it onto a shared queue (where another thread would have to wake up, take the mutex, and pop it), the executor hands it directly to the thread that’s already running.

cpp
// src/Processors/Executors/Runtime/V1/ExecutorTasks.cpp:192
size_t ExecutorTasks::pushTasks(Queue & queue, Queue & async_queue, ExecutionThreadContext & context)
{
    /// Take local task from queue if has one.
    if (!queue.empty() && !has_fast_tasks.load(std::memory_order_relaxed)
        && context.num_scheduled_local_tasks < ExecutionThreadContext::max_scheduled_local_tasks)
    {
        ++context.num_scheduled_local_tasks;
        context.setTask(queue.front());
        queue.pop();
    }
    else
    {
        context.num_scheduled_local_tasks = 0;
        context.setTask(nullptr);
    }

and the constant that bounds it:

cpp
// src/Processors/Executors/Runtime/V1/ExecutionThreadContext.h:55
    /// There is a performance optimization that schedules a task to the current thread, avoiding global task queue.
    /// Optimization decreases contention on global task queue but may cause starvation.
    /// See 01104_distributed_numbers_test.sql
    /// This constant tells us that we should skip the optimization
    /// if it was applied more than `max_scheduled_local_tasks` in a row.
    constexpr static size_t max_scheduled_local_tasks = 128;

128 in a row. The comment names the failure mode: starvation. If one thread keeps grabbing work locally while other threads sit idle with nothing to do, the pipeline runs on one core while the other max_threads - 1 cores idle.

The counter resets as soon as the local optimization is skipped, so a thread that has been running alone for 128 tasks will put the next one on the global queue and let somebody else take it.

The referenced test, 01104_distributed_numbers_test.sql, is a good illustration of why this matters: SELECT * FROM numbers(1000000000) with distributed_product_mode and a LIMIT — the limit transform stops the source, and if the source isn’t rebalanced the other threads never wake up.

The has_fast_tasks check in the condition is the second half of the guard: if there’s an asynchronous task waiting, it takes priority over a local task, because asynchronous tasks are latency-critical.

Measuring demand: `pushTasks` returns a number

Now the interesting part. pushTasks doesn’t just enqueue work — it measures how much parallel work was created, and returns it:

cpp
// src/Processors/Executors/Runtime/V1/ExecutorTasks.cpp:208
    if (!queue.empty() || !async_queue.empty())
    {
        std::unique_lock lock(mutex);

        /// Measure the parallelism of the push: how many threads could run the new tasks
        /// concurrently. Used by `PipelineExecutor` to size the slot-allocation ceiling via
        /// `ISlotAllocation::setMax` on the pipeline's `cpu_slots` handle.
        const size_t pushed = queue.size() + async_queue.size();

and then, at the end:

cpp
// src/Processors/Executors/Runtime/V1/ExecutorTasks.cpp:232
        /// Capture state under the lock — tryWakeUp may unlock it internally to wake a
        /// thread, after which `total_slots` could change.
        const size_t idle_before = threads_queue.size();
        const size_t total_slots_now = total_slots;
        const size_t num_threads_now = num_threads;  // const after init but read for clarity

        /// Wake up at least one thread that will wake up other threads if required
        tryWakeUpAnyOtherThreadWithTasks(context, lock);

        /// Return the number of additional pipeline threads the caller should target: pushed
        /// tasks that idle threads can't cover, clamped to remaining pipeline-width headroom
        /// (`num_threads - total_slots`). `PipelineExecutor` adds this delta to the
        /// `cpu_slots->setMax(...)` target so the slot allocator grows in step with demand.
        const size_t coverable = std::min(pushed, idle_before);
        const size_t uncovered = pushed - coverable;
        const size_t headroom = (num_threads_now > total_slots_now) ? (num_threads_now - total_slots_now) : 0;
        return std::min(uncovered, headroom);
    }
    return 0; // No new tasks -- no need for new threads

The arithmetic is: pushed tasks were created. Idle threads can absorb that many without being woken up. The remainder is genuine demand for new threads — clamped to the pipeline’s own width ceiling (num_threads, from max_threads), because the plan has a fixed number of streams and there’s no point running more threads than there is parallelism.

This is the demand signal. It’s computed from two observable facts — how many tasks were pushed, how many threads are parked — and nothing else. No profiling, no heuristics, no timers.

The executor then accumulates that demand and acts on it:

cpp
// src/Processors/Executors/Runtime/V1/PipelineExecutor.cpp:492
            /// Upscaling: `pushTasks` tells us how many additional threads would usefully
            /// parallelize the tasks it just enqueued. Publish our spawn_count into the shared
            /// `pending_demand` counter; the thread that actually holds `spawn_mutex` drains it
            /// so no demand is lost when multiple threads push concurrently.
            if (pool && spawn_count > 0)
                pending_demand.fetch_add(spawn_count, std::memory_order_relaxed);

            // Drain pending_demand under `spawn_mutex`. A thread that fails `try_lock`
            // must not exit while `pending_demand > 0` -- otherwise its add could race
            // with the owner's final `exchange(0)` and be left stranded after the owner
            // unlocks. The outer loop holds the thread until either it drains the demand
            // itself or observes the counter return to zero (meaning the owner caught it).
            while (pool && pending_demand.load(std::memory_order_acquire) > 0)
            {
                if (!spawn_mutex.try_lock())
                {
                    std::this_thread::yield();
                    continue;
                }
                try
                {
                    std::lock_guard lock(spawn_mutex, std::adopt_lock);
                    while (true)
                    {
                        size_t accumulated = pending_demand.exchange(0, std::memory_order_acq_rel);
                        if (accumulated == 0)
                            break;
                        size_t target = std::min<size_t>(max_pipeline_threads, desired_threads + accumulated);
                        if (target > desired_threads)
                        {
                            desired_threads = target;
                            cpu_slots->setMax(target);
                        }
                    }
                    spawnThreads({});
                }

This is small and rather careful concurrent code, and the comments are essentially a post-mortem.

The pending_demand counter is atomic so that threads which fail to take spawn_mutex don’t lose their demand. But a naive fetch_add would race with the owner’s exchange(0): thread A adds 1, the owner exchanges and gets 0, the owner unlocks, and thread A’s 1 sits in the counter forever.

The outer while loop is the fix. A thread that can’t get the mutex keeps spinning (yielding) until either it drains the demand itself or sees the counter back at zero, which means the owner already caught it.

The target computation is the second piece of policy: demand is accumulated into desired_threads, capped at max_pipeline_threads (the plan’s width), and applied by raising the slot allocator’s ceiling with cpu_slots->setMax(target).

spawnThreads then converts slots into actual OS threads:

cpp
// src/Processors/Executors/Runtime/V1/PipelineExecutor.cpp:743
void PipelineExecutor::spawnThreads(AcquiredSlotPtr slot)
{
    while (cpu_slots)
    {
        if (!slot)
            slot = cpu_slots->tryAcquire();
        if (!slot)
            return;

        size_t thread_num = slot->slot_id;

        /// Count of threads in use should be updated for proper finish() condition.
        /// `upscale` returns the remaining headroom for spawning. Even when >0 we spawn
        /// one thread per iteration and re-enter `upscale` after the next `tryAcquire`.
        const size_t remaining_capacity = tasks.upscale(thread_num);

        /// Start new thread
        pool->scheduleOrThrowOnError([this, thread_num, thread_group = CurrentThread::getGroup(), my_slot = std::move(slot)]
        {
            ThreadGroupSwitcher switcher(thread_group, ThreadName::QUERY_ASYNC_EXECUTOR);

            try
            {
                executeSingleThread(thread_num, WorkloadResources(my_slot.get(), process_list_element));
            }
            catch (...)
            {
                /// In case of exception from executor itself, stop other threads.
                finish();
                tasks.getThreadContext(thread_num).setException(std::current_exception());
            }
        });

The tryAcquire is the gate. The slot allocator may refuse to hand out a slot if the server is out of CPU budget. If it refuses, no thread is spawned and the pipeline simply runs with fewer threads.

The thread count is not max_threads; it’s max_threads subject to what the scheduler is willing to give you.

CPU slots: `max_threads` is a ceiling, not a promise

The slot allocation is where the concurrency-control policy lives. There are four cases, and the comment enumerates them:

cpp
// src/Processors/Executors/Runtime/V1/PipelineExecutor.cpp:605
    // There are a few possible modes of concurrency control.
    // 1) If we have `CREATE RESOURCE cpu (MASTER THREAD, WORKER THREAD)`:
    //    all thread should be allocated through scheduler using the same resource link
    // 2) If we have `CREATE RESOURCE cpu (WORKER THREAD)`:
    //    master thread uses "free" noncompeting CPU slot, the other threads are allocated through resource scheduler
    // 3) If we have two resources `CREATE RESOURCE master_cpu (MASTER THREAD)` and `CREATE RESOURCE worker_cpu (WORKER THREAD)`:
    //    all thread should be allocated through scheduler using different resource links
    // 4) If we have no cpu-related resources:
    //    the ConcurrencyControl class is used instead of resource scheduler

In the default case (case 4) the allocation is deliberately lazy:

cpp
// src/Processors/Executors/Runtime/V1/PipelineExecutor.cpp:672
        else
        {
            /// Allocate CPU slots from concurrency control with guaranteed master thread slot.
            /// Default path (lazy_allocation=true): start at max=master_threads and grow on
            /// demand via setMax from the upscaling block. Zero-waste per idle pipeline.
            /// Rollback (lazy_allocation=false): pre-branch eager allocation — request the
            /// full ceiling upfront and never call setMax.
            SlotCount initial_max = lazy_allocation ? master_threads : num_threads;
            return ConcurrencyControl::instance().allocate(master_threads, initial_max);
        }

“Zero-waste per idle pipeline” is the point. A server running a thousand concurrent queries, each of which is a small scan, doesn’t want a thousand times max_threads slots reserved. It wants one slot each, growing if the query turns out to be wide.

The lazy flag is read exactly once, and the reason is a real bug:

cpp
// src/Processors/Executors/Runtime/V1/PipelineExecutor.cpp:696
    /// Read the flag ONCE. A config reload mid-initialization would otherwise split a
    /// single query between lazy and eager strategies (e.g. allocate lazy with max=1 but
    /// then initialize desired_threads=num_threads, stranding subsequent setMax as a no-op).
    const bool lazy_allocation = ConcurrencyControl::instance().getLazyAllocation();

A config reload between “allocate the slots” and “set desired_threads” would leave the query in a state where setMax is a no-op and the pipeline never grows. One read, one strategy, for the lifetime of the query.

Preemption and downscaling

Threads can also be taken away. Under the resource scheduler with preemption enabled, a thread’s CPU lease is renewed periodically, and the renewal can fail:

cpp
// src/Processors/Executors/Runtime/V1/PipelineExecutor.cpp:539
            if (resources.isCPULeaseRenewNeeded()) // Check if preemption is enabled (see `cpu_slot_preemption` server setting)
            {
                try
                {
                    // Preemption point. Renewal could block execution due to CPU overload.
                    // It may trigger callbacks to tasks.preempt() and tasks.resume()
                    if (!resources.renewCPULease())
                    {
                        tasks.downscale(resources.getCPULeaseSlotId());
                        break; // Downscaling. Unable to renew the lease - thread should stop (but could be rerun later).
                    }

The renewal is checked at most every millisecond:

cpp
// src/Processors/Executors/Runtime/V1/PipelineExecutor.cpp:119
    bool isCPULeaseRenewNeeded()
    {
        if (!lease)
            return false;
        UInt64 now_ns = clock_gettime_ns();
        if (now_ns - last_renew_ns < 1000000) // 1 ms: do not renew too often
            return false;
        last_renew_ns = now_ns;
        return true;
    }

Downscaling is not “kill the thread”. It’s “let this thread stop when it finishes its current task, and don’t schedule more onto it”:

cpp
// src/Processors/Executors/Runtime/V1/ExecutorTasks.cpp:334
void ExecutorTasks::downscale(size_t slot_id)
{
    std::lock_guard lock(mutex);

    if (slot_id >= slot_count.size() || slot_count[slot_id] == 0)
        return;
    --slot_count[slot_id];

    if (slot_id + 1 == use_threads)
    {
        // The highest slot was downscaled, find new highest
        for (size_t i = use_threads; i > 0; --i)
        {
            if (slot_count[i - 1] > 0)
            {
                use_threads = i;
                break;
            }
        }
    }
}

Preemption is a third thing: the slot is released without downscaling, so the pipeline’s width is unchanged but the thread isn’t running. If the preempted thread was holding a task, that task is pushed back to the queue so another thread can steal it:

cpp
// src/Processors/Executors/Runtime/V1/ExecutorTasks.cpp:356
void ExecutorTasks::preempt(size_t slot_id)
{
    std::unique_lock lock(mutex);
    --total_slots;

    /// We should make sure that preempted thread has no local task inside context.
    /// It is allowed to have tasks in `task_queue` or `fast_task_queue` because they can be stealed by other threads.
    auto & context = executor_contexts[slot_id];
    if (auto * task = context->popTask())
    {
        task_queue.push(task, slot_id);
        /// Wake up at least one thread to avoid deadlocks (all other threads maybe idle)
        tryWakeUpAnyOtherThreadWithTasks(*context, lock); // this releases the lock if it wakes up a thread
    }

The asymmetry is the point. A local task (held in the thread’s context) is invisible to other threads and must be handed back explicitly, whereas a task in the shared queue is stealable and needs no action.

The pulling mode: `executeUntil`

Not every query is executed to completion in one go. The HTTP protocol, for instance, wants to stream rows to the client as they become available, and the client may disconnect at any moment. That’s what executeUntil() is for:

cpp
// src/Processors/Executors/Runtime/V1/PipelineExecutor.cpp:261
bool PipelineExecutor::executeUntil(std::atomic_bool * yield_flag)
{
    const auto uncaught_exceptions = std::uncaught_exceptions();
    SCOPE_EXIT({
        if (tasks.isFinished() || std::uncaught_exceptions() > uncaught_exceptions)
            detachSpillables();
    });
    if (!is_execution_initialized)
    {
        initializeExecution(1, true);

        // Acquire slot until we are done
        single_thread_cpu_slot = cpu_slots->acquire();
        tasks.upscale(single_thread_cpu_slot->slot_id);
        chassert(single_thread_cpu_slot && "Unable to allocate cpu slot for the first thread, but we just allocated at least one slot");

        if (yield_flag && *yield_flag)
            return true;
    }

    executeStepImpl(0, WorkloadResources(single_thread_cpu_slot.get(), process_list_element), yield_flag);

    if (!tasks.isFinished())
        return true;

The yield_flag is the interesting parameter. It’s checked at the bottom of the worker loop:

cpp
// src/Processors/Executors/Runtime/V1/PipelineExecutor.cpp:576
            /// We have executed single processor. Check if we need to yield execution.
            if (yield_flag && *yield_flag)
                break;

So the executor runs one processor at a time and then checks the flag. If the flag says “enough for now”, it returns to the caller, which can hand the chunk to the client and then call executeUntil() again. This is the “pulling” model at the level of the whole pipeline: the caller pulls.

PullingPipelineExecutor::pull() is the caller:

cpp
// src/Processors/Executors/PullingPipelineExecutor.cpp:49
bool PullingPipelineExecutor::pull(Chunk & chunk)
{
    if (!executor)
    {
        executor = createExecutor(pipeline.processors, pipeline.process_list_element);
        executor->setReadProgressCallback(pipeline.getReadProgressCallback());
        executor->setStepProfiler(pipeline.getStepProfiler());
    }

    /// 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`, and `executeUntil` below
    /// finalizes it - the pending read progress is reported and the format is finalized - before the end of the
    /// data is reported.
    if (pipeline.process_list_element && !pipeline.process_list_element->checkTimeLimitSoft())
    {
        executor->cancel(IProcessor::CancelReason::CancelledByTimeout);
        pipeline.process_list_element->checkTimeLimit();
    }

    if (!executor->executeUntil(&has_data_flag))
        return false;

    chunk = pulling_format->getChunk();
    return true;
}

has_data_flag is set by the output format when it has a chunk ready. So the loop is: run the pipeline until the format has produced something (or the pipeline is finished), then hand the chunk up.

Note that executeUntil runs the executor in single-threaded mode (initializeExecution(1, true)). The pulling executor is deliberately not parallel — it exists for cases like INSERT ... SELECT into a table, or formatting a result to a client that wants to consume it incrementally.

Trade-offs

What this buys you.

The pool grows with the work, not with the setting. A query that turns out to be narrow never pays for more than one or two threads, even if max_threads = 64. A query that turns out to be wide gets threads as fast as the scheduler will give them, without any startup latency. And the measurement is exact rather than heuristic: pushTasks counts the tasks it created and subtracts the threads that could absorb them.

Work stealing with per-thread affinity gives good cache behaviour and good load balance. The common case (a thread feeding itself) is a pointer swap with no locking; the uncommon case (one thread has all the work) is handled by theft.

The separation of “what work exists” from “how many resources it gets” also means the same executor serves queries with max_threads = 1 (no contention at all, the single thread takes everything locally) and queries with max_threads = 128.

What it costs.

The machinery is intricate, and the comments are full of post-mortems: the try_lock/pending_demand race, the lazy_allocation config reload, the max_scheduled_local_tasks starvation. Each of those is a bug that shipped and was fixed. The design is sound, but the implementation is carrying the weight of a decade of production.

There’s also a hard limit on the demand signal. pushTasks measures the tasks pushed by one processor at one moment. If a pipeline is wide but each individual push produces one task, the demand signal is one at a time, and the pool grows one thread per push rather than immediately. The initial_parallel computation in initializeExecution exists precisely to patch this for the first wave:

cpp
// src/Processors/Executors/Runtime/V1/PipelineExecutor.cpp:722
    /// Initial queued parallelism never routes through `pushTasks`, so size setMax here to
    /// cover it. For multi-source pipelines (e.g. UNION ALL of N subqueries) this prevents
    /// the pipeline from being stuck single-threaded if the master's first task happens not
    /// to push more work. Only matters on the lazy path; the eager path already has the full
    /// ceiling. `initial_parallel` is the total target — the master thread is one of its
    /// consumers, no +1 needed.
    if (lazy_allocation && initial_parallel > 1)

The preemption path is the other sharp edge. A thread that can’t renew its lease stops, and the pipeline continues with fewer threads — but if every thread is preempted at once, the pipeline must not deadlock. The threads_queue.size() + 1 == total_slots check in tryGetTask is the mechanism that detects “the last worker is about to go idle” and calls finish():

cpp
// src/Processors/Executors/Runtime/V1/ExecutorTasks.cpp:158
        /// This thread has no tasks to do and is going to wait.
        /// Finish execution if this was the last active thread.
        chassert(task_queue.empty() && fast_task_queue.empty());
        if (threads_queue.size() + 1 == total_slots && async_task_queue.empty())
        {
            lock.unlock();
            finish();
            return;
        }

Getting this wrong in either direction is bad: finish too early and the query returns partial data, never finish and the query hangs. The chassert documents that the two conditions must hold together — if the queues are not empty, the pipeline is not finished and the thread must keep waiting.

Finally, there’s the question of when to stop growing. desired_threads only ever goes up within a query. A pipeline that’s wide at the start and narrow at the end keeps its threads; there’s no downscaling based on demand, only downscaling due to preemption. For a long-running query over a table that becomes narrow, that’s wasted capacity.

Wrapping up

The real win: ClickHouse doesn’t guess how many threads your query needs. It measures. pushTasks returns a number — how many new tasks idle threads can’t cover — and the pool grows by exactly that much, capped by the plan’s own width.

Two things to internalize:

max_threads is a ceiling, not a schedule. If you see a query using 2 threads on a 64-core box, that’s the demand signal working, not a bug.
The server can take threads back from you. Preemption and downscaling are first-class, and your query is expected to keep going on fewer threads.

So when you profile a slow query, don’t count threads first. Count streams, and ask whether the pipeline had enough independent work to push into idle threads in the first place.