[SPARK-57638][4.1][SQL] Avoid busy-waiting in Declarative Pipelines flow resolution

[LuciferYang]

What changes were proposed in this pull request?

DataflowGraphTransformer.transformDownNodes resolves flows on a bounded thread pool and drives them from a while loop that, each pass, partitioned the in-flight futures with the non-blocking future.isDone, reaped the completed ones, and scheduled a new flow if a slot was free. When all slots were in flight (or the queue was drained and only the last futures remained) and none had completed, the pass reaped nothing and scheduled nothing, then looped again immediately - busy-spinning on isDone and pinning a core for the duration of resolution.

This drives the loop with an ExecutorCompletionService instead: completed tasks are drained with the non-blocking poll(), and when nothing can be scheduled but tasks are still running, the loop blocks on take() until the next one finishes rather than spinning. Behavior is otherwise unchanged - the same flows are scheduled in the same order, exceptions are still propagated via Future.get(), and an outstanding counter replaces the ArrayBuffer[Future] for slot bookkeeping.

Why are the changes needed?

Resolving a graph with more flows than the parallelism (10) kept one CPU core busy at 100% doing no useful work for the whole resolution, which is wasteful and shows up as unexplained driver CPU.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Two new cases in ConnectValidPipelineSuite cover the regime this PR changes - more flows than parallelism (10), so the slots fill and the loop reaches the blocking take() branch that replaces the busy-wait. The small graphs in the existing suites never get there.

• resolution terminates and resolves all flows when flow count exceeds parallelism - 25 independent flows.
• resolution re-queues retryable flows under load when consumers exceed parallelism - 20 consumers registered before their source src, so the first batch throws TransformNodeRetryableException, parks as dependents of src, and is re-queued once src resolves; this exercises the retryable re-queue path together with the blocking branch.

Both assert only the outcome (every flow resolves and the call returns), so they are deterministic and have no timing dependence - a regression that deadlocked would hang until the suite times out. Asserting the absence of a busy-wait directly is not included, since that requires CPU-time or timing measurements that are flaky in CI.

Existing graph-resolution suites (ConnectValidPipelineSuite, ConnectInvalidPipelineSuite, SqlPipelineSuite, TriggeredGraphExecutionSuite, MaterializeTablesSuite) still pass; the change only affects how the loop waits, not what it resolves.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Claude Opus 4.8)