aglinxinyuan commented on code in PR #6971:
URL: https://github.com/apache/texera/pull/6971#discussion_r3671186331
##########
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.
--
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]
