[SPARK-57638][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)

[uros-b]

Nit: the else if (outstanding > 0) guard is effectively always true when reached. Control only reaches it when the if is false, i.e. outstanding >= batchSize || queue.isEmpty. Since batchSize >= 1, the first disjunct implies outstanding > 0; and if the queue is empty the loop invariant guarantees outstanding > 0. So it could just be a plain else?

[LuciferYang]

Good question, but the guard is needed - a plain else would deadlock the last iteration. The reasoning misses that the poll() drain loop runs after the loop-condition check and mutates outstanding. The invariant outstanding > 0 || queue.nonEmpty only holds at the top of the loop; by the time we reach this branch, the drain may have reaped the last in-flight tasks and taken outstanding to 0 while the queue is already empty. In that case the if is false (empty queue) and there is nothing left to wait for - take() would block forever. The outstanding > 0 guard lets the loop fall through so the next condition check (0 > 0 || empty) exits cleanly. I added a comment spelling this out.

[yadavay-amzn]

LGTM, non-blocking suggestion on test coverage.

Clean fix : the ExecutorCompletionService poll()/take() structure removes the spin correctly, the submitted task body is unchanged so behavior reduces to the loop control, and the outstanding > 0 guard on the take() branch (nicely explained in the followup commit) closes the one real deadlock trap: a poll() drain that takes outstanding to 0 with an empty queue must exit, not block in take() forever.

Suggestion (non-blocking): a deterministic test for the >parallelism regime. The PR notes a test is omitted because asserting the absence of a busy-wait needs flaky CPU/timing measurement -- that's fair for the perf property. But there's a separate, non-timing property that this rewrite changes and that no existing suite covers: correct resolution when the number of flows exceeds parallelism (10), which is exactly the regime where slots fill and the new take()-blocking path is taken. The existing suites resolve small graphs (the 10s in them are RANGE(10) row counts, not flow counts), so the blocking branch is never exercised deterministically. A graph with, say, 20+ independent flows asserting they all resolve (and the call terminates) would lock in the equivalence on the path the rewrite actually alters, with no timing dependence and no flakiness. Optional, but it covers the riskier half of a concurrency-control change.

What I verified:

• Behavior equivalence: the completionService.submit(...) task body is identical to the old executor.submit(...) body (retry / leader-election / destination logic unchanged); only the scheduling loop changed. Exceptions still propagate via reap -> Future.get() -> throw exn.getCause, matching the old done.foreach(_.get()).
• Termination: loop condition outstanding > 0 || toBeResolvedFlows.peekFirst() != null plus the outstanding > 0 guard on take() -- no path blocks with nothing to wait for, and the retry addFirst re-queue still re-drives the loop.
• Ordering/scheduling: still pollFirst() in submit order, still capped at batchSize; outstanding tracks slots equivalently to the old futures.size.
• Empty-graph edge: the loop condition is false immediately, so take() is never called.

[LuciferYang]

Thanks for the careful review @yadavay-amzn, and good catch on the coverage gap — the >parallelism regime is exactly where the rewritten take() path runs and the existing suites never reach it.

Added two deterministic tests in ConnectValidPipelineSuite:

• 25 independent flows — more flows than parallelism (10), so slots fill and the loop blocks on take(); asserts every flow resolves and the call returns.
• 20 consumers registered before their source view — the consumers are scheduled first, so the first batch hits an unresolved src, each throws TransformNodeRetryableException and is parked as a dependent of src; once src resolves they're re-queued and retried. This deterministically exercises the retry re-queue path together with the blocking branch.

[szehon-ho]

@anew @jose-torres do you want to take a look as well?

[dongjoon-hyun]

+1, LGTM. BTW, shall we remove the following sentence, A dedicated test is not included ..., because this PR added two new test cases, @LuciferYang ?

How was this patch tested?

A dedicated test is not included because asserting the absence of a busy-wait reliably requires CPU-time or timing measurements that are flaky in CI.

[LuciferYang]

+1, LGTM. BTW, shall we remove the following sentence, A dedicated test is not included ..., because this PR added two new test cases, @LuciferYang ?

How was this patch tested?

A dedicated test is not included because asserting the absence of a busy-wait reliably requires CPU-time or timing measurements that are flaky in CI.

Thank you @dongjoon-hyun . I have updated the How was this patch tested? section to describe the new tests.

[LuciferYang]

Merge Summary:

• merged into master 0747e28
• merged into branch-4.x 3401f3d
• merged into branch-4.2 b7c9225

Posted by merge_spark_pr.py

[LuciferYang]

Thanks @dongjoon-hyun @uros-b @yadavay-amzn @szehon-ho