refactor(planner): strip in-process planner band-aids; rely on natural plan shape

[mdashti]

Summary

Stacked on #4. Strips the in-process planner band-aids that were added to make embedders' single-leader topology work: blanket consumer_tc=1 forcing on Shuffle, Coalesce arm elision, nested input_tc=1 forcing, the in_process and has_boundary_ancestor parameters threaded through _distribute_plan.

After this PR, in-process mode means exactly one thing to the planner: prepare_plan skips the gRPC dialer (the only legitimate in-process concern, preserved in src/execution_plans/distributed.rs). Plan shape itself falls out of the annotator's natural task_count math, same as the gRPC path. Peer-mesh emerges from task_count > 1 at a Shuffle boundary without a flag; gather emerges from the outermost stage naturally getting task_count = 1 via cardinality reduction.

Why this is the right design

Stu flagged on #4 that mixing planner-shape concerns into the transport-plumbing PR confused the review and the underlying problem should be resolved differently. Two pieces of the answer:

1. emit_peer_shuffles shouldn't be a flag — it's plan-shape, and the annotator already knows how to produce that shape. The flag exists only because the in-process planner over-customizes by forcing consumer_tc=1 everywhere; once you stop forcing, the shape is what you want.
2. Peer-mesh can't compose at the transport layer — a parallel group-by needs hash-partitioned input flowing into AggregateExec(FinalPartitioned). The cut in the plan tree where data exits one worker and enters another is the planner's call; the transport carries bytes between cuts but doesn't materialize them. Multiplexing across one shared substrate is firmly a transport concern (WorkerConnection already handles it); fan-out topology is plan-shape.

This PR makes the in-process planner say nothing it doesn't have to say.

What changes

Net -83 LOC in src/distributed_planner/distribute_plan.rs:

• Drop in_process and has_boundary_ancestor parameters of _distribute_plan. The function now reads the annotated plan and emits boundaries with the annotator's task_count directly — no in-process branching.
• Drop the nested_in_process variable and the this_is_real_boundary Coalesce special case.
• Shuffle arm: consumer_task_count = task_count, input_task_count = max_child_task_count.unwrap_or(1). Same as the non-in-process path.
• Coalesce arm: emits NetworkCoalesceExec(task_count, max_child_task_count) normally; elides only when both are 1.
• Broadcast arm: emits NetworkBroadcastExec(task_count, max_child_task_count) normally; same elision rule.
• The single planner test test_in_process_single_boundary_default still passes — the natural shape for that query happens to produce exactly one NetworkShuffleExec, same as the band-aided shape, because cardinality factors collapse Stage 2 to task_count=1 on small data.

Embedder impact

Embedders that relied on the band-aided plan shape (e.g. pg_search foundation) get the natural multi-stage plan after this. They need to:

• Walk the produced plan for all NetworkBoundary nodes, not just the outermost.
• Allocate transport slots per (stage_id, task_index, partition) matching each boundary's actual output_partitioning().partition_count(), not a fixed N.
• Host fragments from multiple stages per worker process.
• Handle NetworkCoalesceExec as the outermost worker→leader gather, not NetworkShuffleExec(consumer_tc=1, input_tc=N).

The pg_search natural-shape adoption is being prepared as a separate PR in that repo.

Test plan

• cargo check --workspace clean
• cargo clippy -p datafusion-distributed --all-targets -- -D warnings clean
• cargo test --lib distribute_plan — 24/24 pass
• cargo test --lib — 191/191 pass (1 ignored)
• cargo fmt --all -- --check clean
• Review after feat: enable in-process distributed runtime (WorkerTransport traits + unaddressed Stage + outer/nested boundary arithmetic) #4 lands