Method initializeExecution

src/Processors/Executors/PipelineExecutor.cpp:578–627

}

void PipelineExecutor::initializeExecution(size_t num_threads, bool concurrency_control)
{
    is_execution_initialized = true;
    tryUpdateExecutionStatus(ExecutionStatus::NotStarted, ExecutionStatus::Executing);

    /// Capture the ceiling so the upscaling block can cap `setMax` at this value.
    max_pipeline_threads = num_threads;

    /// 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();

    cpu_slots = allocateCPU(num_threads, concurrency_control, lazy_allocation);

    /// If rollback flag is off, we used eager allocate(1, num_threads) and will not call
    /// setMax in the upscaling block (nothing to grow). Initialize desired_threads
    /// to the full ceiling so the growth check in the block becomes a no-op.
    desired_threads = lazy_allocation ? 1 : num_threads;

    Queue queue;
    Queue async_queue;
    graph->initializeExecution(queue, async_queue);

    /// use_threads should reflect number of thread spawned and can grow with tasks.upscale(...).
    /// Starting from 1 instead of 0 is to tackle the single thread scenario, where no upscale() will
    /// be invoked but actually 1 thread used.
    tasks.init(num_threads, 1, cpu_slots, profile_processors, trace_processors, read_progress_callback.get());
    const size_t initial_parallel = tasks.fill(queue, async_queue);

    /// 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)
    {
        std::lock_guard lock(spawn_mutex);
        const size_t target = std::min<size_t>(max_pipeline_threads, initial_parallel);
        if (target > desired_threads)
        {
            desired_threads = target;
            cpu_slots->setMax(target);
        }
    }

    if (num_threads > 1)
        pool = std::make_unique<ThreadPool>(CurrentMetrics::QueryPipelineExecutorThreads, CurrentMetrics::QueryPipelineExecutorThreadsActive, CurrentMetrics::QueryPipelineExecutorThreadsScheduled, num_threads);
}

void PipelineExecutor::spawnThreads(AcquiredSlotPtr slot)
{