aglinxinyuan commented on code in PR #6971:
URL: https://github.com/apache/texera/pull/6971#discussion_r3671362000
##########
amber/src/main/python/core/runnables/main_loop.py:
##########
@@ -97,21 +99,40 @@ def __init__(
target=self.data_processor.run, daemon=True,
name="data_processor_thread"
).start()
- def _jump_to_loop_start(
- self, executor: LoopEndOperator, coordinator_interface
- ) -> None:
- # The write address is setup config, keyed by the captured id. Fail
- # loud BEFORE the jump RPC so a misconfigured loop does not rewind the
- # schedule without a back-edge write. Anything raised here (a missing
- # URI, or a failed state write after the jump) is reported by
- # complete()'s guard as an operator-facing error.
- uri = self.context.loop_start_state_uris.get(self._loop_start_id)
+ def _loop_start_base_uri(self) -> str:
+ # The loop's bookkeeping base URI is setup config, keyed by the
+ # captured id (see InitializeExecutorRequest.loopStartPortUris). Fail
+ # loud on a missing entry: anything raised here is reported by the
+ # caller's guard as an operator-facing error.
+ uri = self.context.loop_start_port_uris.get(self._loop_start_id)
if not uri:
raise RuntimeError(
- f"no loop-back state URI configured for LoopStart "
+ f"no loop bookkeeping URI configured for LoopStart "
f"'{self._loop_start_id}' "
- f"(have: {sorted(self.context.loop_start_state_uris)})"
+ f"(have: {sorted(self.context.loop_start_port_uris)})"
)
+ return uri
+
+ def _read_loop_input_table(self) -> Table:
+ # The loop's input table is the Loop Start's input-port
+ # materialization -- data that already exists for the whole loop
+ # (Loop Start re-reads it every iteration; the back-edge truncates
+ # only the state doc at the same base URI, never the result doc).
+ # Reading it here at consume time means the table never has to ride
+ # inside the State content through the loop body.
+ result_uri = VFSURIFactory.result_uri(self._loop_start_base_uri())
+ document, _ = DocumentFactory.open_document(result_uri)
Review Comment:
Experiment result: the overlap was the trigger, and moving the read out of
it fixes the failure.
I changed exactly one thing — **when** the read happens. The matching state
is now stashed at consume and the operator's update runs at EndChannel, by
which point this worker's own materialization reader has finished streaming.
Same read target, same semantics (the matching consume emits nothing
downstream).
| | before | after |
|---|---|---|
| `amber-integration` ubuntu | red — 20m cancel, ~56 `Access Denied` |
**green in 9m** |
| `amber-integration` macos | red — 20m cancel, `Access Denied` | **0
`Access Denied`**, suite completes in 10m49s |
So the `Access Denied` signature is gone on both OSes: it only appeared
while a second iceberg/S3 read was issued from the main loop thread
concurrently with the reader thread. (I can't fully separate "the recreate
lands mid-stream because timing shifted" from "the shared pyiceberg/s3fs client
gets disturbed by concurrent use" — both require the overlap, and removing it
removes the failure.)
macos is still red, but for an unrelated and pre-existing reason: `Received
EndWorker before all 1 queued message(s)` → `worker still has unprocessed
messages`, plus iceberg `CatalogCommitConflicts` retries. No `Access Denied`,
no assertion failures. That is the termination race #6960 is fixing; it also
hits sibling PRs and main. I have re-run that job.
Worth noting the recreate-vs-active-reader hazard is real regardless of this
PR: region re-execution drops and recreates a doc while a downstream reader may
still be lazily iterating a pinned snapshot of it (`IcebergDocument.get()`
fetches data files during iteration, and MinIO answers a deleted key with 403).
This PR no longer trips it, but I can file that separately if you'd like it
tracked.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
