Method addTotalsSource

src/QueryPipeline/Pipe.cpp:366–384

}

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

    if (totals_port)
        throw Exception(ErrorCodes::LOGICAL_ERROR, "Totals source was already added to Pipe");

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

    assertBlocksHaveEqualStructure(*header, *source_header, "Pipes");

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

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

void Pipe::addExtremesSource(ProcessorPtr source)
{