aglinxinyuan opened a new pull request, #8639:
URL: https://github.com/apache/texera/pull/8639
### What changes were proposed in this PR?
Inside a control block (LoopStart ... LoopEnd), a property of an operator
may now refer to a loop variable as `$K`: the whole value is a dollar sign plus
a name. The operator's own code does not change. The frontend keeps sending the
literal `"$K"` whatever the property's type.
```
frontend JSON parse compile
worker
{"limit":"$n"} --> {"limit":0, --> inside a block: reference
kept --> placeholder unbound; tuples refused
"stateReferences": outside every block:
state message {n: 3} arrives
{"/limit":"n"}} string ref -> literal,
as today -> {"limit":3} -> real LimitOpExec
typed ref -> compile
error -> opened once, reused for every tuple
(the sidecar rides in the
executor's descString)
```
| Stage | What it does |
|---|---|
| Parse (`StateReferenceModule`, a Jackson module on the shared mapper and
the three Dropwizard mappers) | Turns a `$K` in a numeric or boolean property
into a typed placeholder (0 / 0.0 / false); a string property keeps the
literal. Every match is recorded in a hidden `stateReferences` sidecar on
`LogicalOp` (JSON pointer -> variable name). Objects without a typed
placeholder are read from their own tokens, so their JSON is untouched. |
| Compile (`WorkflowCompiler`, `LoopBlockMembership`) | Decides which
operators lie inside a loop block, with the same walk as the frontend. Inside a
block the sidecar becomes authoritative, including `$name` literals of
descriptors built in Scala. Outside every block, string references are dropped
(they stay literals) and a typed one is a compile error naming the property.
Inside a block, an operator whose executor runs generated code, or whose input
is fed from outside the block, is also a compile error. |
| Worker (`ExecFactory`, `LateBoundExecutor`) | An executor whose descriptor
carries a sidecar is created unbound. The first state message binds each
placeholder, coerced to the property's JSON type, then builds and opens the
real executor, which handles every later call. Values are never rebound. A
tuple before the state, or a variable missing from the state, is an error
naming the property and the variable. |
| Shared registration (`OperatorExecutor.registerState` / `loopState`,
pyamber `Operator.loop_state`) | The worker registers each state message on the
executor before `processState` / `process_state` runs, the same for every
operator. |
| Other entry points | Runtime reconfiguration applies the same rules to the
edited operator and hands the consumed loop state to its replacement, as does
checkpoint restore. The Python export refuses a reference it would otherwise
emit as a placeholder. |
**Behaviour changes**
| Input | Before (main) | After |
|---|---|---|
| `"limit":"$n"` on a JVM operator inside a loop | parse error | bound to
the iteration's `n` |
| whole-string `"$i"` in a string property inside a loop | literal text |
bound to the iteration's `i` |
| whole-string `"$AAPL"` in a string property outside any loop | literal
text | literal text, unchanged |
| `"$n"` in a numeric or boolean property outside any loop | parse error |
compile error: `property /limit refers to loop variable n, but Limit is not
inside a loop block` |
| `"$K"` inside a loop on a UDF or Python-generating operator | parse error
or silent literal | compile error: `<Operator> cannot refer to loop variables
yet` |
| `"$K"` inside a loop on an operator with an input fed from outside the
block | parse error or silent literal | compile error naming that input |
**Known limitations and follow-ups**
- Python-generated operators (sklearn, charts) and UDF properties need
binding on the Python side; for now they are rejected inside a block. UDF code
can already read `self.loop_state`.
- A reference in a property the compiler itself reads (a column name used
for schema propagation or partitioning, Projection's `isDrop`) is not supported
yet. The compiler sees the literal or the placeholder.
- Inside a loop block, a whole-string `$name` in a string property is always
a reference. There is no escape yet.
- Membership is computed on the plan the compiler receives. "Execute up to
operator" and sync execution's "execute to" cut the plan before compilation,
and disabled or invalid operators are left out of it, so an operator whose path
to its LoopEnd is cut counts as outside the block.
### Any related issues, documentation, discussions?
Closes #8635
The frontend half, which lets the property panel accept and autocomplete
`$K`, is #8636 and merges after this PR.
### How was this PR tested?
| Suite | Result |
|---|---|
| WorkflowCompiler full `test`, incl. the new `LoopBlockMembershipSpec` and
the loop-variable cases of `WorkflowCompilerSpec` (lenient and strict) | 89/89 |
| WorkflowCore: `StateReferencingSpec`, `LateBoundExecutorSpec`,
`CoreExecutorReflectionSpec`, `OperatorExecutorSpec`,
`StateReferenceModuleSpec`, `JSONUtilsSpec`, `StateSpec` | 126/126 |
| WorkflowOperator: Filter and Limit desc/exec specs, `LogicalOpSpec`,
`OperatorMetadataGeneratorSpec`, `FilterPredicateSpec` | 64/64 |
| amber: `DataProcessorSpec`, `DataProcessorRPCHandlerInitializerSpec`,
`SerializationManagerSpec`, `PrepareCheckpointHandlerSpec`,
`ExecutionReconfigurationServiceSpec`, `WorkflowWebsocketResourceSpec`,
`TexeraWebApplicationSpec`, `ComputingUnitMasterSpec`, plus `Test/compile` |
121/121 |
| workflow-compiling-service: `WorkflowToPythonResourceSpec`,
`WorkflowCompilationResourceSpec`, `WorkflowToPythonTranslatorSpec`, plus
`Test/compile` | 20/20 |
| pyamber: `test_operator.py`, `test_data_processor.py`, `test_main_loop.py`
| 110 passed |
| Lint | `scalafixAll --check` and `scalafmtCheckAll` on all five touched
sbt projects; `ruff format --check` and `ruff check` on the Python files |
```
sbt "project WorkflowCompiler" test
sbt "project WorkflowCore" "testOnly *StateReferencingSpec
*LateBoundExecutorSpec *CoreExecutorReflectionSpec *OperatorExecutorSpec
*StateReferenceModuleSpec *JSONUtilsSpec *StateSpec"
sbt "project WorkflowExecutionService" Test/compile "testOnly
*DataProcessorSpec *ExecutionReconfigurationServiceSpec
*SerializationManagerSpec *PrepareCheckpointHandlerSpec"
cd amber/src/main/python && python -m pytest
../../test/python/core/models/test_operator.py
../../test/python/core/runnables/test_data_processor.py
../../test/python/core/runnables/test_main_loop.py -q
```
Each case that pins a fix was first run red, against stubs or the pre-fix
code. The tests pin these behaviors:
- **End to end in unit form.** The frontend's
`{"operatorType":"Limit","limit":"$n"}` goes through parse, compile,
`getPhysicalOp` and `ExecFactory` to a real `LimitOpExec`. A tuple sent before
the state is refused, and after the state `{n: 2}` exactly 2 of 5 tuples pass.
- **Outside every block.** A Filter with `"$AAPL"` compiles with an empty
sidecar and keeps exactly the rows equal to `$AAPL`. Limit `"$n"` is a compile
error. Sort `"$USD"` compiles.
- **Inside a block.** Limit `"$n"` keeps its sidecar and is late-bound. A
Scala-built Filter `"$i"` gets its sidecar from the literal scan. Sort
`"$col"`, and an Interval Join fed from outside the block, are compile errors.
- **Membership.** Straight, branching, nested and sequential blocks,
dangling branches, and four cycles through a control operator.
- **Worker.** Never rebinds; coerces per placeholder type; names the
property and variable in every error; never wraps a source.
`LoopIntegrationSpec` gains one `@IntegrationTest` case,
`TextInput("0","1","2") -> LoopStart(i = 0) -> Filter(line = $i) -> LoopEnd`,
asserting 3 materialized rows (a constant `"0"` would give 1). It compiles
locally but was not run, because it spawns Python workers and the authoring
machine is Windows. It runs in the `amber-integration` CI job.
### 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)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
