Method scatterByPartitionIfNeeded

src/Processors/QueryPlan/SortingStep.cpp:316–370

}

void SortingStep::scatterByPartitionIfNeeded(QueryPipelineBuilder& pipeline)
{
    size_t threads = pipeline.getNumThreads();
    size_t streams = pipeline.getNumStreams();

    if (!partition_by_description.empty() && threads > 1)
    {
        /// We are going to shuffle the data from streams to threads. This will create (threads * streams) connections in the pipeline.
        /// Let's limit this by some sane value to avoid explosion.
        const size_t connection_count_limit = 1000000;
        if (threads * streams > connection_count_limit)
            throw Exception(ErrorCodes::LIMIT_EXCEEDED, "Parallelism limit exceeded in SortingStep: {} threads X {} streams, limit {}, try to reduce `max_threads` value",
                threads, streams, connection_count_limit);

        auto stream_header = pipeline.getSharedHeader();

        ColumnNumbers key_columns;
        key_columns.reserve(partition_by_description.size());
        for (auto & col : partition_by_description)
        {
            key_columns.push_back(stream_header->getPositionByName(col.column_name));
        }

        pipeline.transform([&](const OutputPortRawPtrs & ports)
        {
            Processors processors;
            for (auto * port : ports)
            {
                auto scatter = std::make_shared<ScatterByPartitionTransform>(stream_header, threads, key_columns);
                connect(*port, scatter->getInputs().front());
                processors.push_back(scatter);
            }
            return processors;
        });

        if (streams > 1)
        {
            pipeline.transform([&](const OutputPortRawPtrs & ports)
            {
                Processors processors;
                for (size_t i = 0; i < threads; ++i)
                {
                    size_t output_it = i;
                    auto resize = std::make_shared<ResizeProcessor>(stream_header, streams, 1);
                    auto & inputs = resize->getInputs();

                    for (auto input_it = inputs.begin(); input_it != inputs.end(); output_it += threads, ++input_it)
                        connect(*ports[output_it], *input_it);
                    processors.push_back(resize);
                }
                return processors;
            });
        }
    }
}

void SortingStep::finishSorting(
    QueryPipelineBuilder & pipeline, const SortDescription & input_sort_desc, const SortDescription & result_sort_desc, const UInt64 limit_)