aglinxinyuan commented on code in PR #6971:
URL: https://github.com/apache/texera/pull/6971#discussion_r3671132052
##########
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:
You're right, it's this PR — not environment noise. I dug into both job logs
and the root cause is architectural, not a small bug:
The `Access Denied` is MinIO's 403-for-missing-key on a GetObject for a
parquet data file under the **LoopStart's output** doc — the doc the LoopEnd's
own materialization reader is streaming. Two invariants make my consume-time
read unsafe there:
1. `IcebergDocument.get()` returns a **lazy** iterator: it pins the snapshot
up front and fetches data files as iteration proceeds
(`_get_using_file_sequence_order` → `IcebergIterator`).
2. LoopStart's output port is `reuseStorage = false`, so every iteration
**drops and recreates** that doc — deleting the parquet files a still-active
iterator has in its pinned snapshot.
My change adds a full drain of an iceberg table on the **main loop thread**
at consume time, concurrent with that reader thread, which widens the window
where the recreate lands mid-stream.
And the deeper problem, which is really why the table was embedded in the
state to begin with: **states are replayed before tuples** (see the ordering
note in `input_port_materialization_reader_runnable.run`), so at consume time
the LoopEnd genuinely does not have its input table by any local means. Reading
it out of storage instead is exactly what races with the doc's per-iteration
lifecycle.
So the premise of the refactor doesn't hold as written. I'm converting this
to draft rather than patching around it — a safe version needs either a stable
per-loop input doc (related to the back-edge doc discussion) or deferring
update/condition to EndChannel once the buffered table is complete, both of
which are bigger than this PR. Thanks for catching it before it went further.
##########
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:
Correction to my previous reply — I got the root cause wrong, and the two
invariants I cited do not apply here.
I went back through the logs and counted where the failures actually are:
| | |
|---|---|
| `Access Denied` blocks whose traceback contains my new read
(`_read_loop_input_table`) | **0** |
| `Access Denied` blocks inside `input_port_materialization_reader_runnable`
| 45 of 56 |
| Docs that failed | `LoopStart` / `Limit` **output** docs — loop-internal,
recreated on each re-execution |
| Failures on the doc my read actually opens (LoopStart's input port = the
upstream op's output, e.g. TextInput) | **none** |
So my read is not the thing failing, and it is not reading a volatile doc:
the operator immediately before LoopStart is materialized once before the loop
and its result doc is stable for the whole run — it is the same base URI whose
*state* sub-doc the back-edge already rewrites every iteration. By the same
token "states replay before tuples" is not an obstacle to this design; reading
the table from that materialization is exactly the way around it, and the data
is there at consume time.
What the evidence does show is that the failures are in **reader threads
streaming loop-internal docs that get dropped and recreated on re-execution**,
while `IcebergDocument.get()` holds a lazily-consumed pinned snapshot. That
race is pre-existing; what my PR adds is a full table read on the main loop
thread at consume time, which either shifts the timing so the recreate now
lands mid-stream, or disturbs the shared pyiceberg/s3fs client the readers use.
I have not yet distinguished those two, so I am not going to claim which.
Leaving this in draft while I pin that down — but the design premise stands,
contrary to what I said above. Sorry for the noise.
##########
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:
Thanks — that was a real problem on the commit you looked at, and the
diagnosis was right: the loop table read was racing this worker's own
materialization reader. Reading a second iceberg/S3 doc from that thread while
the reader iterates a lazily-pinned snapshot that region re-execution drops and
recreates makes the reader fail with S3 `Access Denied` (MinIO's answer for a
deleted key).
Fixed in `822078ca7f` — the read is deferred to the EndChannel path, where
this worker's reader has already finished, so the two can never overlap. That
landed about half an hour after your comment, so the run you saw predates it.
On the current head:
| leg | before | now |
|---|---|---|
| amber-integration (ubuntu) | cancelled at 20m | **passed, 9m04s** |
| amber-integration (macos) | cancelled at 20m | fails: 40 succeeded, 1
failed |
`Access Denied` no longer appears anywhere in either log (0 occurrences).
The remaining macos failure looks like a different, already-tracked issue
rather than this refactor: it fails with `RuntimeError: worker still has
unprocessed messages` followed by `Failed to terminate region N on attempt 1 of
150` — i.e. #6891 (*Flaky loop e2e tests: EndWorker fails on in-flight RPC acks
during region termination*), whose fix is in flight as #6960.
I'm leaving this thread open rather than resolving it, since macos is still
red — happy to rebase on #6960 once it lands to confirm, if that's the cleanest
way to show it green.
##########
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]
