Method convertToDistributed

src/Processors/QueryPlan/QueryPlan.cpp:794–936

}

void QueryPlan::convertToDistributed(const QueryPlanOptimizationSettings & optimization_settings)
{
    SharedHeader result_header = root->step->getOutputHeader();

    QueryPlan::Nodes old_nodes = std::move(nodes);
    QueryPlan::Node * old_root = root;
    root = nullptr;
    auto distributed_plan = QueryPlanOptimizations::makeDistributedPlan(std::move(old_nodes), old_root, optimization_settings);

    for (const auto & stage : distributed_plan.stages)
    {
        auto it = distributed_plan.stage_depends_on.find(stage.first);
        const auto & dependencies = it != distributed_plan.stage_depends_on.end() ? it->second : std::unordered_map<String, String>{};
        LOG_TRACE(getLogger("optimize"), "Distributed stage: '{}' depends on: [{}] plan:\n{}",
            stage.first, fmt::join(dependencies, ", "), dumpQueryPlan(stage.second.query_plan_fragment));
    }

    if (distributed_plan.stages.size() == 1)
    {
        /// For now just replace the plan with the first and only fragment, but preserve
        /// table locks and storage holders accumulated during planning.
        QueryPlanResourceHolder preserved_resources = std::move(resources);
        *this = std::move(distributed_plan.stages.begin()->second.query_plan_fragment);
        /// QueryPlanResourceHolder's move-assignment appends rhs into lhs without dropping existing entries.
        resources = std::move(preserved_resources);

        QueryPlanOptimizationSettings local_settings = optimization_settings;
        local_settings.make_distributed_plan = false;
        QueryPlanOptimizations::optimizeTreeSecondPass(local_settings, *root, nodes, *this);
    }
    else
    {
        ExchangeDescription final_result_exchange
        {
            .name = "final_result",
            .kind = optimization_settings.distributed_plan_force_exchange_kind == "Persisted" ? ExchangeDescription::Kind::Persisted : ExchangeDescription::Kind::Streaming,
            .source_bucket_count = 1,
            .destination_bucket_count = 1
        };
        auto result_stream_id = ExchangeStreamId(final_result_exchange.name, 0, 0);

        /// Add a step that writes the result of the main stage to the file
        auto & main_stage = distributed_plan.stages["main"];
        if (!main_stage.query_plan_fragment.isCompleted())
        {
            main_stage.query_plan_fragment.addStep(std::make_unique<GatherSendStep>(result_header, final_result_exchange.name));
            main_stage.tasks.front().output_exchange_streams.emplace_back(result_stream_id);
            distributed_plan.exchange_descriptions[final_result_exchange.name] = final_result_exchange;
            distributed_plan.final_result_stream_name = result_stream_id.toString();
        }

        /// Fail early (before execution) if any fragment contains a step that cannot be serialized
        /// for remote execution, instead of throwing late from serializeQueryPlan.
        for (const auto & [stage_name, stage] : distributed_plan.stages)
            assertFragmentSerializable(stage.query_plan_fragment, stage_name);

        /// Collect the list of all temporary files
        Strings all_temporary_files_for_cleanup;