Xiao-zhen-Liu commented on code in PR #6913:
URL: https://github.com/apache/texera/pull/6913#discussion_r3696858250
##########
amber/src/main/python/core/runnables/main_loop.py:
##########
@@ -374,9 +374,23 @@ def _process_state_frame(self, frame: StateFrame) -> None:
return
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.
+ if not frame.loop_start_id:
+ # An UNstamped counter-0 state at a LoopEnd is not the loop's
+ # own boundary state -- it was produced by a loop-body
+ # operator's produce_state_on_start/finish (a public API on
+ # both engine sides), which emits with the "no loop" envelope.
+ # A real loop state is always stamped: the matching LoopStart
+ # stamps its own id on every iteration's output state.
+ # Forward it downstream unchanged, skipping the operator, like
+ # any default pass-through: consuming it would clobber the
+ # captured back-jump id with "" and hand run_update a State
+ # with no `table` payload (#discussion_r3648708075).
Review Comment:
`(#discussion_r3648708075)` is the first reference of this kind under
`amber/src/main` - the tests use them, production code doesn't. A bare id isn't
resolvable by anyone reading this in a year. Either write the full URL or drop
it; the comment above already says why.
##########
amber/src/main/python/core/runnables/main_loop.py:
##########
@@ -374,9 +374,23 @@ def _process_state_frame(self, frame: StateFrame) -> None:
return
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.
+ if not frame.loop_start_id:
+ # An UNstamped counter-0 state at a LoopEnd is not the loop's
+ # own boundary state -- it was produced by a loop-body
+ # operator's produce_state_on_start/finish (a public API on
+ # both engine sides), which emits with the "no loop" envelope.
+ # A real loop state is always stamped: the matching LoopStart
+ # stamps its own id on every iteration's output state.
+ # Forward it downstream unchanged, skipping the operator, like
+ # any default pass-through: consuming it would clobber the
+ # captured back-jump id with "" and hand run_update a State
+ # with no `table` payload (#discussion_r3648708075).
+ self._emit_and_save_state(state, in_counter,
frame.loop_start_id)
Review Comment:
Two different situations produce an unstamped counter-0 frame here, and this
treats them the same:
1. a body operator's own boundary state - forwarding is right;
2. the real loop state that lost its stamp somewhere upstream.
Case 2 isn't hypothetical: it's exactly the bug the "loop whose body
contains a JVM (Scala) operator" e2e was written for
(`LoopIntegrationSpec.scala:286-292` - the Scala hop "zeroed the counter and
blanked the id"). Before this change that failed loudly with `no loop-back
state URI configured for LoopStart ''`. After it, the state is forwarded,
`run_update` never runs, `_loop_table` stays `None`, `condition()` returns
`False`, and the loop quietly stops after one iteration - the workflow
completes and reports success with one row instead of three. The e2e still
catches it, but as `expected 3, got 1` rather than a message naming the cause.
Cheap way to keep the loud failure: a real loop state always carries the
reserved `table` key (`LoopStartOperator.produce_state_on_finish`,
`operator.py:414-418`), and an operator-originated boundary state won't. So
`not frame.loop_start_id and _TABLE_KEY in state` is a lost stamp - raise there
with the old message instead of forwarding, and forward only the rest.
##########
amber/src/test/integration/org/apache/texera/amber/engine/e2e/LoopIntegrationSpec.scala:
##########
@@ -44,6 +44,7 @@ import org.apache.texera.amber.operator.LogicalOp
import org.apache.texera.amber.operator.limit.LimitOpDesc
import org.apache.texera.amber.operator.loop.{LoopEndOpDesc, LoopStartOpDesc}
import org.apache.texera.amber.operator.sleep.SleepOpDesc
+import org.apache.texera.amber.operator.udf.python.PythonUDFOpDescV2
Review Comment:
Import is out of alphabetical order (`udf.python` before
`source.scan.text`). scalafix has no `OrganizeImports` rule here so nothing
will flag it.
##########
amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/StartChannelHandler.scala:
##########
@@ -42,6 +42,12 @@ trait StartChannelHandler {
try {
val outputState = dp.executor.produceStateOnStart(portId.id)
if (outputState.isDefined) {
+ // Deliberate "no loop" envelope defaults (loopCounter = 0,
+ // loopStartId = ""): this is operator-ORIGINATED boundary state, not
+ // a forwarded loop state, so it carries no LoopStart stamp. The
+ // Python LoopEnd runtime keys on that missing stamp to pass such
+ // states through instead of consuming them (see
+ // main_loop._process_state_frame).
Review Comment:
Six identical lines here and in `EndChannelHandler.scala:46-51`. Two would
carry it - "operator-originated boundary state, so no LoopStart stamp; see
`main_loop._process_state_frame` for how the LoopEnd treats it" - and the long
version already lives at the place that acts on it, so it won't drift out of
sync.
##########
amber/src/main/python/core/runnables/main_loop.py:
##########
@@ -374,9 +374,23 @@ def _process_state_frame(self, frame: StateFrame) -> None:
return
if isinstance(executor, LoopEndOperator):
Review Comment:
(About the LoopStart branch just above, at line 367 - outside the diff so I
can't anchor it there.)
The same mix-up exists on that side and the PR doesn't say what happens to
it. An unstamped counter-0 frame falls through to `process_input_state`, and
`LoopStartOperator.process_state` does `self.state.update(state)`
(`operator.py:381`). So:
- `Source -> statefulOp -> LoopStart`: the op's keys are merged into the
loop variables. An op emitting `{"i": ...}` silently overwrites the loop
counter; one emitting `{"table": ...}` trips `_reserved_name_error` at
`produce_state_on_finish`.
- `OuterStart -> statefulOp -> InnerStart -> ...`: the op's boundary state
is absorbed by the inner LoopStart and never reaches downstream - the opposite
of what the LoopEnd now does with the identical frame.
I assume the merge is deliberate (that's how upstream state seeds the loop),
but with the branch below now doing the opposite for the same shape of frame,
that intent should be one sentence in this comment. If it isn't deliberate,
it's a follow-up.
--
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]
