Copilot commented on code in PR #6971:
URL: https://github.com/apache/texera/pull/6971#discussion_r3697230746
##########
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:
`LoopEndOperator.run_update` never clears `_attached_table` after consuming
it. After the first successful iteration, a later update can run without a
fresh `attach_loop_table()` call and will silently reuse a stale table (the
guard only checks for `None`). Clearing `_attached_table` after capturing it
enforces the intended “attach per consume” contract and avoids stale-table bugs
(especially with nested loops where the table can vary per outer iteration).
##########
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:
The deferred LoopEnd consume now runs in `complete()` via
`_consume_pending_loop_state(executor)`, outside
`DataProcessor._executor_session`. That means user `print()` output from the
loop update/condition won’t be captured into console messages (and won’t be
flushed), which is a behavioral regression compared to running `process_state`
on the DataProcessor thread. Wrap the deferred consume/condition/jump in
`replace_print(...)` and flush console messages afterwards to preserve console
output behavior.
--
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]
