
**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 set sink to empty Pipe.", ErrorCodes::LOGICAL_ERROR);
    auto add_transform = [&](OutputPort *& stream, Pipe::StreamType stream_type)
    {
        if (!stream)
            return;
        auto transform = getter(stream->getHeader(), stream_type);
        if (transform)
        {
            if (transform->getInputs().size() != 1)
                throw Exception("Sink for query pipeline transform should have single input, "
                                "but " + transform->getName() + " has " +
                                toString(transform->getInputs().size()) + " inputs.", ErrorCodes::LOGICAL_ERROR);
            if (!transform->getOutputs().empty())
                throw Exception("Sink for query pipeline transform should have no outputs, "
                                "but " + transform->getName() + " has " +
                                toString(transform->getOutputs().size()) + " outputs.", ErrorCodes::LOGICAL_ERROR);
        }
        if (!transform)
            transform = std::make_shared<NullSink>(stream->getHeader());
        connect(*stream, transform->getInputs().front());
        processors.emplace_back(std::move(transform));
    };
    for (auto & port : output_ports)
        add_transform(port, StreamType::Main);
    add_transform(totals_port, StreamType::Totals);
    add_transform(extremes_port, StreamType::Extremes);
    output_ports.clear();