feat(pyamber): register the state before process_state - #8642

The attribute the runtime registers before process_state is now `state`
(apache#8642), so the method that registers it follows: Operator.register_loop_state
is now Operator.register_state. DataProcessor.process_state and the deferred
consume in MainLoop call it under the new name, and its docstring and the
tests say `state` / `register_state` throughout. The loop-variable reads,
loop_variable_text and loop_variable_value, and their _loop_variables are
unchanged.

### 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
(apache#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 apache#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 apache#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 (apache#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 apache#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 apache#8635

Lands together with apache#8750, which makes `$i` in a setting the compiled
plan is built from a compile error. Frontend: apache#8640 (merges after this
PR). Python operators: apache#8642 (independent). User-facing overview:
apache#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 apache#8642
and apache#8640 (the test branch in apache#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
|
|---|---|
| ![Limit inside the loop holding
$i](https://raw.githubusercontent.com/aglinxinyuan/texera/390db38157491c992db75f25dc703bad05b4ab65/panel-after-inside-loop.png)
| ![Loop End result: 15 rows, first page 1, 1, 2, 1,
2](https://raw.githubusercontent.com/aglinxinyuan/texera/ca12ba98464825cb6d16ba1ba90316fac37ca6ca/loop-end-result-15-rows.png)
|

![The completed run: canvas, Loop End's update and condition, and its
15-row
result](https://raw.githubusercontent.com/aglinxinyuan/texera/ca12ba98464825cb6d16ba1ba90316fac37ca6ca/loop-run-binds-i-per-iteration.png)

### 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)

### What changes were proposed in this PR?

Inside a control block (`Loop Start ... Loop End`), any primitive
property of an operator (string, integer, number, boolean) can now hold
a loop-variable reference such as `$i`. The frontend sends the literal
string `"$i"` unchanged. The backend marks it as a placeholder when it
compiles the plan and binds it when the iteration's state message
reaches the worker (apache#8639). Operators outside every block behave as
before.

```
Loop Start (i = 0) -> Limit(limit: "$i") -> Loop End
                        ^ panel offers $i; canvas valid; sent as "$i"
rename i to j in the Loop Start  ->  Limit warns "$i is not a variable of an enclosing block; ..."; Run stays enabled
delete a link upstream           ->  Limit outside the block -> "must be integer"
disable the Loop End             ->  the same: the backend compiles and runs only enabled operators
```

| Piece | What it does |
|---|---|
| Block membership (`enclosingLoopStarts`, `loop-block.util.ts`) |
Worked out in the frontend from the graph as it stands: every enabled
operator, valid or not, and the links between them, which is the plan
the backend compiles and runs (`WorkflowGraph.getSubDAG`). It is a port
of the backend's `LogicalPlan.enclosingLoopStarts` (apache#8639), with the
same treatment of a Loop Start that no Loop End closes, nested and
sequential blocks, branches, and outermost-first order.
`LogicalPlanSpec` and `loop-block.util.spec.ts` both run one shared
fixture (`common/workflow-compiler/src/test/resources/loop-blocks.json`:
named cases plus random graphs, with the backend's answers), so the two
cannot drift apart, and a change to the fixture runs the frontend stack
too (`labeler.yml`). An unfinished operator elsewhere on the canvas no
longer keeps an operator in a block it has left. |
| Loop variables (`extractLoopVariables`, `loopVariablesInScope`) |
Reads the variables from each enclosing Loop Start's `initialization`
code, following Python's module-level binding: assignments at any depth
outside a `def`/`class` body, for-loop targets, one-line compound bodies
such as `try: k = 1`, and names an import binds (`from math import pi`,
`import numpy as np`, relative modules, parenthesized lists). |
| Property panel | Inside a block, each plain primitive field becomes a
text box that offers `$name` for every variable in scope. So does a
field whose accepted values depend on another field of its row (value
rules, as on the sklearn trainers' hyperparameter values): its rules
still judge every value that is not a reference, and the text box
suggests the values they accept after the variables. Enums, arrays and
custom widgets keep their own controls, and so does a property the
backend marks `"noLoopVariable": true` (apache#8750: settings the plan is
built from, such as Projection's columns), together with everything
nested under it. A `$name` that no enclosing block declares gets a
warning under the field; the operator stays valid, and the run itself
reports the name if no Loop Start sets it. |
| Canvas validation | Inside a block, the schema `type` error is waived
exactly where a reference sits in a primitive-typed property.
`loopScopeChangeStream` emits on every edit that can change an
operator's scope: an operator or link added or deleted, an operator
disabled or enabled, or a Loop Start's properties changed. On each one,
validation re-validates every operator that holds a reference, so the
Run button follows edits two hops away. |
| Open panel | Rebuilt on `loopScopeChangeStream` when the operator's
scope changed. |
| Shared editing (`updateYTypeFromObject`) | A string now replaces a
number or boolean instead of throwing, which storing `"$i"` over a
number needs. |

Panel behavior for a field inside a block:

| Field type | Typed text | Stored value |
|---|---|---|
| integer | `7` | `7` (number) |
| integer / number / boolean | `$i` | `"$i"` |
| number | `0.5`, typed one key at a time | `0.5`; partial text such as
`7.` stays text until complete |
| boolean | `true` / `FALSE` | `true` / `false`; other text is flagged |
| string | `hello` / `$i` | `"hello"` / `"$i"` |
| value rules (e.g. `C`, which must be > 0) | `$i` / `1.0` / `-1` |
`"$i"` / `"1.0"` / `"-1"`, flagged by the rules |
| any | `$foo` (undeclared) | `"$foo"`, with a warning; the operator
stays valid |

**Related.** apache#8639 (on `main`) binds the references and computes the
blocks. apache#8750 (on `main`) adds the `noLoopVariable` marks.

**Screenshots.** The Limit outside the loop keeps the number field it
has on `main`. The Limit inside the loop gets a text box that suggests
the loop's variables and flags a name the loop does not declare. The
third shot predates the change of that flag to a warning, which leaves
Run enabled.

| Before: outside a loop (same as `main`) | After: inside a loop |
After: an undeclared name |
|---|---|---|
| ![Limit outside the loop: number field holding
3](https://raw.githubusercontent.com/aglinxinyuan/texera/390db38157491c992db75f25dc703bad05b4ab65/panel-before-outside-loop.png)
| ![Limit inside the loop: $i with the $i
suggestion](https://raw.githubusercontent.com/aglinxinyuan/texera/390db38157491c992db75f25dc703bad05b4ab65/panel-after-inside-loop.png)
| ![Limit inside the loop: $foo flagged as not a variable of an
enclosing
block](https://raw.githubusercontent.com/aglinxinyuan/texera/390db38157491c992db75f25dc703bad05b4ab65/panel-after-undeclared.png)
|

The workflow in the shots: `Text Input -> Loop Start (i = 0) -> Limit
($i) -> Loop End`, plus a second Limit fed straight from Text Input,
outside the loop.

![Workflow with the in-loop Limit selected and its panel suggesting
$i](https://raw.githubusercontent.com/aglinxinyuan/texera/390db38157491c992db75f25dc703bad05b4ab65/after-inside-loop-suggests.png)

### Any related issues, documentation, discussions?

Closes apache#8636

### How was this PR tested?

New and updated Vitest specs, run from `frontend/`:

| Spec | Covers |
|---|---|
| `LogicalPlanSpec` (backend) | `enclosingLoopStarts` on every case and
random graph of the shared fixture, outermost first. The fixture's Loop
Start and Loop End are read by the backend's JSON type names, so a
renamed type fails here. The walk-based check on named cases and 400
random DAGs stays. |
| `loop-block.util.spec.ts` (new) | The frontend's port on every case
and random graph of the same fixture, and on every operator type it
names. A graph with a cycle, which the backend rejects but the canvas
allows, ends and counts as outside every block. A link to an operator it
was not given is ignored. |
| `loop-variable.util.spec.ts` | Assignment forms; if/else, try/except,
while and for bodies; for targets; one-line compound bodies; import
forms (`from ... import`, aliases, relative modules, parenthesized
lists, `import a.b`, several on one line; `*` binds nothing); def/class
bodies and with/except-as names excluded; Python keywords never taken
for names. `loopVariablesInScope`: outermost first without duplicates,
undefined outside every block, a disabled operator and its links left
out. `loopScopeChangeStream` emits on adds, deletes, disables, enables
and Loop Start property changes, and on nothing else. |
| `loop-variable-field.util.spec.ts` | Complete vs partial numerals
(`7.`, `-0.`, `1e-` kept as text); the parser sets the control without
redrawing the box; the string `""` default; the boolean validator
rejects `maybe`. On a field with value rules: a reference passes the
rules, every other value is still judged by them, the re-validation hook
survives, and `props` stays the one object `templateOptions` names. An
unknown name gives a warning, not a validation error.
`inheritNoLoopVariable` carries a mark to every schema under it, through
each keyword that holds a schema and through `$ref`s, without marking a
definition an unmarked property shares. |
| `loop-variable-input.component.spec.ts` (new) | Rendered through a
real formly form: typing `1.05`, `-0.5` and `7.5` one key at a time
stores the right number and keeps the box as typed; `$i` and integers;
formly attributes reach the input. An unknown name shows the warning.
With value rules: the suggestions add the accepted values of the row's
parameter (empty box, typed prefix in either case, `$` prefix), and a
changed parameter re-judges the value. |
| `validation-workflow.service.spec.ts` | The waiver applies only to
primitive types. An unknown name leaves the operator valid. Canvas state
follows each edit at once: a block opened or closed, and an operator
moved out of its block and back while another operator is invalid. Edits
two hops away are covered too: an upstream link deleted, a block closed
by a link between two other operators, an operator inserted into the
body, and a Loop End disabled and enabled again. Each of these four
fails if validation stops listening to `loopScopeChangeStream`. |
| `operator-property-edit-frame.component.spec.ts` | Primitive fields
converted inside a block; file picker, enum and array keep their
controls; the form rebuilds when a link, a disabled or enabled operator,
or the Loop Start's variables change its scope, and only then. The open
operator leaves its block and comes back while another operator on the
canvas is invalid. A value-rules field takes `$i` inside a block and
keeps its own control outside; with a schema enum or a custom widget it
keeps its control inside too. A `noLoopVariable` property, and a field
nested under one, keeps its control inside a block. Typing `$` into a
number field inside a block keeps the form as it is. |
| `shared-editing.interface.spec.ts` | A string replaces a number in a
Y.Map instead of throwing. |

Each new behavior spec was checked red against the pre-fix code before
the final run.

```
yarn ng test --watch=false --include=<the 7 loop and shared-editing specs>   # 488 passed, 1 skipped (existing it.skip)
yarn ng test --watch=false                                                 # 240 files, 6815 passed, 1 skipped (existing it.skip)
yarn exec tsc --noEmit -p src/tsconfig.app.json   (and src/tsconfig.spec.json)  # clean
yarn exec ngc -p src/tsconfig.app.json --noEmit                            # AOT + strictTemplates clean
yarn prettier-eslint --list-different <changed files> && yarn eslint <changed files>  # clean
sbt "WorkflowCompiler/testOnly *LogicalPlanSpec"                           # 27 passed
sbt scalafmtCheckAll "WorkflowCompiler/scalafixAll --check"                # clean
```

`yarn ng build` did not complete locally, because the worktree's
`node_modules` is a symlink; CI's frontend build passes on macOS, Ubuntu
and Windows.

An earlier revision was checked by hand on a local deployment with apache#8639
and apache#8642 (the test branch in apache#8645): the loop workflow runs, and the
panel behaves as in the screenshots. The changes since (membership from
the whole graph, imports, the warning, `noLoopVariable`) are covered by
the specs above. Manual steps for the reviewer (Chrome, desktop): link
`Loop Start -> Limit -> Loop End` and set the Loop Start initialization
to `i = 0`. Open `Limit`: its integer field is a text box, typing `$`
offers `$i`, and choosing it keeps the operator valid. Typing `$foo`
shows a warning, "$foo is not a variable of an enclosing block; ...",
and Run stays enabled. Delete a link so `Limit` leaves the block, or
disable the Loop End: the field is a number box again and `$i` is a type
error.

### 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)

…o values

Since main's apache#8642, reading a name that got two different values from one
depth fails. bind_loop_variables copies loop_variables with dict(), which reads
every name, so a generated operator fails at its first tuple, or when an input
ends, as soon as any name has two values, even one no property refers to. A JVM
operator fails only for a name its setting refers to. main's
test_main_loop_thread_can_process_state (four values of one name, then
EndChannel) fails the same way.

Pin that binding fails only for a required name, listed with the other
problems, and that the bound values keep the two values: reading that name
fails, the other names read, and a deeper value before binding replaces the
two.