```sql
SELECT count()
FROM 
(
    SELECT number
    FROM numbers_mt(100000000)
    ORDER BY number DESC
    LIMIT 100
    UNION ALL
    SELECT number
    FROM numbers_mt(100000000)
    ORDER BY number DESC
    LIMIT 100
    UNION ALL
    SELECT number
    FROM numbers_mt(100000000)
    ORDER BY number DESC
    LIMIT 100
)


SET experimental_use_processors = 0;
-- 1 rows in set. Elapsed: 1.581 sec. Processed 300.00 million rows, 2.40 GB (189.78 million rows/s., 1.52 GB/s.) 

SET experimental_use_processors = 1;
-- 1 rows in set. Elapsed: 3.569 sec. Processed 300.00 million rows, 2.40 GB (84.06 million rows/s., 672.50 MB/s.) 
```

Similar for MergeTree:

```sql
create table test engine=MergeTree order by tuple() as select * from numbers(30000000);

select count() from (select number from test group by number order by number limit 10 union all select number from test group by number order by number limit 10);

set experimental_use_processors=0;
-- Elapsed: 2.664 sec

set experimental_use_processors=1;
-- Elapsed: 3.656 sec
```
void QueryPipeline::addPipe(Processors pipe)
{
    checkInitializedAndNotCompleted();
    if (pipe.empty())
        throw Exception("Can't add empty processors list to QueryPipeline.", ErrorCodes::LOGICAL_ERROR);
    auto & first = pipe.front();
    auto & last = pipe.back();
    auto num_inputs = first->getInputs().size();
    if (num_inputs != streams.size())
        throw Exception("Can't add processors to QueryPipeline because first processor has " + toString(num_inputs) +
                        " input ports, but QueryPipeline has " + toString(streams.size()) + " streams.",
                        ErrorCodes::LOGICAL_ERROR);
    auto stream = streams.begin();
    for (auto & input : first->getInputs())
        connect(**(stream++), input);
    Block header;
    streams.clear();
    streams.reserve(last->getOutputs().size());
    for (auto & output : last->getOutputs())
    {
        streams.addStream(&output, 0);
        if (header)
            assertBlocksHaveEqualStructure(header, output.getHeader(), "QueryPipeline");
        else
            header = output.getHeader();
    }
    processors.emplace(pipe);
    current_header = std::move(header);
}
void QueryPipeline::addDelayedStream(ProcessorPtr source)