```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
```
const Block & header, const QueryPipeline::ProcessorGetterWithStreamKind & getter, QueryPipeline::StreamType kind)
{
    return getter(header, kind);
}
template <typename TProcessorGetter>
void QueryPipeline::addSimpleTransformImpl(const TProcessorGetter & getter)
{
    checkInitializedAndNotCompleted();
    Block header;
    auto add_transform = [&](OutputPort *& stream, StreamType stream_type)
    {
        if (!stream)
            return;
        auto transform = callProcessorGetter(stream->getHeader(), getter, stream_type);
        if (transform)
        {
            if (transform->getInputs().size() != 1)
                throw Exception("Processor for query pipeline transform should have single input, "
                                "but " + transform->getName() + " has " +
                                toString(transform->getInputs().size()) + " inputs.", ErrorCodes::LOGICAL_ERROR);
            if (transform->getOutputs().size() != 1)
                throw Exception("Processor for query pipeline transform should have single output, "
                                "but " + transform->getName() + " has " +
                                toString(transform->getOutputs().size()) + " outputs.", ErrorCodes::LOGICAL_ERROR);
        }
        auto & out_header = transform ? transform->getOutputs().front().getHeader()