apache / apache/texera

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

Open
#5,855 0 comments 0 reactions 1 assignee Claimed by @bobbai00 View on GitHub
engine proposal
Dominant language
Scala
Stars
314
Forks
187
Avg merge
1d 21h
Merged PRs (30d)
214

Description

### 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

- [X] Refactor / Cleanup
- [ ] DevOps / Deployment / CI
- [X] Testing / QA
- [ ] Documentation
- [X] Performance
- [ ] Other

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.