refactor(amber): centralize worker-id parsing in virtual_identity - #5706
Merged
Conversation
Generalize the worker-name regex so the workflow id and operator id are captured explicitly, and add `get_operator_id` to extract the logical operator id from a worker actor name. Both `get_worker_index` and `get_operator_id` use `re.fullmatch`, so a malformed worker id with trailing junk now fails loudly with a `ValueError` instead of parsing silently -- mirroring the Scala `VirtualIdentityUtils.getPhysicalOpId` full-match semantics referenced in the docstring. `get_worker_index` returns the same value for well-formed ids (the worker index is still the final capture group); `get_operator_id` is new and gains its production caller with the loop feature. Split out of the loop-operators PR (apache#5700) to keep that PR reviewable. This helper is independent of the rest of that work and behavior-preserving for well-formed worker ids.
Contributor
There was a problem hiding this comment.
Pull request overview
This PR refactors Amber’s Python virtual_identity utilities to centralize worker actor-name parsing and to introduce a single helper for extracting logical operator IDs from worker IDs, aligning behavior more closely with the Scala parsing semantics.
Changes:
- Generalize
worker_name_patternto capture workflow id + operator id explicitly, and switch parsing tore.fullmatch. - Add
get_operator_id(worker_id)helper that extracts the logical operator id and raisesValueErroron invalid formats. - Add unit tests covering
get_operator_idfor canonical names, hyphenated operator ids, and invalid inputs.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 3 comments.
| File | Description |
|---|---|
amber/src/main/python/core/util/virtual_identity.py |
Updates worker-id regex + parsing strictness; adds get_operator_id; adjusts group indexing. |
amber/src/test/python/core/util/test_virtual_identity.py |
Adds new TestGetOperatorId coverage for operator-id extraction and invalid formats. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #5706 +/- ##
============================================
- Coverage 52.87% 52.84% -0.04%
+ Complexity 2627 2622 -5
============================================
Files 1090 1090
Lines 42176 42182 +6
Branches 4531 4531
============================================
- Hits 22302 22292 -10
- Misses 18569 18579 +10
- Partials 1305 1311 +6
*This pull request uses carry forward flags. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Contributor
|
| config | throughput | MB/s | latency | max Δ latest / 7d | |
|---|---|---|---|---|---|
| 🔴 | bs=10 sw=10 sl=64 | 412 | 0.252 | 22,939/34,665/34,665 us | 🔴 +16.3% / ⚪ within ±5% |
| 🔴 | bs=100 sw=10 sl=64 | 925 | 0.565 | 107,898/133,719/133,719 us | 🔴 +9.5% / ⚪ within ±5% |
| ⚪ | bs=1000 sw=10 sl=64 | 1,082 | 0.66 | 922,319/962,987/962,987 us | ⚪ within ±5% / 🟢 -5.9% |
Baseline details
Latest main b21540d from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---|---|---|---|---|
| bs=10 sw=10 sl=64 | throughput | 412 tuples/sec | 457 tuples/sec | 410.82 tuples/sec | -9.8% | +0.3% |
| bs=10 sw=10 sl=64 | MB/s | 0.252 MB/s | 0.279 MB/s | 0.251 MB/s | -9.7% | +0.5% |
| bs=10 sw=10 sl=64 | p50 | 22,939 us | 19,732 us | 23,785 us | +16.3% | -3.6% |
| bs=10 sw=10 sl=64 | p95 | 34,665 us | 31,351 us | 34,980 us | +10.6% | -0.9% |
| bs=10 sw=10 sl=64 | p99 | 34,665 us | 31,351 us | 34,980 us | +10.6% | -0.9% |
| bs=100 sw=10 sl=64 | throughput | 925 tuples/sec | 1,006 tuples/sec | 891.94 tuples/sec | -8.1% | +3.7% |
| bs=100 sw=10 sl=64 | MB/s | 0.565 MB/s | 0.614 MB/s | 0.544 MB/s | -8.0% | +3.8% |
| bs=100 sw=10 sl=64 | p50 | 107,898 us | 98,574 us | 112,277 us | +9.5% | -3.9% |
| bs=100 sw=10 sl=64 | p95 | 133,719 us | 124,883 us | 139,802 us | +7.1% | -4.4% |
| bs=100 sw=10 sl=64 | p99 | 133,719 us | 124,883 us | 139,802 us | +7.1% | -4.4% |
| bs=1000 sw=10 sl=64 | throughput | 1,082 tuples/sec | 1,127 tuples/sec | 1,041 tuples/sec | -4.0% | +3.9% |
| bs=1000 sw=10 sl=64 | MB/s | 0.66 MB/s | 0.688 MB/s | 0.635 MB/s | -4.1% | +3.9% |
| bs=1000 sw=10 sl=64 | p50 | 922,319 us | 888,800 us | 972,714 us | +3.8% | -5.2% |
| bs=1000 sw=10 sl=64 | p95 | 962,987 us | 938,261 us | 1,023,057 us | +2.6% | -5.9% |
| bs=1000 sw=10 sl=64 | p99 | 962,987 us | 938,261 us | 1,023,057 us | +2.6% | -5.9% |
Raw CSV
config_idx,batch_size,schema_width,string_len,num_batches,total_ms,total_tuples,total_bytes,tuples_per_sec,mb_per_sec,lat_p50_us,lat_p95_us,lat_p99_us
0,10,10,64,20,484.86,200,128000,412,0.252,22939.32,34664.94,34664.94
1,100,10,64,20,2161.81,2000,1280000,925,0.565,107897.77,133718.74,133718.74
2,1000,10,64,20,18487.53,20000,12800000,1082,0.660,922318.91,962987.01,962987.01
…sage Address Copilot review on apache#5706: - Add trailing-junk regression tests for get_worker_index and get_operator_id so the re.match -> re.fullmatch change (whose only observable effect is rejecting a worker id with trailing junk) cannot silently regress. The existing partial-match tests only cover a missing component, which fails under both match and fullmatch. - Include the offending worker_id in get_worker_index's ValueError, matching get_operator_id, for easier debugging and consistency.
Yicong-Huang
approved these changes
Jun 14, 2026
aglinxinyuan
enabled auto-merge
June 14, 2026 05:30
aglinxinyuan
disabled auto-merge
June 14, 2026 05:32
Address Yicong's feedback on apache#5706 — the Python side added `core.util.virtual_identity.get_operator_id` as a one-shot helper for extracting the logical operator id from a worker actor id. Mirror that on the Scala side via `VirtualIdentityUtils.getOperatorId` so both languages have the same API surface. - Add `getOperatorId(workerId: ActorVirtualIdentity): String` as a thin wrapper around `getPhysicalOpId(workerId).logicalOpId.id`. Docstring notes the deliberate semantic difference from the Python sibling (Scala preserves the existing `"__DummyOperator"` sentinel-on-miss behavior so it stays a drop-in replacement for inline call sites; Python raises ValueError for fail-loud semantics). - Migrate the one existing call site in `ErrorUtils.getOperatorFromActorIdOpt` to use the new helper. - Add VirtualIdentityUtilsSpec cases pinning: extraction on a worker actor name, agreement with `getPhysicalOpId(...).logicalOpId.id`, and the sentinel fallback for non-worker actor names.
Address Yicong's follow-up on apache#5706 — the bare name 'operator id' doesn't make the logical-vs-physical distinction visible at the call site. Rename to mirror the existing Scala 'getPhysicalOpId' convention: - Scala: VirtualIdentityUtils.getOperatorId -> getLogicalOpId - Python: core.util.virtual_identity.get_operator_id -> get_logical_op_id Updated all callers (ErrorUtils, both spec files) and the cross-language docstrings.
aglinxinyuan
added a commit
to aglinxinyuan/texera
that referenced
this pull request
Jun 14, 2026
apache#5706) The worker-id helper PR (apache#5706) renamed the helper to get_logical_op_id to make the logical-vs-physical distinction explicit. Apply the same rename here so loop-feb stays internally consistent and rebases cleanly onto main once apache#5706 lands: - virtual_identity.py + its test: brought in line with apache#5706's final versions (rename, fail-loud error message including the worker id, and trailing-junk regression coverage) - main_loop._compute_loop_start_id: call get_logical_op_id - test_main_loop: update the helper name in two explanatory comments
yangzhang75
pushed a commit
to yangzhang75/texera
that referenced
this pull request
Jun 22, 2026
…ache#5706) ### What changes were proposed in this PR? Centralizes the Python worker's worker-id parsing in `core/util/virtual_identity.py`: - Adds `get_operator_id(worker_id)` — extracts the logical operator id from a worker actor name (`Worker:WF<wf>-<op>-<layer>-<idx>`), raising `ValueError` on a malformed id. - Generalizes `worker_name_pattern` to capture the workflow id and operator id explicitly. - Switches both `get_worker_index` and `get_operator_id` to `re.fullmatch`, so a malformed id with trailing junk now fails loudly instead of parsing silently — matching the Scala `VirtualIdentityUtils.getPhysicalOpId` full-match semantics the docstring already claims. | case | before | after | |---|---|---| | `get_worker_index`, well-formed id | worker index | same value | | `get_worker_index`, malformed id (trailing junk) | parsed silently | raises `ValueError` | | `get_operator_id` | — | new helper | Behavior-preserving for well-formed worker ids. `get_operator_id`'s production caller lands with the for-loop feature; the helper and its test are independent and mergeable now. ### Any related issues, documentation, discussions? Resolves apache#5708 (sub-issue of apache#4442 "Introduce for loop"). Split out of apache#5700 to keep that PR reviewable, per @Xiao-zhen-Liu's [review](apache#4206 (review)). ### How was this PR tested? `pytest src/test/python/core/util/test_virtual_identity.py` — 23 passing, covering well-formed ids, the new `get_operator_id`, and malformed ids that now raise `ValueError`. `ruff check`/`format` clean. ### Was this PR authored or co-authored using generative AI tooling? Co-authored with Claude Opus 4.8 in compliance with ASF.
gupta-sahil01
pushed a commit
to gupta-sahil01/texera
that referenced
this pull request
Jul 18, 2026
> _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 [apache#4206](apache#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 apache#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 apache#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 apache#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 apache#4442. Builds on apache#4490 (cross-region state materialization) and apache#5085 (`DocumentFactory.documentExists`). Prerequisites split out of this PR per @Xiao-zhen-Liu's request, all merged: apache#5706 (worker-id helper), apache#5707 (state test harness), **apache#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](https://github.com/user-attachments/files/27985168/Loop.json) | `TextInput → LoopStart → LoopEnd` | 3 iterations, terminates. | | [Nested Loop.json](https://github.com/user-attachments/files/27985169/Nested.Loop.json) | `TextInput → OuterStart → InnerStart → InnerEnd → OuterEnd` | 3 outer × 3 inner = **9** inner iterations, terminates. | Basic Loop: <img width="1715" height="368" alt="loop" src="https://github.com/user-attachments/assets/4b9ea672-b5c9-4392-9ac3-764cc9cbb772" /> Nested Loop: <img width="1715" height="368" alt="nested" src="https://github.com/user-attachments/assets/240ea180-43ca-4815-aa6b-0ff82eb81d7b" /> ### 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. --------- Signed-off-by: Xinyuan Lin <xinyual3@uci.edu> Co-authored-by: Xiaozhen Liu <xiaozl3@uci.edu> Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> Co-authored-by: Meng Wang <mengw15@uci.edu> Co-authored-by: probe <probe@x>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this PR?
Centralizes the Python worker's worker-id parsing in
core/util/virtual_identity.py:get_operator_id(worker_id)— extracts the logical operator id from a worker actor name (Worker:WF<wf>-<op>-<layer>-<idx>), raisingValueErroron a malformed id.worker_name_patternto capture the workflow id and operator id explicitly.get_worker_indexandget_operator_idtore.fullmatch, so a malformed id with trailing junk now fails loudly instead of parsing silently — matching the ScalaVirtualIdentityUtils.getPhysicalOpIdfull-match semantics the docstring already claims.get_worker_index, well-formed idget_worker_index, malformed id (trailing junk)ValueErrorget_operator_idBehavior-preserving for well-formed worker ids.
get_operator_id's production caller lands with the for-loop feature; the helper and its test are independent and mergeable now.Any related issues, documentation, discussions?
Resolves #5708 (sub-issue of #4442 "Introduce for loop"). Split out of #5700 to keep that PR reviewable, per @Xiao-zhen-Liu's review.
How was this PR tested?
pytest src/test/python/core/util/test_virtual_identity.py— 23 passing, covering well-formed ids, the newget_operator_id, and malformed ids that now raiseValueError.ruff check/formatclean.Was this PR authored or co-authored using generative AI tooling?
Co-authored with Claude Opus 4.8 in compliance with ASF.