
**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 (totals && !found_totals)
        throw Exception("Cannot add transform " + transform->getName() + " to Pipes because "
                        "specified totals port does not belong to it", ErrorCodes::LOGICAL_ERROR);
    if (extremes && !found_extremes)
        throw Exception("Cannot add transform " + transform->getName() + " to Pipes because "
                        "specified extremes port does not belong to it", ErrorCodes::LOGICAL_ERROR);
    if (output_ports.empty())
        throw Exception("Cannot add transform " + transform->getName() + " to Pipes because it has no outputs",
                        ErrorCodes::LOGICAL_ERROR);
    header = output_ports.front()->getHeader();
    for (size_t i = 1; i < output_ports.size(); ++i)
        assertBlocksHaveEqualStructure(header, output_ports[i]->getHeader(), "Pipes");
    
    
    
    if (extremes_port)
        assertBlocksHaveEqualStructure(header, extremes_port->getHeader(), "Pipes");
    if (collected_processors)
        collected_processors->emplace_back(transform);
    processors.emplace_back(std::move(transform));
    max_parallel_streams = std::max<size_t>(max_parallel_streams, output_ports.size());
}
void Pipe::addTransform(ProcessorPtr transform, InputPort * totals, InputPort * extremes)
{
    if (output_ports.empty())
        throw Exception("Cannot add transform to empty Pipe.", ErrorCodes::LOGICAL_ERROR);
    auto & inputs = transform->getInputs();
    size_t expected_inputs = output_ports.size() + (totals ? 1 : 0) + (extremes ? 1 : 0);
    if (inputs.size() != expected_inputs)
        throw Exception("Cannot add transform " + transform->getName() + " to Pipes because "