aglinxinyuan opened a new pull request, #6661:
URL: https://github.com/apache/texera/pull/6661
### What changes were proposed in this PR?
A loop with a Java/Scala built-in operator in its body could never iterate —
the Loop End worker crashed on the first iteration:
```
RuntimeError: no loop-back state URI configured for LoopStart ''
```
**Root cause.** Per-iteration loop state rides a `StateFrame` envelope —
`loop_counter` (nesting-depth countdown) and `loop_start_id` (the back-jump
target) — materialized as their own columns next to `content` (#5900). The
Python worker carries the envelope on every hop; the Scala side stripped it at
every seam, so any JVM hop re-emitted states with the "no loop" defaults `(0,
"")`:
```
LoopStart ──(0, LS-id)──▶ Limit(Scala) ──(0, "")──▶ LoopEnd
▲ envelope stripped │
▼
back-jump lookup by "" fails
```
In a nested loop the zeroed `loop_counter` additionally makes the inner Loop
End mis-consume the enclosing loop's state instead of passing it through.
**Fix.** Loop operators are Python-only, so a JVM operator only ever needs
to *carry the envelope through unchanged* (the +1/−1 bookkeeping lives in the
Python runtime). The columns already exist in the materialized State schema and
on the Arrow flight wire — no format change, purely Scala-side plumbing:
| Seam | Change |
|---|---|
| `StateFrame` (`DataPayload.scala`) | gains `loopCounter` / `loopStartId`
fields (defaults `0` / `""`, mirroring Python's `StateFrame`) |
| `State` (`State.scala`) | `loopCounterFrom` / `loopStartIdFrom` extractors
read the envelope columns back off a row |
| `InputPortMaterializationReaderThread` | rebuilds the envelope when
replaying materialized states |
| `DataProcessor` → `OutputManager` | the incoming envelope is threaded
through `processInputState` to `emitState` / `saveStateToStorageIfNeeded` /
`sendState` |
| `PythonProxyClient` / `PythonProxyServer` | envelope preserved across the
Arrow flight bridge in both directions |
Scala-originated states (start/end-channel handlers) keep the defaults —
correct, since a JVM operator can never be a Loop Start.
### Any related issues, documentation, discussions?
Closes #6660
### How was this PR tested?
- **Unit** — each seam is pinned locally: `StateSpec` (envelope column
round-trip + no-loop defaults), `DataProcessorSpec` (a state pass-through emits
the exact incoming envelope, not just *a* `StateFrame`), `OutputManagerSpec`
(`emitState` stamps the envelope onto every buffer), `NetworkOutputBufferSpec`
(`sendState` stamps the frame).
- **E2E** — two new `LoopIntegrationSpec` cases run real workflows with a
Scala `Limit` operator in the loop body: a single loop (`TextInput → LoopStart
→ Limit → LoopEnd`, asserts 3 accumulated iterations — this reproduces the
reported failure verbatim and fails against the pre-fix code) and a nested 3×3
loop with `Limit` in the inner body (asserts 9 outer / 3 inner rows — pins that
the counter *magnitude* survives the JVM hop, not just the id).
- `scalafmtCheckAll` + `scalafixAll --check` + full test-compile pass
locally (Java 17).
### Was this PR generated or co-authored using GenAI tooling?
Generated-by: Claude Code (Fable 5)
--
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]
