Method addSource

src/QueryPipeline/Pipe.cpp:347–364

}

void Pipe::addSource(ProcessorPtr source)
{
    checkSource(*source);
    const auto & source_header_ptr = source->getOutputs().front().getSharedHeader();

    if (output_ports.empty())
        header = source_header_ptr;
    else
        assertBlocksHaveEqualStructure(*header, *source_header_ptr, "Pipes");

    if (collected_processors)
        collected_processors->emplace_back(source);

    output_ports.push_back(&source->getOutputs().front());
    processors->emplace_back(std::move(source));

    max_parallel_streams = std::max<size_t>(max_parallel_streams, output_ports.size());
}

void Pipe::addTotalsSource(ProcessorPtr source)
{