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



}
    processors.insert(processors.end(), new_processors.begin(), new_processors.end());
    max_parallel_streams = std::max<size_t>(max_parallel_streams, output_ports.size());
}
void Pipe::setLimits(const StreamLocalLimits & limits)
{
    for (auto & processor : processors)
    {
        if (auto * source_with_progress = dynamic_cast<ISourceWithProgress *>(processor.get()))
            source_with_progress->setLimits(limits);
    }
}
void Pipe::setLeafLimits(const SizeLimits & leaf_limits)
{
    for (auto & processor : processors)
    {
        if (auto * source_with_progress = dynamic_cast<ISourceWithProgress *>(processor.get()))
            source_with_progress->setLeafLimits(leaf_limits);
    }
}
void Pipe::setQuota(const std::shared_ptr<const EnabledQuota> & quota)
{
    for (auto & processor : processors)
    {
        if (auto * source_with_progress = dynamic_cast<ISourceWithProgress *>(processor.get()))
            source_with_progress->setQuota(quota);
    }
}
}