feat: add loop operators

[aglinxinyuan]

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.

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 pickled table — into the iceberg 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.

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
Operator runtime base	core/models/operator.py — LoopStartOperator, LoopEndOperator	Python superclasses the generated code extends; manage loop_counter and the state dict
Scheduler	RegionExecutionCoordinator.scala	Don't recreate Loop End's iceberg output between iterations (the output accumulates the body's rows; recreating it would erase what we just wrote)
Output-storage reuse flag	PhysicalOp.scala — reusesOutputStorageOnReExecution: Boolean + withReusesOutputStorageOnReExecution	A flag the scheduler checks (instead of string-matching the operator id) to skip recreating an operator's output on region re-execution; Loop End sets it so a loop's own iterations accumulate
Worker runtime	MainLoop._attach_loop_start_id, _jump_to_loop_start	On Loop End completion: send the DCM and write the next state
Worker output	OutputManager.reset_output_storage	A Loop End's materialized output accumulates the results of all of its own iterations (a single / outermost loop keeps everything — the scheduler reuses its output doc across re-runs and the writer appends). reset_output_storage exists only for the inner Loop End of a nested loop: it fires once per outer iteration, on the outer-loop pass-through (loop_counter > 0) in _process_state_frame, dropping and recreating that Loop End's result/state iceberg tables so the new outer iteration accumulates from empty instead of concatenating across outer iterations. A single / outermost Loop End never sees loop_counter > 0, so it never resets. Safe because loops run MATERIALIZED — downstream doesn't read until the loop region completes.
Execution mode	WorkflowExecutionService.scala	Reject a loop workflow that is submitted with a non-MATERIALIZED execution mode by throwing an IllegalArgumentException with an actionable message. WorkflowService catches it and routes it through errorHandler → WorkflowFatalError → the existing diff handler, so the frontend sees a clear "switch the mode and re-run" error instead of the UI and the engine silently disagreeing.
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.process_state	If the state already came from a Loop Start (LoopStartStateURI is set), loop_counter += 1 and pass it through.
Inner LoopEnd.process_state	If loop_counter > 0, decrement and pass it through (this state belongs to an outer loop).
Matching LoopEnd	If loop_counter == 0, the state is mine: run update, evaluate condition, jump back to my paired Loop Start.

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.

Any related issues, documentation, discussions?

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

How was this PR tested?

• LoopStartOpDescSpec.scala / LoopEndOpDescSpec.scala — code-gen output, ports, isLoopEnd flag.
• test_loop_operators.py — runtime base classes: flat-loop matching branch (runs update / condition) and nested-loop pass-through (loop_counter +/− symmetric across an outer × inner traversal).
• sbt scalafmtCheckAll scalafixAll --check and ruff check clean.

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 in compliance with ASF.

[aglinxinyuan]

@Xiao-zhen-Liu, please review the PR again.

[Xiao-zhen-Liu]

Suggestion: consider splitting this PR

This is now 38 files / +2280 −94. It's mostly tests (good), but it bundles four separate pieces of work, and one of them — the change to how operator state is stored and sent — affects every stateful operator, not just loops. That part is hard to review alongside the loop feature, because approving the loops also means approving a change to all state handling.

This is only a suggestion. At a minimum I'd split it in two: the state-format change on its own, and the loop feature on top. If you want to go further, it splits cleanly into four:

PR 1  Worker-id helper            (~48 lines)   independent
      get_operator_id + test (replaces the worker-name string parsing)

PR 2  Generalize how state is stored and sent  (~250 lines)
      4-column State schema (Scala + Python) + bytes-in-JSON +
      StateFrame + network sender/receiver + materialization reader

PR 3  Reuse output storage on a region re-run  (~30 lines)
      RegionExecutionCoordinator skip-create + a flag named for the
      behavior + its test

PR 4  Loop operators              (~700 + tests)  builds on 1-3
      LoopStart/EndOpDesc, operator.py loop classes, main_loop logic,
      execution-mode validation, reset_storage, icons, tests

If you only do the two-way split, the state-format change is the part to split out: state written before this PR has the old one-column schema, so it needs backward-compatible reading and a Scala-side round-trip test — neither of which is here yet.

One process note first. Several resolved threads (11, 17, 19, 21, 22) were closed with replies pointing to fix commits (e6bea518f2, e281c61b4c, ca9e5ce8cc, 1848ce00fb, 540b7ba274, bbec98282e) that aren't in the branch (070a844). The code they describe isn't there — it looks lost in a rebase or never pushed. So those threads read as done, but the code still has the original issue. Could you re-push or confirm? I've re-opened the specifics inline.

How to split, given the history can't be rebased apart (300+ commits, many merges): branch off main and copy over one group of files at a time — the four pieces touch almost entirely separate files.

[github-actions]

✅ No material benchmark regressions detected

🟢 13 better · 🔴 0 worse · ⚪ 2 noise (<±5%) · 0 without baseline

CI benchmark results are noisy; treat <±5% as noise unless repeated.

Dashboard · Run

	config	throughput	MB/s	latency	max Δ latest / 7d
🟢	bs=10 sw=10 sl=64	443	0.271	22,282/33,194/33,194 us	🟢 +12.3% / 🟢 +8.1%
🟢	bs=100 sw=10 sl=64	939	0.573	104,394/124,887/124,887 us	🟢 -13.9% / 🟢 -10.5%
🟢	bs=1000 sw=10 sl=64	1,098	0.67	905,236/984,614/984,614 us	🟢 +15.2% / 🟢 -6.6%

Baseline details

Latest main a044287 from 2026-06-13T22:20:02.070Z

config	metric	PR	latest main	7d avg	Δ latest	Δ 7d
bs=10 sw=10 sl=64	throughput	443 tuples/sec	395.36 tuples/sec	410.7 tuples/sec	+12.0%	+7.9%
bs=10 sw=10 sl=64	MB/s	0.271 MB/s	0.241 MB/s	0.251 MB/s	+12.3%	+8.1%
bs=10 sw=10 sl=64	p50	22,282 us	24,519 us	23,798 us	-9.1%	-6.4%
bs=10 sw=10 sl=64	p95	33,194 us	34,935 us	35,169 us	-5.0%	-5.6%
bs=10 sw=10 sl=64	p99	33,194 us	34,935 us	35,169 us	-5.0%	-5.6%
bs=100 sw=10 sl=64	throughput	939 tuples/sec	833.83 tuples/sec	894.84 tuples/sec	+12.6%	+4.9%
bs=100 sw=10 sl=64	MB/s	0.573 MB/s	0.509 MB/s	0.546 MB/s	+12.6%	+4.9%
bs=100 sw=10 sl=64	p50	104,394 us	120,315 us	111,887 us	-13.2%	-6.7%
bs=100 sw=10 sl=64	p95	124,887 us	145,017 us	139,602 us	-13.9%	-10.5%
bs=100 sw=10 sl=64	p99	124,887 us	145,017 us	139,602 us	-13.9%	-10.5%
bs=1000 sw=10 sl=64	throughput	1,098 tuples/sec	953.24 tuples/sec	1,045 tuples/sec	+15.2%	+5.1%
bs=1000 sw=10 sl=64	MB/s	0.67 MB/s	0.582 MB/s	0.638 MB/s	+15.2%	+5.1%
bs=1000 sw=10 sl=64	p50	905,236 us	1,052,108 us	969,370 us	-14.0%	-6.6%
bs=1000 sw=10 sl=64	p95	984,614 us	1,094,010 us	1,019,271 us	-10.0%	-3.4%
bs=1000 sw=10 sl=64	p99	984,614 us	1,094,010 us	1,019,271 us	-10.0%	-3.4%

Raw CSV

config_idx,batch_size,schema_width,string_len,num_batches,total_ms,total_tuples,total_bytes,tuples_per_sec,mb_per_sec,lat_p50_us,lat_p95_us,lat_p99_us
0,10,10,64,20,451.00,200,128000,443,0.271,22281.94,33193.83,33193.83
1,100,10,64,20,2129.84,2000,1280000,939,0.573,104393.57,124887.02,124887.02
2,1000,10,64,20,18220.65,20000,12800000,1098,0.670,905235.65,984613.93,984613.93

[aglinxinyuan]

Superseded by #5700, which re-opens this from my fork (aglinxinyuan/texera) to satisfy the requirement that contributions come from a fork rather than a branch on the main repo. The code is identical and the labels/CI carry over. Continuing on #5700 — the full review discussion here remains for reference. Thanks!