aglinxinyuan commented on code in PR #6971:
URL: https://github.com/apache/texera/pull/6971#discussion_r3697253754
##########
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:
+ raise RuntimeError(
+ "loop input table was not attached before the update; the "
+ "runtime must call attach_loop_table on the matching consume"
+ )
+ input_table = self._attached_table
namespace = {**state, _TABLE_KEY: input_table}
Review Comment:
Fixed — `run_update` now takes `_attached_table` and clears it, so the
missing-table guard fires on every iteration instead of only the first and a
stale table can never be silently reused. Same point Xiao-zhen-Liu raised on
this line.
##########
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)
if executor.condition():
self._jump_to_loop_start(executor, coordinator_interface)
except Exception as err:
Review Comment:
Both halves fixed.
Print capture: the `update` (in the deferred consume) and `condition()` now
run under `replace_print`, so their output reaches the console instead of the
worker's stdout. `condition()` had this gap before this PR too.
Flushing: you were right that capture alone wasn't enough. `complete()`
flushes on entry — *before* `condition()` runs — and then shuts the worker
down, so a print inside the condition was captured and never sent. Added a
flush after the condition/jump (the error path already flushed via
`_check_exception`). Tests pin both:
`test_deferred_consume_captures_user_prints` and
`test_complete_flushes_prints_from_the_user_condition`.
One correction to the premise: the consume no longer runs in `complete()` —
it moved into `_process_end_channel`, ahead of the port-completed sends, so a
failure there can still hold the region.
--
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]
