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



{
        pipes[0].holder = std::move(res.holder);
        return std::move(pipes[0]);
    }
    OutputPortRawPtrs totals;
    OutputPortRawPtrs extremes;
    res.header = pipes.front().header;
    res.collected_processors = collected_processors;
    for (auto & pipe : pipes)
    {
        assertBlocksHaveEqualStructure(res.header, pipe.header, "Pipe::unitePipes");
        res.processors.insert(res.processors.end(), pipe.processors.begin(), pipe.processors.end());
        res.output_ports.insert(res.output_ports.end(), pipe.output_ports.begin(), pipe.output_ports.end());
        res.max_parallel_streams += pipe.max_parallel_streams;
        if (pipe.totals_port)
            totals.emplace_back(pipe.totals_port);
        if (pipe.extremes_port)
            extremes.emplace_back(pipe.extremes_port);
    }
    size_t num_processors = res.processors.size();
    res.totals_port = uniteTotals(totals, res.header, res.processors);
    res.extremes_port = uniteExtremes(extremes, res.header, res.processors);
    if (res.collected_processors)
    {
        for (; num_processors < res.processors.size(); ++num_processors)
            res.collected_processors->emplace_back(res.processors[num_processors]);
    }
    return res;
}
void Pipe::addSource(ProcessorPtr source)