
**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 (ports.empty())
        return nullptr;
    if (ports.size() == 1)
        return ports.front();
    
    
    
    auto concat = std::make_shared<ConcatProcessor>(header, ports.size());
    auto limit = std::make_shared<LimitTransform>(header, 1, 0);
    auto * totals_port = &limit->getOutputPort();
    auto in = concat->getInputs().begin();
    for (const auto & port : ports)
        connect(*port, *(in++));
    connect(concat->getOutputs().front(), limit->getInputPort());
    processors.emplace_back(std::move(concat));
    processors.emplace_back(std::move(limit));
    return totals_port;
}
Pipe::Holder & Pipe::Holder::operator=(Holder && rhs)
{
    table_locks.insert(table_locks.end(), rhs.table_locks.begin(), rhs.table_locks.end());
    storage_holders.insert(storage_holders.end(), rhs.storage_holders.begin(), rhs.storage_holders.end());
    interpreter_context.insert(interpreter_context.end(),
                               rhs.interpreter_context.begin(), rhs.interpreter_context.end());
    for (auto & plan : rhs.query_plans)
        query_plans.emplace_back(std::move(plan));
    return *this;
}
Pipe::Pipe(ProcessorPtr source, OutputPort * output, OutputPort * totals, OutputPort * extremes)