Method addSplitResizeTransform

src/QueryPipeline/Pipe.cpp:688–750

}

void Pipe::addSplitResizeTransform(size_t num_streams, size_t min_outstreams_per_resize_after_split, bool strict)
{
    OutputPortRawPtrs resize_output_ports(num_streams);

    size_t groups = std::min<size_t>(numOutputPorts(), num_streams / min_outstreams_per_resize_after_split);
    size_t instream_per_group = (numOutputPorts() + groups - 1) / groups;
    size_t groups_with_extra_instream = numOutputPorts() % groups;
    size_t outstreams_per_group = (num_streams + groups - 1) / groups;
    size_t groups_with_extra_outstream = num_streams % groups;

    chassert(groups > 1);

    for (size_t i = 0, next_input = 0, next_output = 0; i < groups; ++i)
    {
        ProcessorPtr resize;
        if (strict)
            resize = std::make_shared<StrictResizeProcessor>(getSharedHeader(), instream_per_group, outstreams_per_group);
        else
            resize = std::make_shared<ResizeProcessor>(getSharedHeader(), instream_per_group, outstreams_per_group);

        for (auto it = resize->getInputs().begin(); it != resize->getInputs().end(); ++it)
        {
            if (std::next(it) != resize->getInputs().end() || groups_with_extra_instream == 0 || i < groups_with_extra_instream)
            {
                connect(*output_ports[next_input], *it);
                ++next_input;
            }
            else
            {
                auto null_source = std::make_shared<NullSource>(getSharedHeader());
                connect(null_source->getPort(), *it);
                processors->emplace_back(std::move(null_source));
            }
        }

        for (auto it = resize->getOutputs().begin(); it != resize->getOutputs().end(); ++it)
        {
            if (std::next(it) != resize->getOutputs().end() || groups_with_extra_outstream == 0 || i < groups_with_extra_outstream)
            {
                resize_output_ports[next_output] = &*it;
                ++next_output;
            }
            else
            {
                auto null_sink = std::make_shared<NullSink>(getSharedHeader());
                connect(*it, null_sink->getPort());
                processors->emplace_back(std::move(null_sink));
            }
        }

        if (collected_processors)
            collected_processors->emplace_back(resize);
        processors->emplace_back(std::move(resize));
    }

    output_ports = std::move(resize_output_ports);

    header = output_ports.front()->getSharedHeader();