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 operators — Loop Start and Loop End — that let users write a for-loop over the rows of a pandas table inside a visual workflow. 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 between Loop Start and Loop End are the loop body and run once per iteration. While condition is true the runtime starts another iteration; when it turns 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 Loop End → Loop Start arrow is not a graph edge — the region DAG stays acyclic. When an iteration ends, Loop End (1) sends a jump_to_operator_region DCM asking the coordinator to schedule the Loop Start region again, and (2) writes the next iteration's state into the iceberg state channel Loop Start reads from (the same cross-region channel from #4490, reused here as the back-edge). Details in the code comments on MainLoop._jump_to_loop_start and _process_state_frame.

The write address is the same every iteration, so it is config, not per-iteration data: the scheduler resolves each Loop Start's input-port state URI and ships it to workers at setup (InitializeExecutorRequest.loopStartStateUris); a Loop End selects its entry by the loop_start_id stamped on the state it consumes. This is why the State format is only 3 columns (content/loop_counter/loop_start_id, from #5900) — see the proto field doc and WorkflowExecutionManager.loopStartStateUris for the rationale. Nested loops work with no extra machinery: a loop_counter on the state (+1 through an inner Loop Start, −1 through an inner Loop End, consumed at 0) routes each state to its owning loop — see MainLoop._process_state_frame.

State serialization

The loop state = the user's variables + the input table. Variables ride the State content column as JSON (raw bytes base64-encoded); the table is serialized as an Apache Arrow IPC stream, not pickle — the state is persisted to iceberg and read back by another worker, so pickle.loads would be a remote-code-execution surface. Arrow's nested-type slowness doesn't apply here (Texera tables are flat) and the channel is low-volume, so safety wins. See table_to_ipc_bytes in core/models/table.py.

File map

Area	File	Purpose
Descriptors	loop/{LoopOpDesc,LoopStartOpDesc,LoopEndOpDesc}.scala	Code-gen the Python operator from the user expressions; Loop Start sets isLoopStart, Loop End sets output-port reuseStorage
Runtime base	core/models/operator.py	LoopStartOperator / LoopEndOperator the generated code extends; guarded expression eval
Physical plan	PhysicalOp.scala	Compile-time isLoopStart (loop-back resolution) + requiresMaterializedExecution (forces whole-plan MATERIALIZED) markers
Loop-back resolution + delivery	WorkflowExecutionManager.scala, controlcommands.proto, RegionExecutionManager.scala, initialize_executor_handler.py, context.py	Derive {Loop Start id → input-port state URI} from the final schedule and ship it to workers at setup
Worker runtime	MainLoop (_process_state_frame, _jump_to_loop_start)	loop_counter bookkeeping, nested pass-through, and the jump-DCM + state-write on Loop End completion
Storage reuse	OutputPort.reuseStorage + RegionExecutionManager; OutputManager.reset_output_storage	A Loop End accumulates across its own iterations; an inner nested Loop End resets once per outer iteration
Frontend	LoopStart.png, LoopEnd.png	Operator icons

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 merged: #5706 (worker-id helper), #5707 (state test harness), #5900 (the 3-column State format, dormant on main until this PR activates it).

How was this PR tested?

• Unit — test_main_loop.py (loop-counter routing, nested pass-through, the full _jump_to_loop_start contract incl. DCM-before-write ordering and the fail-loud missing-URI case, and complete()'s error reporting for condition / loop-back-write failures); test_loop_operators.py (operator base classes, guarded eval, the generated-code shape via a base64 round-trip, reserved-table collision raising, a multi-iteration loop to completion); test_output_manager.py (reset_output_storage contract); test_initialize_executor_handler.py (proto field → Context); LoopStartOpDescSpec / LoopEndOpDescSpec (codegen, ports, JSON round-trip, flag pins).
• Integration (@IntegrationTest, CI) — LoopIntegrationSpec: single loop (3 iterations) and nested 3×3 loop (9 inner iterations), asserted via materialized iceberg row counts.
• scalafmtCheckAll + scalafixAll --check and Python ruff format clean; full Python unit suite passes.

Manual workflows

Input for both is a 3-row table from TextInput("1\n2\n3"); each condition is i < len(table).

Workflow	Topology	Expected
Loop.json	TextInput → LoopStart → LoopEnd	3 iterations, terminates.
Nested Loop.json	TextInput → OuterStart → InnerStart → InnerEnd → OuterEnd	3 outer × 3 inner = 9 inner iterations, terminates.

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.

[Yicong-Huang]

ok, as a related context, we are opening PRs to improve nested data types deserialization in pyarrow. please take this into considerations.

[chenlica]

Thanks @aglinxinyuan . Do we want to embed this design decision in this PR or do we want to create a separate discussion?

[aglinxinyuan]

Thanks @aglinxinyuan . Do we want to embed this design decision in this PR or do we want to create a separate discussion?

I think we can keep the design here because it's an implementation choice.

[Yicong-Huang]

Thanks @aglinxinyuan . Do we want to embed this design decision in this PR or do we want to create a separate discussion?

I think we can keep the design here because it's an implementation choice.

you can keep it on PR description, PR comments will be collapsed and no one will read them. PR description can become squashed commit message once merged, so it will be retained.

[chenlica]

Thanks @aglinxinyuan . Do we want to embed this design decision in this PR or do we want to create a separate discussion?

I think we can keep the design here because it's an implementation choice.

you can keep it on PR description, PR comments will be collapsed and no one will read them. PR description can become squashed commit message once merged, so it will be retained.

Agreed. I believe this design decision should be more visible on GitHub, not embedded deep in the comments.

[Xiao-zhen-Liu]

Reviewed against the full #4206 history. This is much improved — I checked each prior concern against the actual code (not just the replies), and almost all are genuinely fixed: worker-id parsing, the pickle-to-Arrow switch, codegen escaping, the name-mangled access, reserved-key isolation, and the scheduler flag coupling. The URI-in-State design question also got resolved: State is 3 columns and the write URI is shipped to workers at setup, so no address is stored in the data.

On size: the diff is large (~2.3k lines), but about 70% of that is tests — the source is only ~690 lines, and the separable pieces already went out as #5900 / #5706 / #5707. What's left is one coherent feature, so I don't think it needs a further split.

Approving. The comments below are non-blocking: two design decisions worth making explicitly (the per-iteration re-scheduling and worker recreation, and the silent mode coercion), one concrete test bug (the specId collision), a codegen-test gap, and a few cleanups.