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.
--
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]
