feat: add loop operators

[aglinxinyuan]

Re-opened from my fork to satisfy the requirement that contributions come from a fork rather than a branch on the main repository (the prior PR #4206 was on an apache/texera branch). The full review history — Copilot's and @Xiao-zhen-Liu's review threads and my replies — is preserved on #4206 for reference.

---

What changes were proposed in this PR?

Adds two new operators — Loop Start and Loop End — that let users write a for-loop inside a visual workflow. The loop iterates over rows of a pandas table. The user supplies four small Python snippets:

Field	Where	Example
initialization	Loop Start	i = 0
output	Loop Start	table.iloc[i] — the row passed into the loop body each iteration
update	Loop End	i += 1
condition	Loop End	i < len(table) — keep looping while this is true

Operators placed between Loop Start and Loop End make up the loop body and run once per iteration. When condition returns true, the runtime starts another iteration; when it returns false, downstream operators run on the accumulated output.

initialization and update are Python statements; output and condition are single Python expressions (evaluated with eval, so multi-statement input is rejected up front). The only reserved name in loop state is table (the runtime seeds it into the expression namespace and strips it from persisted state).

How an iteration works

   Upstream Table
        │
        ▼
   ┌──────────┐   loop variables: row i,   ┌──────────┐               ┌─────────┐
   │ Loop     ├───counters, accumulators ─►│   loop   ├──────────────►│  Loop   │
   │ Start    │   (the loop's "state")     │   body   │               │   End   │
   └──────────┘                            └──────────┘               └────┬────┘
        ▲                                                                  │
        │  (1) DCM: "schedule the Loop Start region again"                 │
        │  (2) write the next iteration's state (i, accumulators, table)   │
        │      to the iceberg table that Loop Start reads its input from   │
        └──────────────────────────────────────────────────────────────────┘
                                when condition() == True

The arrow from Loop End back to Loop Start is not an edge in the workflow graph — the region DAG stays acyclic. The loop-back is done with two separate steps when an iteration ends:

1. Loop End sends a DCM (Direct Control Message — Texera's worker→controller control-channel message; it does not flow along data edges, so it doesn't break the acyclic DAG) named jump_to_operator_region, asking the controller to schedule the Loop Start region one more time.
2. Loop End writes the updated state — a dict with i, any accumulators, and the table (Arrow IPC) — into the iceberg state table that Loop Start reads from at the start of every iteration.

The newly scheduled Loop Start region then picks up that state and runs the next iteration. The "iceberg table Loop Start reads from" is the same cross-region state channel introduced in #4490; this PR reuses it as the back-edge for loops.

How Loop End knows where to write (setup-time config)

The loop-back write address is the same every iteration — same Loop Start, same input port, same execution — so it is config, not per-iteration data, and it is deliberately not carried in State (settled during the #5900 review; the materialized State format has exactly 3 columns: content, loop_counter, loop_start_id). Instead:

• LoopStartOpDesc marks its physical op with a compile-time isLoopStart flag (mirroring the existing requiresMaterializedExecution precedent).
• After resource allocation, WorkflowExecutionCoordinator derives a whole-plan map {Loop Start logical op id → state URI of its single input port} from the final schedule — the same URIs AssignPort ships to that Loop Start's input readers.
• The map rides the existing per-operator setup RPC: InitializeExecutorRequest gains a loopStartStateUris field, re-sent identically on every region (re-)run, and the Python worker stores it on its Context.
• No static Loop Start ↔ Loop End pairing is needed: a Loop End selects the map entry by the loop_start_id stamped on the StateFrame it consumes — the same id it already uses for the jump DCM — so nested loops work unchanged. A missing entry fails loudly before the jump, so a misconfigured loop can't rewind the schedule without a back-edge write.

What changed

Area	File	Purpose
Operator definitions	LoopStartOpDesc.scala, LoopEndOpDesc.scala	Code-gen the Python operator from the user's initialization / output / update / condition expressions; Loop Start additionally sets isLoopStart, Loop End sets reuseStorage on its output port
Operator runtime base	core/models/operator.py — LoopStartOperator, LoopEndOperator	Python superclasses the generated code extends; guarded expression evaluation keeps the reserved table name out of user loop state
Physical plan	PhysicalOp.scala — isLoopStart + requiresMaterializedExecution	Compile-time markers: the scheduler resolves each Loop Start's loop-back write address from isLoopStart, and any loop op forces whole-plan MATERIALIZED scheduling via requiresMaterializedExecution (the back-edge is a cross-region materialized state channel)
Loop-back address resolution	WorkflowExecutionCoordinator.scala	Derives {Loop Start op id → input-port state URI} from the final resource-allocated schedule; fails at StartWorkflow (not mid-loop) if a Loop Start's input port isn't materialized
Setup delivery	controlcommands.proto — InitializeExecutorRequest.loopStartStateUris; RegionExecutionCoordinator.scala; initialize_executor_handler.py + Context	Ships the map to every worker at executor initialization, each region (re-)run — always before StartWorker spawns the readers that replay states
Worker runtime	MainLoop._process_state_frame, _compute_loop_start_id, _jump_to_loop_start	Loop Start stamps its op id on emitted state; the runtime owns loop_counter and the nested pass-through; on Loop End completion: look up the write address by the captured id, send the jump DCM, write the next state
Storage reuse	OutputPort.reuseStorage (set by Loop End) + RegionExecutionCoordinator	A Loop End's materialized output accumulates across its own iterations: the scheduler reuses its output doc on region re-runs (DocumentFactory.createOrReuseDocument) instead of recreating it
Worker output	OutputManager.reset_output_storage	Only for the inner Loop End of a nested loop: fires once per outer iteration (on the loop_counter > 0 pass-through), dropping and recreating that Loop End's result/state tables so each outer iteration accumulates from empty. A single / outermost Loop End never resets. Safe because loops run MATERIALIZED — downstream doesn't read until the loop region completes
Frontend	LoopStart.png, LoopEnd.png	Operator icons

Nested loops

Each state carries a loop_counter integer marking which loop's iteration the state belongs to. This is what keeps an inner Loop End from accidentally consuming an outer loop's state.

Operator	Rule
Inner LoopStart pass-through	If the state already came from a Loop Start (loop_start_id is stamped on the envelope), loop_counter += 1 and pass it through — the operator is skipped.
Inner LoopEnd pass-through	If loop_counter > 0, decrement and pass it through (this state belongs to an outer loop); also reset this inner Loop End's output so the new outer iteration accumulates from empty.
Matching LoopEnd	If loop_counter == 0, the state is mine: run update, evaluate condition, jump back to the Loop Start whose id is stamped on the state.

So when two loops are nested, the outer loop's state walks through the inner Loop Start (+1) and the inner Loop End (−1) untouched, arrives at the outer Loop End at loop_counter == 0, and only there is it consumed. Each Loop End writes back to whichever Loop Start stamped the state it consumed — the setup-injected map has an entry per Loop Start, so no extra machinery is needed for nesting.

Any related issues, documentation, discussions?

Closes #4442. Builds on #4490 (cross-region state materialization) and #5085 (DocumentFactory.documentExists).

Prerequisites split out of this PR per @Xiao-zhen-Liu's request, all now merged: #5706 (worker-id helper), #5707 (state test harness), #5900 (the 3-column State materialization format — content / loop_counter / loop_start_id — dormant on main until this PR activates it).

How was this PR tested?

• test_main_loop.py::TestMainLoop (loop-runtime cases) — nested pass-through (+1 at an inner Loop Start / −1 + reset-once at an inner Loop End, operator skipped, id rides through), consume at loop_counter == 0 with envelope metadata never leaking into user State, and the full _jump_to_loop_start contract (jump DCM target, DCM-before-storage-write ordering, exact iceberg write sequence) driven off the setup-injected config — including the fail-loud case when the map lacks the captured id.
• test_initialize_executor_handler.py — the new proto field lands on Context.loop_start_state_uris (and resets to empty when absent, since recreated workers re-run the RPC).
• test_loop_operators.py — runtime base classes: flat-loop matching branch (runs update / condition), nested-loop pass-through, guarded expression evaluation, and a multi-iteration loop driven to completion.
• LoopStartOpDescSpec.scala / LoopEndOpDescSpec.scala — code-gen output, ports, JSON round-trip, and the flag pins: requiresMaterializedExecution (both), isLoopStart (true / false), reuseStorage (Loop End only).
• LoopIntegrationSpec.scala (@IntegrationTest, CI) — end-to-end single loop (3 iterations) and nested 3×3 loop (9 inner iterations), verified via materialized iceberg row counts; adopts the per-suite id isolation from fix(test): isolate e2e suites by unique workflow/execution id #5888.
• sbt scalafmtCheckAll + scalafixAll --check and Python black clean; full Python unit suite passes.

Manual workflows

Input for both is a 3-row table from TextInput("1\n2\n3"). Each loop's condition is i < len(table).

Workflow	Topology	Expected
Loop.json	TextInput → LoopStart → LoopEnd	3 iterations, workflow terminates.
Nested Loop.json	TextInput → OuterLoopStart → InnerLoopStart → InnerLoopEnd → OuterLoopEnd	Outer runs 3 times, inner runs 3 times per outer iteration = 9 total inner iterations. Workflow terminates.

Demo:
Basic Loop:

Nested Loop:

Was this PR authored or co-authored using generative AI tooling?

Co-authored with Claude (Opus 4.7, Opus 4.8) in compliance with ASF.

[github-actions]

Automated Reviewer Suggestions

Based on the git blame history of the changed files, we recommend the following reviewers:

• Contributors with relevant context: @Ma77Ball, @Yicong-Huang, @mengw15
  You can notify them by mentioning @Ma77Ball, @Yicong-Huang, @mengw15 in a comment.