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



"Processor has " + std::to_string(inputs.size()) + " input ports, "
                        "but " + std::to_string(expected_inputs) + " expected", ErrorCodes::LOGICAL_ERROR);
    if (totals && !totals_port)
        throw Exception("Cannot add transform consuming totals to Pipe because Pipe does not have totals.",
                        ErrorCodes::LOGICAL_ERROR);
    if (extremes && !extremes_port)
        throw Exception("Cannot add transform consuming extremes to Pipe because it already has extremes.",
                        ErrorCodes::LOGICAL_ERROR);
    if (totals)
    {
        connect(*totals_port, *totals);
        totals_port = nullptr;
    }
    if (extremes)
    {
        connect(*extremes_port, *extremes);
        extremes_port = nullptr;
    }
    bool found_totals = false;
    bool found_extremes = false;
    size_t next_output = 0;
    for (auto & input : inputs)
    {
        if (&input == totals)
            found_totals = true;
        else if (&input == extremes)
            found_extremes = true;
        else
        {
            connect(*output_ports[next_output], input);