Workflow [PR], commit [2b38544]

Summary: ❌

@devcrafter Im planning to spend this weekend working on this hopefully, but if you've suggestion, please let me know

@matt-metivier Here is my consideration with some explanations which can help

Parallel replicas is a mechanism to parallelize query execution for a replicated MergeTree table among cluster nodes. So, the reading from a table will be parallelized among nodes and read data will be process to some mergeable stage on each node. The parallelization is done by reading different ranges of parts by different nodes.
The reading from a table is coordinated by parallel replicas coordinator on initiator node.
The table snapshot (what table parts are considered to be read during query execution) is taken from first announcement request from a replica, - in case parallel_replicas_local_plan = true, it's local replicas, otherwise first replicas we've got the announcement from.

I can image the following ...
Since merge table/merge() table function works like UNION of table which match a particular regular expression, we can treat such query as UNION of queries from the corresponding tables.
On initiator, we can decide which tables participate in the query. For each MergeTree table, we'll need to build an execution plan with parallel replicas and UNION them (also, with others plans for non-MT tables). So, each MT table we'll have separate parallel replicas coordinator. It should be very similar if you'll create a UNION query where table are mentioned explicitly.

Please let me know if you have other considerations or questions.

@KochetovNicolai Please comment if you have something to add

I did not read the current implementation, but I find it difficult to implement parallel replicas for MergeTables with the current infrastructure. I think the best we can do now, as @devcrafter mentioned, is to treat StorageMerge as a UNION and enable parallel replicas for each branch independently (each branch will have its own connection, coordinator, tasks, etc). This way, we could parallelize only the reading. Changes in findParallelReplicasQuery are likely not needed (or, I don't understand why).

Offloading other steps, for example aggregation, will require changes in both planner and protocol (to support multiplexing between many storages).