Build full-fledged plan for local replica

[nickitat]

as we do for distributed queries. i.e. instead of

   ┌─explain────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┐
1. │ Expression (Project names)                                                                                                                                                                                                                                 │
2. │   ReadFromRemoteParallelReplicas (Replica: 127.0.0.1, Replica: 127.0.0.2, Replica: 127.0.0.3, Replica: 127.0.0.4, Replica: 127.0.0.5, Replica: 127.0.0.6, Replica: 127.0.0.7, Replica: 127.0.0.8, Replica: 127.0.0.9, Replica: 127.0.0.10, Replica: 127.0.0.11) │
   └────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘

build something like this:

Expression ((Project names + Projection))
  MergingAggregated
    Union
      Aggregating
        Expression ((Before GROUP BY + Change column names to column identifiers))
          ReadFromSystemNumbers
      ReadFromRemote (Read from remote replica)

this will enable any analysis/optimisation work based on query plan on the initiator

[nikitamikhaylov]

For me it looks similar to prefer_localhost_replica=1 for Distributed table. I tried to support this previously and failed, because it turned out to be difficult to balance properly between executing the local part of the pipeline and constantly answer for the incoming requests from remote replicas (from the scheduling standpoint). Pipeline executor has lot of affinities and there is no fairness there, so from what I observed ClickHouse either prefers executing local plan only with all the threads it has or prefers answering the incoming requests.

[nickitat]

interesting, thx for sharing

[devcrafter]

For me it looks similar to prefer_localhost_replica=1 for Distributed table.

There is an important difference to a distributed table, - in the case of task-based parallel replicas, the data to read is assigned dynamically across replicas, where for distributed table, replicas read different data according to sharding schema (so there is no dynamic scheduling among replicas).