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



check_port_from_source(extremes, "extremes");
        if (num_specified_ports != outputs.size())
            throw Exception("Cannot create Pipe from source because it has " + std::to_string(outputs.size()) +
                            " output ports, but " + std::to_string(num_specified_ports) + " were specified",
                            ErrorCodes::LOGICAL_ERROR);
    }
    totals_port = totals;
    extremes_port = extremes;
    output_ports.push_back(output);
    processors.emplace_back(std::move(source));
    max_parallel_streams = 1;
}
Pipe::Pipe(ProcessorPtr source)
{
    if (auto * source_from_input_stream = typeid_cast<SourceFromInputStream *>(source.get()))
    {
        
        totals_port = source_from_input_stream->getTotalsPort();
        extremes_port = source_from_input_stream->getExtremesPort();
    }
    else if (source->getOutputs().size() != 1)
        checkSource(*source);
    if (collected_processors)
        collected_processors->emplace_back(source);
    output_ports.push_back(&source->getOutputs().front());
    header = output_ports.front()->getHeader();
    processors.emplace_back(std::move(source));
    max_parallel_streams = 1;
}
Pipe::Pipe(Processors processors_) : processors(std::move(processors_))