aglinxinyuan commented on code in PR #6971:
URL: https://github.com/apache/texera/pull/6971#discussion_r3697235780
##########
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:
Good catch, and agreed — moved. The consume now runs in
`_process_end_channel` right after `process_input_state()`, i.e. before the
`has_exception()` hold rather than after it, so a failing read or `update` is
caught by that same guard and nothing is reported complete. It is still past
the reader (EndChannel means it finished streaming).
I put it before the hold rather than after `process_input_tuple()` so it
reuses the existing guard instead of needing a second one; the ordering
property you asked for is the same. It also drops the dependency you flagged on
the other thread — the stash no longer needs `complete()` to run at all, so the
`is_missing_output_ports()` early return can't strand it.
New test `test_end_channel_holds_the_region_when_the_deferred_consume_fails`
pins it: a failing `update` reports the error, and asserts no `port_completed`
is sent and the worker does not complete.
##########
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:
Fixed rather than documented — `replace_print` is a standalone context
manager, so it can be applied from this thread even though the rest of
`_executor_session` (the context switch) can't. Both the `update` in the
deferred consume and `condition()` now run under it, which also closes the
pre-existing gap on `condition()`.
Test `test_deferred_consume_captures_user_prints` asserts a `print()` in the
user's update comes back as a PRINT console message.
##########
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:
You're right, and that is the comment that would mislead. Reworded to say
the doc is stable *for the duration of the loop level that reads it*, and
spelled out the nested case explicitly: an inner loop's upstream is the outer
Loop Start's output, recreated on each outer iteration, which is still stable
across every inner iteration that reads it — the window that actually matters.
Fixed the same overstatement in `_consume_pending_loop_state`'s comment too.
##########
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:
Added the assert — `_pending_loop_state` must be empty at the stash, with
the reason (Loop End is non-parallelizable and its matching LoopStart emits one
state per iteration, so a second frame means the routing invariants broke) so
the assumption is checkable instead of implicit.
The `complete()` dependency is gone: with the consume moved into
`_process_end_channel` (your other comment), it runs before the
`is_missing_output_ports()` early return, so that path can no longer strand a
stashed state.
##########
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:
Fixed — `run_update` now takes `_attached_table` and clears it, so the guard
fires on every iteration and a stale table can never be silently reused. Also
documented why the two fields exist: `_loop_table` doubles as the "consumed"
marker that `condition()` short-circuits on, and only a successful update may
set it.
##########
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:
Both added.
`test_read_loop_input_table_opens_the_result_uri_of_the_configured_base`
patches `DocumentFactory.open_document` and asserts the URI is
`result_uri(base)` of the configured base, with a sibling pinning the fail-loud
path when no entry is configured — that's the half of the split that had no
coverage.
`test_end_channel_holds_the_region_when_the_deferred_consume_fails` covers
the failing consume, and as you predicted it is what demonstrates the ordering
problem you raised on `complete()`: it only passes with the consume moved ahead
of the port-completed sends.
--
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]
