
**Describe the bug**
See the code in `Pipe.cpp`. The logic of `if` not consistent with the logs below.  Can `checkSource` be passed when `source.getOutputs().size() == 2` ? 
```
static void checkSource(const IProcessor & source)
{
    ......
    if (source.getOutputs().size() > 1)
        throw Exception("Source for pipe should have single or two outputs, but " + source.getName() + " has " +
                        toString(source.getOutputs().size()) + " outputs.", ErrorCodes::LOGICAL_ERROR);
}
``` 



{
    checkSource(*source);
    const auto & source_header = source->getOutputs().front().getHeader();
    if (output_ports.empty())
        header = source_header;
    else
        assertBlocksHaveEqualStructure(header, source_header, "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)
{
    if (output_ports.empty())
        throw Exception("Cannot add totals source to empty Pipe.", ErrorCodes::LOGICAL_ERROR);
    if (totals_port)
        throw Exception("Totals 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);
    totals_port = &source->getOutputs().front();
    processors.emplace_back(std::move(source));
}
void Pipe::addExtremesSource(ProcessorPtr source)