What I'm wondering before looking into code... is that distributed_group_by_no_merge just sets query processing stage to Complete instead of WithMergeableState and it works not only for GROUP BY queries, but will also prevent merging of streams for queries with ORDER BY and without GROUP BY.

Let's check that.

prevent merging of streams for queries with ORDER BY

Indeed, but there is if (select.orderBy()) return false and test

What if there are other operations that require merging, e.g.
LIMIT - calculates preliminary limit on shards, apply final LIMIT on initiator;
DISTINCT - the same...
?

Ok. The code becomes complicated and bug-prone but still acceptable.

Maybe we can solve the task by introducing another QueryProcessingStage: WithMergeableStateAfterAggregation (it means that aggregate functions were calculated and finalized) and process it inside InterpreterSelectQuery?

(we can experiment in a separate PR)

Maybe we can solve the task by introducing another QueryProcessingStage: WithMergeableStateAfterAggregation (it means that aggregate functions were calculated and finalized) and process it inside InterpreterSelectQuery?

Yeah, I was just "experimenting"
Another stage is a great idea, will take a look (even distinct by shading key is ok actually)

Maybe we can solve the task by introducing another QueryProcessingStage: WithMergeableStateAfterAggregation (it means that aggregate functions were calculated and finalized) and process it inside InterpreterSelectQuery?

Decided to address this separately (to avoid polluting this patch set, since I guess there will be some corner cases that I will forget about)

Ok.

Maybe we can solve the task by introducing another QueryProcessingStage: WithMergeableStateAfterAggregation (it means that aggregate functions were calculated and finalized) and process it inside InterpreterSelectQuery?

@alexey-milovidov #10373 implements this