Method addExtremesSource

src/Processors/Pipe.cpp:433–451

}

void Pipe::addExtremesSource(ProcessorPtr source)
{
    if (output_ports.empty())
        throw Exception("Cannot add extremes source to empty Pipe.", ErrorCodes::LOGICAL_ERROR);

    if (extremes_port)
        throw Exception("Extremes source was already added to Pipe.", ErrorCodes::LOGICAL_ERROR);

    checkSource(*source);
    const auto & source_header = output_ports.front()->getHeader();

    assertBlocksHaveEqualStructure(header, source_header, "Pipes");

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

    extremes_port = &source->getOutputs().front();
    processors.emplace_back(std::move(source));
}

static void dropPort(OutputPort *& port, Processors & processors, Processors * collected_processors)
{