Method transformPipeline

src/QueryPlan/MergingAggregatedStep.cpp:110–159

}

void MergingAggregatedStep::transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings & build_settings)
{
    if (hasNonParallelAggregateFunctions(params->params.aggregates))
    {
        pipeline.resize(1);
    }

    // optimizer use MergingAggregateStep in by-name style, regenerate aggregator params of by-position style
    if (!keys.empty())
    {
        ColumnNumbers key_positions;
        const auto & header = pipeline.getHeader();
        for (const auto & key : keys)
            key_positions.emplace_back(header.getPositionByName(key));

        Aggregator::Params new_params(
            header,
            key_positions,
            params->params.aggregates,
            params->params.overflow_row,
            build_settings.context->getSettingsRef().max_threads);
        params = std::make_shared<AggregatingTransformParams>(new_params, params->final);
    }

    // @FIXME: grouping sets + two-level aggregation is incompatible with memory efficient merge
    // see also: https://meego.feishu.cn/clickhousech/story/detail/14744099
    if (!memory_efficient_aggregation || input_streams.front().header.has("__grouping_set"))
    {
        /// We union several sources into one, paralleling the work.
        pipeline.resize(1);

        /// Now merge the aggregated blocks
        pipeline.addSimpleTransform([&](const Block & header)
        {
            return std::make_shared<MergingAggregatedTransform>(header, params, max_threads);
        });
    }
    else
    {
        auto num_merge_threads = memory_efficient_merge_threads
                                 ? static_cast<size_t>(memory_efficient_merge_threads)
                                 : static_cast<size_t>(max_threads);

        pipeline.addMergingAggregatedMemoryEfficientTransform(params, num_merge_threads);
    }

    computeGroupingFunctions(pipeline, groupings, keys, grouping_sets_params, build_settings);

    pipeline.resize(should_produce_results_in_order_of_bucket_number ? 1 : max_threads);
}

void MergingAggregatedStep::describeActions(FormatSettings & settings) const
{