Xiao-zhen-Liu commented on code in PR #6971:
URL: https://github.com/apache/texera/pull/6971#discussion_r3696864691
##########
amber/src/main/python/core/runnables/main_loop.py:
##########
@@ -97,21 +104,64 @@ 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 -- a doc created once UPSTREAM of the loop, so it is
+ # stable for the whole run (the back-edge rewrites only the state doc
+ # under the same base URI, never this result doc). Reading it means the
+ # table never has to ride inside the State content through the loop
+ # body. Callers must invoke this OUTSIDE the window where this worker's
+ # own materialization reader is streaming -- see
+ # _consume_pending_loop_state.
+ result_uri = VFSURIFactory.result_uri(self._loop_start_base_uri())
+ document, _ = DocumentFactory.open_document(result_uri)
+ return Table(list(document.get()))
+
+ def _consume_pending_loop_state(self, executor: LoopEndOperator) -> None:
+ # Run the matching consume that _process_state_frame deferred.
+ #
+ # The loop's input table is read from the Loop Start's input-port
+ # materialization -- a doc that is created once upstream of the loop
+ # and is stable for the whole run. The read is deferred to here (the
+ # EndChannel path) rather than done at consume time because at consume
+ # time THIS worker's own materialization reader is still streaming its
+ # input; issuing a second iceberg/S3 read from this thread while that
+ # reader iterates a lazily-pinned snapshot of a doc that region
+ # re-execution drops and recreates makes the reader fail with S3
+ # "Access Denied" (MinIO's answer for a deleted key). By EndChannel the
+ # reader has finished, so the two never overlap.
+ if self._pending_loop_state is None:
+ return
+ pending = self._pending_loop_state
+ self._pending_loop_state = None
+ executor.attach_loop_table(self._read_loop_input_table())
+ # A Loop End has exactly one input port (port 0); the generated
+ # operator's process_state ignores the port anyway.
+ executor.process_state(pending, 0)
Review Comment:
The consume used to run inside `DataProcessor._executor_session`. Calling
the executor directly here drops the `replace_print` capture, so a `print()` in
the user's `update` no longer reaches the console -- it goes to the worker's
stdout and disappears. It also skips the debug-command hook.
`condition()` already had this problem, so you're widening an existing gap
rather than opening one, but `update` is where people actually put prints.
Either wrap this in the same print capture, or add a line saying loop
`update`/`condition` deliberately run outside it.
##########
amber/src/main/python/core/runnables/main_loop.py:
##########
@@ -97,21 +104,64 @@ 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 -- a doc created once UPSTREAM of the loop, so it is
Review Comment:
"a doc created once UPSTREAM of the loop, so it is stable for the whole run"
-- true for a single loop, not true for the inner loop of a nested one. The
inner Loop Start's upstream is the outer Loop Start's output port, and that doc
is dropped and recreated on every outer iteration; it's one of the
loop-internal docs the second commit blames for the `Access Denied`.
The PR description gets this right. This comment is stronger than the truth,
and it's the comment someone will trust in a year. "Stable for the duration of
the loop level that reads it" would be accurate.
##########
amber/src/main/python/core/runnables/main_loop.py:
##########
@@ -142,6 +192,7 @@ def complete(self) -> None:
# worker, instead of killing the thread through run()'s
# @logger.catch(reraise=True).
try:
+ self._consume_pending_loop_state(executor)
Review Comment:
This is the one I'd want settled before merge. `complete()` runs at the tail
of `_process_end_channel`, *after* `close_port_storage_writers()` and after
`port_completed` has been sent for every output port (lines 483-493). So when
the read or the user's `update` raises, you report the exception -- but the
coordinator has already been told all output ports are done. That's exactly
what the guard at lines 457-463 exists to prevent, in its own words: region
completion is port-based, so holding the region is what keeps a reported error
from reading as a false success.
The back-edge write already sits in this window, so the shape isn't new --
but a user's `update` expression fails far more often than an iceberg commit
does.
Can the consume move up to right after `process_input_tuple()` (line 464)?
That's still past the reader, but before anything is reported complete, and it
would sit under the existing `has_exception()` hold.
##########
amber/src/main/python/core/runnables/main_loop.py:
##########
@@ -375,9 +426,18 @@ def _process_state_frame(self, frame: StateFrame) -> None:
if isinstance(executor, LoopEndOperator):
# Matching LoopEnd (in_counter == 0): it will consume this state
- # and jump back. Remember which LoopStart to jump to (it rides
- # the envelope) for complete()/_jump_to_loop_start.
+ # and jump back. Remember which LoopStart to jump to (it rides the
+ # envelope) for complete()/_jump_to_loop_start, and STASH the state
+ # -- the operator runs its update at EndChannel instead of here,
+ # because the loop's input table is read from storage and that read
+ # must not overlap this worker's own materialization reader (see
+ # _consume_pending_loop_state). The LoopEnd emits no state
+ # downstream on the matching consume, so deferring it changes
+ # nothing observable outside the operator.
self._loop_start_id = frame.loop_start_id
+ self._pending_loop_state = state
Review Comment:
The single slot assumes at most one matching state per region execution.
That holds today because `LoopOpDesc` sets `withParallelizable(false)`, but
nothing here says so, and a second frame would silently overwrite the first and
skip an update. A one-line assert would make the assumption checkable.
Related: the stash also assumes `complete()` always runs. The
`is_missing_output_ports()` early return at line 481 skips it, and the stashed
state is then dropped and the loop ends quietly. Probably unreachable for a
Loop End, but the deferral is what creates the dependency.
##########
amber/src/main/python/core/models/operator.py:
##########
@@ -452,21 +443,38 @@ def __init__(self):
# AttributeError; a None _loop_table means "nothing consumed yet" and
# condition() short-circuits to False (see eval_condition).
self.state: State = State()
+ # Set by the runtime (attach_loop_table) right before the matching
+ # consume; run_update reads it. Distinct from _loop_table so the
+ # "consumed" marker is still only set by a SUCCESSFUL update.
+ self._attached_table: Optional[Table] = None
self._loop_table: Optional[Table] = None
@overrides.final
def process_table(self, table: Table, port: int) ->
Iterator[Optional[TableLike]]:
yield table
+ @overrides.final
+ def attach_loop_table(self, table: Table) -> None:
+ # Runtime-only hook: MainLoop reads the loop's input table from the
+ # Loop Start's input-port materialization (loopStartPortUris) and
+ # attaches it here right before the matching consume, so the table
+ # never has to ride inside the State content through the loop body.
+ self._attached_table = table
+
@overrides.final
def run_update(self, update_code: str, state: State) -> None:
# Run the user's `update` in a throwaway namespace seeded with the
# incoming loop variables and the input table, then persist the user
- # variables back into self.state. The table arrives as an Arrow IPC
- # stream, not pickle (see `table_to_ipc_bytes` in core.models.table
- # for why); the decoded table is kept on self._loop_table so
- # condition() can read it after the update.
- input_table = table_from_ipc_bytes(state[_TABLE_KEY])
+ # variables back into self.state. The table is attached by the runtime
+ # (attach_loop_table) from the Loop Start's input materialization; on
+ # a successful update it is kept on self._loop_table so condition()
+ # can read it afterwards.
+ if self._attached_table is None:
Review Comment:
`_attached_table` is never cleared, so this guard only bites on the first
iteration. From iteration 2 on it still holds the previous read, and a runtime
path that forgot to attach would run the update against a stale table instead
of raising. Setting it back to `None` after you take it (line 477, or after the
successful update at 491) makes the fail-loud hold every iteration, and drops
the reference when the loop is done.
That also shrinks the reason for having two fields -- right now
`_attached_table` / `_loop_table` exist purely so "consumed" keeps meaning
"successful update", which isn't obvious from the names. Worth a word in the
comment.
##########
amber/src/test/python/core/runnables/test_main_loop.py:
##########
@@ -2469,31 +2476,55 @@ def
test_loopend_consume_invokes_operator_at_counter_zero(
"get_output_state",
lambda: None,
)
+ # Stub the runtime's table read (the real one opens the Loop Start's
+ # input-port materialization; pinned by the jump/read URI tests).
Review Comment:
"pinned by the jump/read URI tests" -- the state URI is pinned
(`test_jump_to_loop_start_sends_rpc_then_writes_state_in_order`), but the read
URI isn't. `_read_loop_input_table` is stubbed in every test, so
`result_uri(base)` and the `open_document` call have no coverage, and that's
the half of the base-URI split this PR is actually about. A sibling of the jump
test that patches `DocumentFactory.open_document` and asserts the URI would
close it.
Also missing: nothing covers the deferred consume *failing*. The PR body
says a read failure is reported like a UDF error, and there are good tests for
a bad `condition()` and a failed back-edge write, but not for a failed read or
a failed `update` inside `_consume_pending_loop_state`. It's the same test with
a different boom -- and it's the test that would show whether the
port-completed ordering I flagged on `complete()` is actually a problem.
--
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]
