
**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 transform empty Pipe.", ErrorCodes::LOGICAL_ERROR);
    auto new_processors = transformer(output_ports);
    
    std::unordered_set<const IProcessor *> set;
    for (const auto & processor : new_processors)
        set.emplace(processor.get());
    for (const auto & port : output_ports)
    {
        if (!port->isConnected())
            throw Exception("Transformation of Pipe is not valid because output port (" +
                            port->getHeader().dumpStructure() + ") is not connected", ErrorCodes::LOGICAL_ERROR);
        set.emplace(&port->getProcessor());
    }
    output_ports.clear();
    for (const auto & processor : new_processors)
    {
        for (const auto & port : processor->getInputs())
        {
            if (!port.isConnected())
                throw Exception("Transformation of Pipe is not valid because processor " + processor->getName() +
                                " has not connected input port", ErrorCodes::LOGICAL_ERROR);
            const auto * connected_processor = &port.getOutputPort().getProcessor();
            if (set.count(connected_processor) == 0)
                throw Exception("Transformation of Pipe is not valid because processor " + processor->getName() +
                                " has input port which is connected with unknown processor " +