aglinxinyuan commented on code in PR #6913:
URL: https://github.com/apache/texera/pull/6913#discussion_r3697390279
##########
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:
You're right that this trades a loud failure for a silent wrong result, and
that's not acceptable — fixed in c5fa702. But I went with a different
discriminator than `table`, for two reasons.
**`_TABLE_KEY` stops working one PR from now.** #6971 takes the table out of
the State: `produce_state_on_finish` becomes `return State(self.state)` with no
`table`, and that PR even pins the absence (`assert "table" not in produced`).
So after it lands `_TABLE_KEY in state` is always False for a real loop state,
the raise-arm becomes unreachable, and every unstamped frame — including a lost
stamp — takes the silent path again. Nothing in the suite would notice; the
guard would rot green, leaving behind a comment asserting an invariant that
another PR test-pins as false.
**It can also misfire today.** `_reserved_name_error` is only checked inside
`LoopStartOperator.produce_state_on_finish` and `run_update`. Nothing stops an
ordinary body UDF from returning `State({"table": ...})` from
`produce_state_on_finish`, and that would be misread as a lost stamp and
hard-fail a valid workflow.
So the check moved off the State content and onto the envelope, decided once
the input port is drained:
```python
if self._forwarded_unstamped_state and not self._loop_state_consumed:
raise RuntimeError("... the loop envelope was lost upstream ...")
```
A Loop End that forwarded an unstamped state and never took a stamped one
never received its own state. That's order-independent (the body op's state may
arrive before or after the loop's — EndChannel is PORT_ALIGNMENT, so the port
is drained by then) and reads nothing out of the State, so #6971 doesn't affect
it.
Two details worth flagging:
- It runs in `_process_end_channel`, **not** `complete()` — same ordering
point you raised on #6971. `complete()` is downstream of `port_completed` for
the input port and every output port, and region completion is port-based, so a
raise there would land after the coordinator already considers the region done.
- It's deliberately narrow — *"forwarded an unstamped state AND never took a
stamped one"*, not *"never took a stamped one"*. The broad version isn't safe:
`eval_condition`'s `if self._loop_table is None: return False` explicitly
declares a Loop End completing without a matching state legal, and four
existing tests rely on that.
Pinned by
`test_end_channel_holds_the_region_when_only_unstamped_states_arrived` (asserts
the error is reported *and* no port is reported complete) plus
`test_complete_accepts_unstamped_state_alongside_the_loop_state`, which runs
the legitimate shape in both arrival orders. Verified red without the guard.
##########
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:
Deliberate, and the reason is stronger than "that's how upstream state seeds
the loop" — documented at the branch in c5fa702, with a test.
The back-edge writes the next iteration's variables to the Loop Start's
**own input-port state URI** with the identical "no loop" envelope
(`_jump_to_loop_start` → `State.to_tuple(0)`, whose defaults are
`loop_counter=0, loop_start_id=""`). So an unstamped counter-0 frame at a Loop
Start is indistinguishable from — and normally *is* — the loop's own state. It
has to merge. A Loop End can forward instead, because its inbound loop state is
always stamped by the matching Loop Start; a Loop Start has no such signal. The
asymmetry follows from a real protocol asymmetry rather than being an oversight.
One correction to the nested case: the body op's state isn't lost. Its keys
are merged into `self.state` and re-emitted by `produce_state_on_finish`, so
the content does travel downstream — as inner-loop variables rather than as a
separate frame. Misattributed, not dropped.
Both consequences you name are real and now in the comment: a colliding key
overwrites a loop variable, and a `table` key trips `_reserved_name_error`.
That hazard predates this PR (`operator.py:381` is from #5700) and this PR
widens its reach rather than creating it, so I'd rather not defend it here —
I'll file it separately.
Added `test_loopstart_merges_unstamped_state_instead_of_forwarding_it` so
the asymmetry is pinned, not just asserted in prose.
##########
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:
Fixed — dropped. You're right it's the only one under any `src/main` in the
repo; the 5 pre-existing ones are all in test code, where the surrounding
`Reviewer feedback (...)` prefix makes them resolvable. The comment above
already carries the reasoning, so the anchor added nothing.
##########
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:
Fixed — both sites are down to a 3-line pointer:
```scala
// Operator-ORIGINATED boundary state, so no LoopStart stamp
// (loopCounter = 0, loopStartId = ""); see
// `main_loop._process_state_frame` for how a Loop End treats it.
```
The full rationale stays in one place, on the Python side that acts on it.
##########
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:
Fixed — `source.scan.text` now precedes `udf.python`. And confirmed nothing
would have caught it: both `.scalafix.conf` files enable only `ProcedureSyntax`
+ `RemoveUnused`, there's no `OrganizeImports` rule or `scalafixDependencies`
anywhere, and neither `.scalafmt.conf` has an import-sorting setting.
--
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]
