Function doExecuteTask

src/QueryPipeline/DistributedPlanExecutor.cpp:607–731

}

void doExecuteTask(const DistributedQueryTaskDescription & task_description, ObjectStoragePtr object_storage,
    const String & object_storage_path, const String & distributed_query_id, ContextMutablePtr context,
    std::function<bool()> is_cancelled, ProgressCallback progress_callback)
{
    Stopwatch execute_task_watch;
    const auto & task = task_description.task;

    std::shared_ptr<OpenTelemetry::SpanHolder> query_span = std::make_shared<OpenTelemetry::SpanHolder>(task.task_id);

    auto logger = Poco::Logger::getShared("executeDistributedQuery");

    /// Disable the query condition cache: its per-worker state could make workers read inconsistent
    /// data for the same fragment.
    context->setSetting("use_query_condition_cache", false);

    Strings input_exchange_streams;
    for (const auto & stream_id : task.input_exchange_streams)
        input_exchange_streams.push_back(stream_id.toString());

    Strings output_exchange_streams;
    for (const auto & stream_id : task.output_exchange_streams)
        output_exchange_streams.push_back(stream_id.toString());

    LOG_TRACE(logger, "Task '{}' input exchange streams: [{}], output exchange streams: [{}]",
        task.task_id, fmt::join(input_exchange_streams, ", "), fmt::join(output_exchange_streams, ", "));

#ifdef OS_LINUX
    /// Release this task's pending streaming exchange connections on the worker when it ends. A
    /// consumer that never connects (e.g. its query was cancelled) would otherwise leave them behind.
    /// Only this task's output streams are dropped, so sibling tasks of the same query are unaffected.
    SCOPE_EXIT_SAFE(ExchangeConnections::instance()->removePendingStreams(distributed_query_id, output_exchange_streams));
#endif

    auto temporary_files = createTemporaryFilesLookup(
        object_storage, object_storage_path, input_exchange_streams, output_exchange_streams);

    auto pipeline_settings = BuildQueryPipelineSettings(context);
    pipeline_settings.temporary_file_lookup = temporary_files;
    pipeline_settings.parameter_lookup = std::make_shared<TaskParameters>(task.parameters);
    pipeline_settings.exchange_lookup = createExchangeLookup(
        distributed_query_id,
        task_description.exchanges,
        task_description.exchange_stream_sources,
        temporary_files,
        context);

    auto optimization_settings = QueryPlanOptimizationSettings(context);

    /// Disable stats-driven plan-shape rewrites on the worker side: per-worker
    /// stats can diverge and produce incompatible plans across workers (e.g. one
    /// swaps the join sides while the others don't), breaking exchange partitioning.
    optimization_settings.join_swap_table = std::make_optional(false);
    optimization_settings.query_plan_optimize_join_order_limit = 0;
    optimization_settings.query_plan_optimize_join_order_randomize = 0;
    optimization_settings.convert_join_to_in = false;
    optimization_settings.convert_outer_join_to_inner_join = false;
    optimization_settings.convert_any_join_to_semi_or_anti_join = false;
    optimization_settings.merge_filter_into_join_condition = false;