Make insert_broadcast_execs idempotent for existing broadcast inputs

[sesteves]

Problem

insert_broadcast_execs always inserts a new BroadcastExec beneath an eligible join, even when the build input already has the canonical broadcast shape:

CoalescePartitionsExec
  BroadcastExec
    input

For a fetch-less CoalescePartitionsExec, the pass unwraps the coalesce and uses its input as the payload for a new broadcast. If that input is already a BroadcastExec, the resulting plan becomes:

CoalescePartitionsExec
  BroadcastExec
    BroadcastExec
      input

Applying the pass repeatedly adds another broadcast layer each time.

Impact

The nested broadcasts preserve query results, but they create unnecessary cache state and duplicate memory reservations for the same batches. This increases memory pressure and can make spilling or resource-exhaustion failures more likely.

It also means the optimizer pass is not idempotent, which is surprising when plans have already been processed or when an earlier custom optimizer has inserted the canonical broadcast shape.

Proposed behavior

Before inserting a broadcast, inspect the build input:

• If it is already CoalescePartitionsExec(BroadcastExec(...)) with no fetch, leave it unchanged.
• If the build is directly a BroadcastExec, add only the required coalesce wrapper rather than another broadcast layer, if that shape is supported.
• Preserve the existing handling for fetch-bearing CoalescePartitionsExec, where the fetch must remain below the broadcast.

The exact normalization can vary, but a second application of insert_broadcast_execs should produce the same plan as the first.

Testing

Add focused tests for:

• HashJoinExec with an existing canonical broadcast build.
• NestedLoopJoinExec with an existing canonical broadcast build.
• CrossJoinExec with an existing canonical broadcast build.
• A direct BroadcastExec build input, if supported.
• A fetch-bearing coalesce to ensure its limit remains below the broadcast.
• Applying insert_broadcast_execs twice and asserting that the second result is structurally unchanged.

Plan assertions should verify that every eligible build contains exactly one BroadcastExec layer.

Acceptance criteria

• Existing canonical broadcast build inputs are not wrapped again.
• Repeated optimizer application is idempotent.
• HashJoin, NestedLoopJoin, and CrossJoin are covered.
• Fetch-bearing coalesce semantics are preserved.
• Existing broadcast planning and correctness tests continue to pass.

References

• Broadcast insertion: src/distributed_planner/insert_broadcast.rs
• NLJ and CrossJoin broadcast support: Network broadcast for CrossJoinExec and NestedLoopJoinExec #459 / Add broadcast support for cross and nested loop joins #484

[WilliamK112]

I'd like to take this. I verified that current main still unwraps a fetch-less coalesce and then inserts a new BroadcastExec even when that coalesce already contains the canonical broadcast shape, and I found no open PR for this issue.

I plan to make the build-side normalization preserve an existing canonical coalesce-plus-broadcast, wrap a direct BroadcastExec with only the required coalesce, and leave the fetch-bearing coalesce behavior unchanged. Regression coverage will exercise repeated application for HashJoinExec, NestedLoopJoinExec, and CrossJoinExec, plus the direct-broadcast normalization; the existing fetch-bearing test will continue to pin the limit below the broadcast.

I will run the focused planner tests first, then formatting, Clippy, and the relevant full Rust suite.

AI assistance disclosure: I am using OpenAI Codex to help inspect, implement, and validate the change, and I will review the final diff and evidence.