aglinxinyuan commented on code in PR #6661:
URL: https://github.com/apache/texera/pull/6661#discussion_r3653409240
##########
amber/src/main/scala/org/apache/texera/amber/engine/architecture/messaginglayer/OutputManager.scala:
##########
@@ -192,9 +192,16 @@ class OutputManager(
buffersToFlush.foreach(_.flush())
}
- def emitState(state: State): Unit = {
- networkOutputBuffers.foreach(kv => kv._2.sendState(state))
- saveStateToStorageIfNeeded(state)
+ /**
+ * Emit a State to every network buffer and (if configured) the state
+ * storage. `loopCounter` / `loopStartId` are the loop envelope riding
+ * alongside the State (see `StateFrame`); a JVM hop inside a loop body
+ * passes the incoming envelope through unchanged, while a Scala-originated
+ * state (start/end-channel handlers) uses the "no loop" defaults.
+ */
+ def emitState(state: State, loopCounter: Long = 0L, loopStartId: String =
""): Unit = {
Review Comment:
Good catch — confirmed, and it's actually worse than the id clobber: the
unstamped state also lands in `run_update`, which expects the loop `table`
payload, so it raises KeyError before the jump even runs. It's
language-independent too — a Python UDF's `produce_state_on_finish` hits the
identical path (no JVM built-in overrides `produceStateOnFinish` today, so
that's also the practical repro).
Fixed in #6913 on the consumer side: a real loop state is always stamped by
its LoopStart, so LoopEnd now keys on the stamp — unstamped counter-0 states
forward through unchanged (operator skipped, captured id untouched). Also added
the comment you suggested at the two boundary-state emit sites. Includes a unit
test plus an e2e case with a state-emitting UDF in the loop body.
--
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]
