
**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);
}
``` 



{
    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)
{
    if (port == nullptr)
        return;
    auto null_sink = std::make_shared<NullSink>(port->getHeader());
    connect(*port, null_sink->getPort());
    if (collected_processors)
        collected_processors->emplace_back(null_sink);
    processors.emplace_back(std::move(null_sink));
    port = nullptr;
}
void Pipe::dropTotals()