aglinxinyuan opened a new pull request, #8642:
URL: https://github.com/apache/texera/pull/8642
### What changes were proposed in this PR?
Inside a control block (LoopStart ... LoopEnd), the iteration's loop
variables reach every operator as a state message. Until now a Python operator
saw that message only as the argument of `process_state`. This PR registers the
message on the operator before the callback runs, so `process_state` and every
later call, such as `process_tuple`, can read the current iteration's variables.
```
state message {K: 2} arrives
-> operator.loop_state = State({K: 2}) registered by the worker, for
every operator
-> operator.process_state(state, port) default unchanged: forwards the
state
-> operator.process_tuple(...) can read self.loop_state["K"]
```
| Where the worker hands a state to an operator | Before | After |
|---|---|---|
| `DataProcessor.process_state` | callback only | `loop_state` set first,
then the callback |
| LoopEnd's deferred consume in the main loop, which calls the operator
directly | callback only | `loop_state` set first, then the callback |
| any operator before its first state message | no attribute | `loop_state`
is `None` |
The default `process_state` still returns the state, so pass-through
behavior is unchanged. This is the Python counterpart of the JVM registration
in #8639.
### Any related issues, documentation, discussions?
Closes #8641
### How was this PR tested?
New cases in `test_operator.py` (`loop_state` is `None` until a state
arrives), `test_data_processor.py` (the callback observes `loop_state` already
set and the default still forwards; a later message replaces it; only the
running operator is registered) and `test_main_loop.py` (the LoopEnd's deferred
consume registers the state first).
```
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
# 110 passed
ruff format --check <the 6 changed files> && ruff check <the 6 changed files>
# clean
```
With the three source changes reverted, the same run gives 5 failed and 105
passed, so the new cases pin the change.
### 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]
