Add forced-spill broadcast NestedLoopJoin regression coverage

[sesteves]

Problem

datafusion-distributed inserts network broadcasts beneath eligible NestedLoopJoinExec plans. DataFusion memory-limited NLJ execution currently evaluates the left child once during the initial collection attempt and again when spill fallback starts.

Distributed broadcast streams use exactly-once consumption:

• BroadcastExec allocates a fixed set of consumer streams and removes one for each execute(partition) call. An additional call can return Too many consumers for real partition N.
• NetworkBroadcastExec worker streams are stored as Option<Stream> and taken on first use. An additional call can return WorkerConnections stream for partition N was already consumed.

Apache DataFusion is tracking removal of the redundant left-side evaluation in apache/datafusion#24661. Once that change is adopted, datafusion-distributed should have regression coverage proving that broadcast NLJs remain compatible with memory-limited execution.

This issue does not propose making the distributed broadcast protocol generally replayable.

Proposed work

1. Add a forced-spill integration test with a broadcast build beneath NestedLoopJoinExec.
2. Constrain the memory pool sufficiently to activate DataFusion memory-limited fallback.
3. Verify the build child is evaluated once after adopting the DataFusion fix.
4. Assert correct query results and confirm neither consumer-exhaustion error is produced.
5. Cover remote and colocated worker execution paths.
6. If releases must support DataFusion versions without the fix, add a selective compatibility guard or configuration option that prevents NLJ broadcast insertion when memory-limited fallback can activate. Do not disable HashJoin or CrossJoin broadcasts.

Suggested test shape

• Use an inner non-equi join that plans as NestedLoopJoinExec.
• Ensure the left/build input is broadcast to multiple consumer tasks.
• Use a counting input or equivalent instrumentation to observe build execution count.
• Configure temporary spill storage and a small bounded memory pool.
• Assert spill occurs and the result matches an unconstrained execution.
• Run through both remote-worker and colocated-worker channels.

The regression should fail against the affected DataFusion behavior and pass after the dependency containing apache/datafusion#24661 is adopted.

Acceptance criteria

• Forced-spill broadcast NLJ coverage exists.
• The broadcast build is evaluated once.
• Results are correct.
• No Too many consumers for real partition or already consumed error occurs.
• Remote and colocated paths are covered.
• Existing HashJoin and CrossJoin broadcast tests remain unchanged and passing.
• If older DataFusion versions remain supported, incompatible NLJ broadcast plans are selectively prevented or documented.

References

• DataFusion follow-up: NestedLoopJoinExec spill fallback evaluates the left input twice apache/datafusion#24661
• DataFusion independent stream semantics: fix: LazyMemoryExec should produce independent streams per execute() apache/datafusion#21565
• NLJ and CrossJoin broadcast support: Network broadcast for CrossJoinExec and NestedLoopJoinExec #459 / Add broadcast support for cross and nested loop joins #484