1. Protocol (src/Core/Protocol.h, ProtocolDefines.h, MergeTree/RequestResponse.{h,cpp})

  - New client packet MergeTreeAllRangesAnnouncementResponse (= 14), protocol bumped to DBMS_PARALLEL_REPLICAS_PROTOCOL_VERSION = 8 with gate
  DBMS_PARALLEL_REPLICAS_MIN_VERSION_WITH_ANNOUNCEMENT_RESPONSE.
  - New struct InitialAllRangesAnnouncementResponse { parts, stream_id }: initiator's reply to a follower's announcement, carrying the authoritative parts list for that stream.
  - MergeTreeAllRangesCallback signature changes from void(...) → std::optional<...Response>(...). nullopt = old initiator, no response sent. Engaged-but-empty parts = stream
  doesn't exist on the coordinator (over-announce) → follower's pool finishes immediately.

  2. Coordinator (ParallelReplicasReadingCoordinator.{h,cpp})

  - handleInitialAllRangesAnnouncement now returns InitialAllRangesAnnouncementResponse.
  - New setSnapshotReplicaNum(replica_num): lets the initiator pin itself as the snapshot replica BEFORE any announcement arrives. Once pinned, only the snapshot replica can
  create a new stream coordinator; announcements from other replicas for unknown streams are silently dropped (empty parts returned).
  - Per-stream getRegisteredParts() virtual on ImplInterface: the post-normalization working set (InOrderCoordinator drops covered/covering parts at normalization time). Cached
  in stream_to_registered_parts on first announcement and echoed back to every subsequent announcer.
  - InOrderCoordinator::isReadingCompleted() newly implemented; pure virtual on the base.

  3. Planner (ReadFromMergeTree.{h,cpp}, ParallelReplicasLocalPlan.cpp, ClusterProxy/executeQuery.{h,cpp})

  - spreadMarkRangesAmongStreamsWithOrder refactored into three branches gated by isParallelReplicasLocalPlanForInitiator / isParallelReplicasLocalPlanForFollower:
    - Initiator + local plan: split ranges into num_streams per-stream pools (genuine splitting). Each split gets its own stream_id = <table>#split_{i}.
    - Follower + local plan: construct num_streams pools, each over ALL local parts. Each pool's announcement gets back from the coordinator the authoritative sub-set; non-owned
  parts are filtered out during source construction so no phantom consumers are built. Streams owning no parts on this follower produce empty pipes and are dropped.
    - parallel_replicas_local_plan=0: preserved old single-pool behavior.
  - make_per_split_pool_settings: divides pool_settings.threads evenly across splits to avoid min_marks_per_request = min_marks_per_task × threads being inflated num_splits-fold.
  - ParallelReplicasLocalPlan::createLocalPlanForParallelReplicas calls coordinator->setSnapshotReplicaNum(replica_number) before any announcement is sent.
  - New helper ClusterProxy::canUseLocalPlanForParallelReplicas(context): extracts the 4-clause check that gates the local-plan branch in executeQueryWithParallelReplicas
  (analyzer + parallel_replicas_local_plan + parallel_replicas_prefer_local_replica + _shard_num == 0). Now also called by ReadFromMergeTree so followers keep lock-step with the
  initiator's topology decision. This is the fix in the last two commits — without it, followers inside a Distributed sub-query took the 32-pool over-announce path while the
  initiator skipped local plan entirely, causing every split's coordinator to register the full part view and amplify reads ~`num_streams`×.

  4. Read pools (MergeTreeReadPoolParallelReplicas{,InOrder}.{h,cpp}, MergeTreeReadPoolBase.{h,cpp})

  - MergeTreeReadPoolBase::buildAnnouncementDescriptions(): lifted from the InOrder pool to be shared (fills in per-part min_marks_per_task). The announcement is now sent by
  ReadFromMergeTree directly (not the pool constructor), so the response can be consumed at the call site.
  - MergeTreeReadPoolParallelReplicasInOrder:
    - Steady-state task size now capped at min_marks_per_task (matching the Default pool) rather than max_block_size / index_granularity — fixes a long-standing over-small task
  issue while keeping warmup growth for early-LIMIT termination.
    - Response matching is now keyed by (part_info, projection_name) instead of by position, since coordinators can return parts in arbitrary order.
  - ParallelReadingExtension::sendInitialRequest returns the response. Split into sendReadRequest (Default) / sendReadInOrderRequest (positional → keyed).

  5. Connection plumbing

  - IServerConnection / IConnections / Connection / MultiplexedConnections / LocalConnection / HedgedConnections: new sendMergeTreeAllRangesAnnouncementResponse virtual + impl.
  - TCPHandler: the announcement callback now blocks for the response (when client protocol ≥ 8) via new receiveAllRangesAnnouncementResponse. Errors set stop_query = true and
  rethrow.
  - RemoteQueryExecutor::processMergeTreeInitialReadAnnouncement: relays the coordinator's response back to the announcing replica.

  6. Tests

  - New 04073_parallel_replicas_in_order_splits.sql: verifies that a single part is genuinely split into ≥ max_threads MergeTreeSelect sources for both WithOrder and
  ReverseOrder.
  - New integration test test_parallel_replicas_protocol (separate fixture under tests/integration/).
  - Modified 00177_memory_bound_merging.sh (test4): runs the in-order query with count() (which amplifies under work duplication, unlike the idempotent max(URL) in test1/test2)
  and compares against a single-node baseline. With max_threads=16 and parallel_replicas_local_plan=1 pinned so random-settings can't mask the regression. This catches the
  over-announce bug that the last two commits fix.
  - Reference updates in 02404_memory_bound_merging, 02883_parallel_replicas_join_algo_and_analyzer_{2,3}, 03222_parallel_replicas_memory_bound_merging_projection,
  03724_parallel_replicas_duplicate_requests for the new split topology.

  ---
  Suggested review order

  1. Protocol + coordinator (sections 1–2) — sets the contract.
  2. Planner (section 3) — most of the new code; the three-branch split in spreadMarkRangesAmongStreamsWithOrder plus the snapshot pinning are the heart of the change.
  2a. The last two commits in isolation (0d3f0456c06, a92a094ca72) — small, self-contained canUseLocalPlanForParallelReplicas factoring + regression test.
  3. Read pool + connection plumbing (sections 4–5).
  4. Tests (section 6) — 04073 is the cleanest functional spec of the new behavior.