aglinxinyuan commented on code in PR #8126:
URL: https://github.com/apache/texera/pull/8126#discussion_r3942886408
##########
amber/src/main/python/core/models/operator.py:
##########
@@ -366,12 +366,16 @@ class LoopStartOperator(TableOperator):
@overrides.final
def process_state(self, state: State, port: int) -> Optional[State]:
- # First-entry only: merge upstream state into self.state. The nested
- # pass-through (a frame already stamped with a LoopStartId) and all
- # loop_counter bookkeeping are owned by the worker runtime
- # (main_loop._process_state_frame), so this operator never sees the
- # counter and never mutates the State it is handed.
- self.state.update(state)
+ # First-entry only: merge non-conflicting upstream state into
self.state.
+ # The nested pass-through (a frame already stamped with a LoopStartId)
and
+ # all loop_counter bookkeeping are owned by the worker runtime
+ # (main_loop._process_state_frame), so this operator never sees the
counter
+ # and never mutates the State it is handed.
+ for key, value in state.items():
+ if key in self.state:
+ raise ValueError(f"Loop state variable cannot be overwritten:
'{key}'")
Review Comment:
With the back-edge stamped, this only sees outside state — nice. One note
for the PR body: outside state can still *add* new loop variables, just not
overwrite them.
##########
amber/src/main/python/core/runnables/main_loop.py:
##########
@@ -216,7 +216,7 @@ def _jump_to_loop_start(
writer = DocumentFactory.create_document(uri, State.SCHEMA).writer("0")
# The back-edge fires only after the matching LoopEnd consumed at
# loop_counter == 0, so the next iteration's input starts at depth 0.
- writer.put_one(executor.state.to_tuple(0))
+ writer.put_one(executor.state.to_tuple(0, self._loop_start_id))
Review Comment:
Stamping the back-edge is the key move — it's what makes the rest safe. LGTM.
##########
amber/src/main/python/core/runnables/main_loop.py:
##########
@@ -515,31 +515,40 @@ def _process_state_frame(self, frame: StateFrame) -> None:
self._emit_and_save_state(state, in_counter - 1,
frame.loop_start_id)
self._check_and_process_control()
return
- if isinstance(executor, LoopStartOperator) and frame.loop_start_id:
- # Outer loop's state flowing through an inner LoopStart -- detected
- # by the outer LoopStart's id stamped on the envelope (a
first-entry
- # state has no stamp): step one level deeper and forward, keeping
- # the outer loop's id.
- self._emit_and_save_state(state, in_counter + 1,
frame.loop_start_id)
+ if (
+ isinstance(executor, LoopStartOperator)
+ and frame.loop_start_id
+ and frame.loop_start_id !=
get_logical_op_id(self.context.worker_id)
+ ):
+ # State belongs to an outer loop flowing through this inner
LoopStart.
+ # Forward it one level deeper while preserving the outer loop's id.
+ self._emit_and_save_state(
+ state,
+ in_counter + 1,
+ frame.loop_start_id,
+ )
+ self._check_and_process_control()
+ return
+
+ if (
+ isinstance(executor, LoopStartOperator)
+ and frame.loop_start_id
+ and frame.loop_start_id ==
get_logical_op_id(self.context.worker_id)
+ ):
+ # This is this LoopStart's own back-edge state. Restore the loop
+ # variables directly instead of routing the state through
+ # process_state(), which is reserved for external/first-entry
state.
+ executor.state = state
Review Comment:
This replaces the state instead of merging it, so a variable deleted in
`update` stays deleted next iteration. I think that's better than before — just
mention it in the PR body. Optional: merge the two stamped branches into one
`if/else`.
--
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]
