The GitHub Actions job "Backport Approval Check" on texera.git/gh-readonly-queue/main/pr-8639-b7993fbafac472d0ef1b87d86169978c42343ac4 has succeeded. Run started by GitHub user aglinxinyuan (triggered by aglinxinyuan).
Head commit for run: d7ce3a6bff97a43a380ce71ff48f3bc016b261ae / Xinyuan Lin <[email protected]> feat(amber): bind $i loop-variable references in operators (#8639) ### What changes were proposed in this PR? Inside a loop block (LoopStart ... LoopEnd), a property of a JVM operator may now hold `$i`, a reference to the loop variable `i`. This works for string and numeric/boolean properties alike; the frontend (#8640) sends the literal `"$i"`, and the operator's own code does not change. The reference is handled in three places: - **Parse.** When `"$i"` cannot be converted to a numeric or boolean property's type, a Jackson problem handler (`StateReferenceModule`) puts in a typed placeholder (0 / 0.0 / false) and records the property's JSON pointer in a hidden `stateReferences` sidecar on `LogicalOp`. Strings are left alone. - **Compile.** The compiler works out which operators sit inside a loop block (`LogicalPlan.enclosingLoopStarts`: each operator's Loop Starts, from folds over the plan's topological order). Inside a block it also records every whole-string `"$i"` literal. Outside every block, a typed reference is a compile error. - **Worker.** The operator's own executor is built and opened as usual, with the placeholders in its setting: the descriptor its constructor parses. The worker registers every state message on the executor right before `processState` (`OperatorExecutor.registerState` and `state`, the JVM counterpart of Python's `self.state` in #8642). Until the operator first sees data, each referenced variable that a message carries is written into that setting, in place. Right before the first tuple, or the end of input, the worker checks that every reference has a value, and the setting is not written after that. Copies of one loop's state that disagree are an error; an inner loop's value shadows an outer one's (the worker passes each message's nesting depth). **Which settings can hold `$i`** (inside a loop block) | Setting | Holds `$i`? | What the user sees | |---|---|---| | Text read while the operator processes rows (a Filter's value or column, ...) | Yes | The iteration's value, written into the setting before the operator's first row | | Number, integer or decimal (Limit's `limit`, Interval Join's constant, ...) | Yes | The iteration's value, converted to the setting's type. Before the run, the compiler sees the placeholder `0` / `0.0` | | True/false | Yes | Same as a number, with the placeholder `false` | | A setting the compiled plan is built from: an output column's name or type, or what an input is partitioned on (Projection's columns and Drop Option, Unnest String's and Dictionary Matcher's result column, Aggregate, join keys, Type Casting, Sort Partitions) | No | With #8750, a compile error: `Projection cannot refer to loop variables (/isDrop -> $drop): its output schema or partitioning is built from that property when the workflow is compiled, before the loop runs`. With this PR alone, a wrong result or the operator's own error (Projection's missing-column message) | | Any setting of an operator that runs generated or user code: Python UDFs (code editor included), sklearn, Python charts, Sort | No | Compile error: `Sort cannot refer to loop variables yet (/attributes/0/attribute -> $col): its code is generated before the loop runs` | | A dropdown of fixed choices (an enum, e.g. a Filter's condition) | No; the panel (#8640) doesn't offer `$i` there | If it is sent anyway (e.g. through `SyncExecutionResource`), the workflow fails to parse with Jackson's `not one of the values accepted for Enum class` | **Other cases** (replaces the current Case / Behaviour table) | Case | Behaviour | |---|---| | Outside every loop block | A text `"$AAPL"` stays a literal, as on main. A number or true/false `"$n"` is a compile error: `Limit refers to loop variables (/limit -> $n), but is not inside a loop block` | | An input the operator reads first is fed, even by a single link, from outside one of the operator's loop blocks: from outside the loop, from an outer loop's body, or from a sibling block | Compile error: `Interval Join refers to loop variables (/constant -> $i), but its input 'left table' is fed from outside the loop block, so tuples from there can arrive before the loop state`. An input read only after one that carries the state (a join's probe side) may come from anywhere | | A name no enclosing Loop Start assigns (the panel flags it; `SyncExecutionResource` doesn't) | Fails at the operator's first row, or at its finish if it gets no rows: `property /limit refers to loop variable n, but no state message carried it` | | A value that doesn't fit the setting | Fails at the same point: `property /limit refers to loop variable n, but its value abc does not fit it: ...` | | Nested loops that name the same variable | The inner loop's value wins | | Copies of one loop's state that reach an operator disagree (e.g. a UDF changed `i` on one branch, and the branches meet) | Fails: `loop variable i got two different values in one iteration, 1 and 2: a loop's variables must not change inside its body`. On a single path there is only one copy, so the operators after the UDF get its changed value | | Editing the operator while the workflow runs | Refused, with #8750 | Which operators are inside a block is decided on the plan the compiler receives. That plan has no disabled operators, and for run-to-operator it keeps only the target's upstream. The panel decides on the whole canvas. **The operator's lifecycle is unchanged.** The value is set in the operator's setting when the state arrives, so the operator runs exactly as it does outside a loop: ``` worker starts state messages arrive first tuple executor built and opened, registered on the executor (state), and every reference has a value, or the operator placeholders in its setting each variable written into the setting, fails naming the missing ones; in place (a later message wins) the setting is not written after this ``` `open()` runs at worker start, and `produceStateOnStart` and `processState` run on the operator itself, with what they return used. Three executors used to read their setting before a state can arrive, and now read it at first use: Reservoir Sampling in its constructor, and Split and Dictionary Matcher in `open()`. Every other JVM executor already reads its setting while it processes data. ### Any related issues, documentation, discussions? Closes #8635 Lands together with #8750, which makes `$i` in a setting the compiled plan is built from a compile error. Frontend: #8640 (merges after this PR). Python operators: #8642 (independent). User-facing overview: https://github.com/apache/texera/discussions/8643 ### How was this PR tested? | Suite | Result | |---|---| | WorkflowCore: `org.apache.texera.amber.core.state.*` and `core.executor.*`, incl. `StateReferenceModuleSpec`, `StateReferenceBindingSpec`, `CoreExecutorReflectionSpec` | 71/71 | | WorkflowCompiler full `test`, incl. `LogicalPlanSpec` (loop-block membership and enclosing Loop Starts, checked against a port of the panel's walk on every case and 400 random plans) and the loop-variable cases of `WorkflowCompilerSpec` (lenient and strict) | 80/80 | | WorkflowOperator: Reservoir Sampling, Split, Dictionary Matcher and metadata specs | 260/260 | | workflow-compiling-service `test` | 55/55 | | amber: `Test/compile`, `DataProcessorSpec`, `DPThreadSpec`, the worker's promise handlers incl. `EndChannelHandlerSpec`, `ComputingUnitMasterSpec` | 116/116 | | Lint | `scalafixAll --check` and `scalafmtCheckAll` on all five touched sbt projects | ``` sbt "project WorkflowCore" "testOnly *StateReferenceModuleSpec *StateReferenceBindingSpec *CoreExecutorReflectionSpec" sbt "project WorkflowCompiler" test ``` What the tests pin: - **End to end in unit form.** The frontend's `{"limit":"$n","operatorType":"Limit"}` is compiled inside `TextInput -> LoopStart -> Limit -> LoopEnd` and built through `ExecFactory`. After the state `{n: 2}`, exactly 2 of 5 tuples pass. - **Parse handovers.** Typed placeholders get the right pointers whether the object is parsed directly, with the type id first, in the middle or last, or nested in another document. - **Compile rules.** Outside every block, Filter `"$AAPL"` stays a literal and Limit `"$n"` is a compile error. Inside a block, a Scala-built Filter `"$i"` gets its sidecar, while Sort `"$col"` is a compile error, and so is an operator with a reference whose read-first input has a link from outside one of its loop blocks (outside the loop, an outer loop's body, a sibling block, or one outside link next to an inside one). - **Worker.** The executor is built and opened at worker start. `produceStateOnStart` and `processState` run on the operator itself, and what `processState` returns is emitted. Each state message writes the variables into the setting in place, into every property that names them, top-level and nested pointers alike, keeping the object the executor holds; an outer `{i: 1}` followed by an inner `{i: 5}` leaves 5. A tuple before any state fails, naming only the missing variables, and the setting is not written after the first tuple; an operator with no rows fails the same way at its finish. Copies of one loop's state with different values fail, equal ones pass, and a deeper loop's value replaces an outer one's in either arrival order. Values are coerced to the property's type, and a null, a list or a map for a string property is refused. - **Executors.** Reservoir Sampling, Split and Dictionary Matcher use the value written after construction and `open()`. - **Mappers.** Every Dropwizard application that parses workflows registers the module through `JSONUtils.registerWorkflowModules`; `WorkflowCompilingServiceRunSpec` and `ComputingUnitMasterSpec` parse a typed `"$n"` through each bootstrap mapper. `LoopIntegrationSpec` gains three `@IntegrationTest` runs, which CI's `amber-integration` job runs: - a number setting: `Limit($n)` with n = 1, 2, 3 gives 1 + 2 + 3 = 6 rows; - nested loops that both name `i`: `Filter(line = $i)` in the inner body passes all 2 x 2 rows only with the inner `i`; - a text setting on two branches meeting at a Union before `Filter(line = $i)`: 2 copies x 3 iterations = 6 rows (a constant or a `$i` that never rebinds gives 2). **Run by hand.** On a local deployment of this PR together with #8642 and #8640 (the test branch in #8645), a workflow `Text Input (1..6) -> Loop Start (i = 0, output table) -> Limit ($i) -> Loop End (i += 1 while i < len(table))` runs to completion. Each iteration binds `$i` to its own `i`, so Loop End collects 0 + 1 + 2 + 3 + 4 + 5 = 15 rows: `1`, then `1, 2`, then `1, 2, 3`, and so on. The Limit fed straight from Text Input, outside the loop, keeps its constant 3. | The Limit's property inside the loop | Loop End's result after the run | |---|---| |  |  |  ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Fable 5.1, Claude Opus 5.5) 🤖 Generated with [Claude Code](https://claude.com/claude-code) Report URL: https://github.com/apache/texera/actions/runs/37107138036 With regards, GitHub Actions via GitBox
