Function narrowPipe

src/QueryPipeline/narrowPipe.cpp:28–61

}

void narrowPipe(Pipe & pipe, size_t width)
{
    size_t size = pipe.numOutputPorts();
    if (size <= width)
        return;

    VectorWithMemoryTracking<OutputPortRawPtrs> partitions(width);

    auto distribution = getDistribution(size, width);

    pipe.transform([&](const OutputPortRawPtrs & ports)
    {
        for (size_t i = 0; i < size; ++i)
            partitions[distribution[i]].emplace_back(ports[i]);

        Processors concats;

        for (size_t i = 0; i < width; ++i)
        {
           auto concat = std::make_shared<ConcatProcessor>(partitions[i].at(0)->getSharedHeader(),
                                                           partitions[i].size());
           size_t next_port = 0;
           for (auto & port : concat->getInputs())
           {
               connect(*partitions[i][next_port], port);
               ++next_port;
           }

           concats.emplace_back(std::move(concat));
        }

        return concats;
    });
}

}