The GitHub Actions job "Required Checks" on texera.git/main has succeeded. Run started by GitHub user github-merge-queue[bot] (triggered by github-merge-queue[bot]).
Head commit for run: ed16a605dacc0185bc5f069d14157ee3e128b68a / Xinyuan Lin <[email protected]> refactor(amber): read the loop input table from its materialization instead of shipping it in state (#6971) ### What changes were proposed in this PR? The loop's input table used to ride **inside the State content**: LoopStart encoded its buffered input as Arrow IPC bytes, base64'd into the JSON `content` column, and that payload was re-written and re-read at **every loop-body hop, every iteration**. That data already exists. In the fully-materialized mode loops require, the Loop Start's input-port materialization holds exactly the loop's input table for the whole loop — Loop Start re-reads it every iteration, and the back-edge truncates only the *state* doc at the same base URI, never the *result* doc. So this PR ships the port's **base URI** in the setup config and derives both addresses from it: ``` loopStartPortUris[LoopStart-id] = <base URI of LoopStart's input port> ├── state_uri(base) → back-edge write address (as before, derived) └── result_uri(base) → the loop's input table (NEW: read at EndChannel) ``` | Piece | Before | After | |---|---|---| | proto field 4 | `loopStartStateUris` = state URI | `loopStartPortUris` = base URI (renamed so the semantic change is loud) | | LoopStart's produced state | user vars + IPC-encoded table (base64 in JSON) | user vars only — small, pure JSON | | Loop-body hops | fat state re-materialized per hop, per iteration | tiny state | | LoopEnd's table | decoded from state content | read once per iteration from `result_uri(base)` at EndChannel, injected via a new runtime-only `attach_loop_table` hook | | `table_to_ipc_bytes` / `table_from_ipc_bytes` | second, divergent Arrow codec (lossy `from_pandas` inference) | deleted — the read goes through the canonical iceberg reader | **When the read happens matters.** The read is issued at **EndChannel** (the matching state is stashed at consume and the operator's update runs in `complete()`), not at consume time. At consume time this worker's own materialization reader is still streaming, and issuing a second iceberg/S3 read from the main loop thread in that window made the reader fail with S3 `Access Denied` — `LoopIntegrationSpec` hung to the CI job timeout on both OSes. Deferring past the reader removes the overlap: integration went from a 20-minute cancel to green in ~9 minutes. The matching consume emits no state downstream, so moving it is unobservable outside the operator. Semantics deliberately preserved: - the reserved-`table` collision raise stays (a user var named `table` would now be *silently shadowed* by the injected table — worse than before); - the "consumed" marker (`_loop_table`) is still set only by a **successful** `run_update`, so `condition()`'s short-circuit for pass-through-only Loop Ends is unchanged; - nested loops work by construction: the inner Loop Start's entry points at the outer Loop Start's output port, whose result doc is recreated per *outer* iteration but persists across *inner* iterations (the jump rewinds to the inner level only). Wins: no ~33% base64 bloat, no JSON-column size ceiling on the table (large-table loops become viable), strictly less I/O for any non-empty loop body (one read per iteration replaces N state-doc writes+reads per hop), and one Arrow codec instead of two. Note: this deepens the read-side use of `storagePairs.head._1` — the same shared upstream URI as the known back-edge fan-out design discussion; if that ever moves to a per-loop private doc, this read moves with it. ### Any related issues, documentation, discussions? Builds on #5900 (State columns) and #6661 (envelope through JVM hops). Related design context: #6660. ### How was this PR tested? - **Unit** — `test_loop_operators.py` rewritten for the attach-based flow plus new pins: the produced state carries no `table`; attaching alone does **not** mark the loop consumed; `run_update` fails loud when no table was attached. `test_main_loop.py` pins the base-URI derivation for the back-edge write (`state_uri(base)`), the missing-config fail-loud, and that the matching consume stashes the state without touching storage, with the read + update happening once at EndChannel. `test_initialize_executor_handler.py` covers the renamed proto field. 237 tests green locally (the only failures in a full sweep are pre-existing environment ones, identical on unmodified main). - **Scala** — full test-compile (proto regen included), `scalafmtCheckAll`, `scalafixAll --check`, and the worker/descriptor spec suites (`WorkerSpec`, `WorkflowWorkerSpec`, `SerializationManagerSpec`, `WorkflowExecutionManagerSpec`, `LoopStartOpDescSpec`, `LoopEndOpDescSpec`) all pass on Java 17. - **E2E** — the four `LoopIntegrationSpec` cases (single, nested 3×3, JVM chain, nested JVM chain) exercise the full read path in the `amber-integration` CI job (both jobs green in ~9 min); the nested cases specifically cover the inner-loop read against the outer Loop Start's per-outer-iteration output doc. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Fable 5) Report URL: https://github.com/apache/texera/actions/runs/31468008416 With regards, GitHub Actions via GitBox
