SyncExecutionResource's result-readiness wait is redundant: the engine commits results before COMPLETED (not a sleep-vs-poll problem)

[bobbai00]

Task Summary

Remove the redundant result-readiness wait in SyncExecutionResource.executeWorkflowSync and rely on the engine's existing commit-before-COMPLETED guarantee instead.

The core issue is not sleep(500) versus a poll versus any other waiting mechanism — it is that the wait itself is redundant. Today the method waits before reading results from storage via two unconditional Thread.sleep(500) calls; PR #5714 proposes replacing them with a bounded poll that compares the in-memory output-stats count against the count committed to result storage. But the engine already guarantees that an operator's result storage is durably committed before that operator is observable as COMPLETED, and both result-reading paths in SyncExecutionResource are gated on COMPLETED. So by the time we read, the data is guaranteed present — no sleep, no poll, and no stats-vs-storage count comparison is needed. Swapping sleep→poll just trades one heuristic for another.

Why the wait is redundant. The result writer is committed synchronously on the worker's DP thread before the worker reports the port/operator complete:

1. OutputManager.finalizeOutput() queues FinalizePort(port, input=false) for each output port before FinalizeExecutor (amber/.../messaginglayer/OutputManager.scala:278-284).
2. Processing FinalizePort(port, false) calls OutputManager.closeOutputStorageWriterIfNeeded(port), which sends a terminate signal and then blocks on writerThread.join() (OutputManager.scala:257-264). The writer thread's finally runs IcebergTableWriter.close() → flushBuffer() → table.newAppend().appendFile(...).commit() (common/workflow-core/.../iceberg/IcebergTableWriter.scala:106-135,140-144). After join() returns, every row is committed. getFailure.foreach(throw _) re-throws a failed commit, so a commit failure becomes a FatalError (→ FAILED/KILLED), never COMPLETED.
3. Only then does the worker send portCompleted (DataProcessor.scala:174-181).
4. FinalizeExecutor afterward transitions the worker to COMPLETED and sends workerExecutionCompleted (DataProcessor.scala:159-173).

So commit happens-before the port/worker is reported COMPLETED. Since the read side (IcebergDocument.getCount/get()) reloads fresh catalog metadata on each call, a read taken after observing COMPLETED always sees the full result — including across the decoupled compute-unit process boundary. Both reading paths are already gated on COMPLETED: TerminalStateReached(COMPLETED) and TargetResultsReady (fires on allTargetsCompleted(stats)). The old "RegionExecutionCoordinator caches upstream results asynchronously after operators complete" comment that motivated the sleep no longer corresponds to any code; the synchronous writerThread.join() barrier closed that gap.

Proposed resolution.

• Remove the wait from SyncExecutionResource (rather than replace it with a poll) and document the invariant it relies on.
• Lock the invariant with engine-side tests, since removing the safety net turns a regression from "a slow response" into "a silent short read":
  • The commit + failure-propagation half is already covered by OutputPortStorageWriterThreadSpec.
  • Add an e2e spec that reads result storage the instant the workflow reports COMPLETED (no wait) and asserts the committed row count (getCount) equals the rows actually readable, across a multi-region DAG including an intermediate operator.

References. PR #5714 (the sleep→poll approach this issue argues against as the framing); #5713 (target of #5714). Key engine seam: OutputManager.closeOutputStorageWriterIfNeeded (writerThread.join()) → DataProcessor.outputOneTuple (portCompleted / COMPLETED).

Filed with assistance from Claude (Opus 4.8) in compliance with the ASF generative-AI tooling policy.

Task Type

• Refactor / Cleanup
• DevOps / Deployment / CI
• Testing / QA
• Documentation
• Performance
• Other