```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
```
{
    checkInitializedAndNotCompleted();
    auto * format = dynamic_cast<IOutputFormat * >(output.get());
    if (!format)
        throw Exception("IOutputFormat processor expected for QueryPipeline::setOutputFormat.", ErrorCodes::LOGICAL_ERROR);
    if (output_format)
        throw Exception("QueryPipeline already has output.", ErrorCodes::LOGICAL_ERROR);
    output_format = format;
    resize(1);
    auto & main = format->getPort(IOutputFormat::PortKind::Main);
    auto & totals = format->getPort(IOutputFormat::PortKind::Totals);
    auto & extremes = format->getPort(IOutputFormat::PortKind::Extremes);
    if (!totals_having_port)
    {
        auto null_source = std::make_shared<NullSource>(totals.getHeader());
        totals_having_port = &null_source->getPort();
        processors.emplace(std::move(null_source));
    }
    if (!extremes_port)
    {
        auto null_source = std::make_shared<NullSource>(extremes.getHeader());
        extremes_port = &null_source->getPort();
        processors.emplace(std::move(null_source));
    }
    processors.emplace(std::move(output));
    connect(*streams.front(), main);
    connect(*totals_having_port, totals);