Method updatePipeline

src/QueryPlan/UnionStep.cpp:58–133

}

QueryPipelinePtr UnionStep::updatePipeline(QueryPipelines pipelines, const BuildQueryPipelineSettings & settings)
{
    auto pipeline = std::make_unique<QueryPipeline>();
    QueryPipelineProcessorsCollector collector(*pipeline, this);

    if (pipelines.empty())
    {
        pipeline->init(Pipe(std::make_shared<NullSource>(output_stream->header)));
        processors = collector.detachProcessors();
        return pipeline;
    }

    size_t index = 0;
    for (auto & cur_pipeline : pipelines)
    {
        ASTPtr expr_list = std::make_shared<ASTExpressionList>();
        NamesWithAliases output_names;
        bool need_rename = false;
        for (const auto & item : output_stream->header)
        {
            auto rename_from = output_to_inputs.at(item.name).at(index);
            output_names.emplace_back(rename_from, item.name);
            ASTPtr identifier = std::make_shared<ASTIdentifier>(rename_from);
            identifier->setAlias(item.name);
            expr_list->children.emplace_back(identifier);
            if (item.name != rename_from)
            {
                need_rename = true;
            }
        }
        if (need_rename)
        {
            auto project_action
                = createExpressionActions(settings.context, cur_pipeline->getHeader().getNamesAndTypesList(), output_names, expr_list);
            auto expression = std::make_shared<ExpressionActions>(project_action, settings.getActionsSettings());
            cur_pipeline->addSimpleTransform(
                [&](const Block & header_) { return std::make_shared<ExpressionTransform>(header_, expression); });

            if (!blocksHaveEqualStructure(cur_pipeline->getHeader(), getOutputStream().header))
            {
                auto actions_dag = ActionsDAG::makeConvertingActions(
                    cur_pipeline->getHeader().getColumnsWithTypeAndName(),
                    getOutputStream().header.getColumnsWithTypeAndName(),
                    ActionsDAG::MatchColumnsMode::Position);
                auto converting_actions = std::make_shared<ExpressionActions>(std::move(actions_dag));
                cur_pipeline->addSimpleTransform(
                    [&](const Block & cur_header) { return std::make_shared<ExpressionTransform>(cur_header, converting_actions); });
            }
        }

        /// Headers for union must be equal.
        /// But, just in case, convert it to the same header if not.
        if (!isCompatibleHeader(cur_pipeline->getHeader(), getOutputStream().header))
        {
            auto converting_dag = ActionsDAG::makeConvertingActions(
                cur_pipeline->getHeader().getColumnsWithTypeAndName(),
                getOutputStream().header.getColumnsWithTypeAndName(),
                ActionsDAG::MatchColumnsMode::Name);