Proto commits in Texera/texera

These commits are when the Protocol Buffers files have changed: (only the last 100 relevant commits are shown)

2026-08-11

refactor(amber): read the loop input table from its materialization instead of shipping it in state (#6971) ### What changes were proposed in this PR? The loop's input table used to ride **inside the State content**: LoopStart encoded its buffered input as Arrow IPC bytes, base64'd into the JSON `content` column, and that payload was re-written and re-read at **every loop-body hop, every iteration**. That data already exists. In the fully-materialized mode loops require, the Loop Start's input-port materialization holds exactly the loop's input table for the whole loop — Loop Start re-reads it every iteration, and the back-edge truncates only the *state* doc at the same base URI, never the *result* doc. So this PR ships the port's **base URI** in the setup config and derives both addresses from it: ``` loopStartPortUris[LoopStart-id] = <base URI of LoopStart's input port> ├── state_uri(base) → back-edge write address (as before, derived) └── result_uri(base) → the loop's input table (NEW: read at EndChannel) ``` | Piece | Before | After | |---|---|---| | proto field 4 | `loopStartStateUris` = state URI | `loopStartPortUris` = base URI (renamed so the semantic change is loud) | | LoopStart's produced state | user vars + IPC-encoded table (base64 in JSON) | user vars only — small, pure JSON | | Loop-body hops | fat state re-materialized per hop, per iteration | tiny state | | LoopEnd's table | decoded from state content | read once per iteration from `result_uri(base)` at EndChannel, injected via a new runtime-only `attach_loop_table` hook | | `table_to_ipc_bytes` / `table_from_ipc_bytes` | second, divergent Arrow codec (lossy `from_pandas` inference) | deleted — the read goes through the canonical iceberg reader | **When the read happens matters.** The read is issued at **EndChannel** (the matching state is stashed at consume and the operator's update runs in `complete()`), not at consume time. At consume time this worker's own materialization reader is still streaming, and issuing a second iceberg/S3 read from the main loop thread in that window made the reader fail with S3 `Access Denied` — `LoopIntegrationSpec` hung to the CI job timeout on both OSes. Deferring past the reader removes the overlap: integration went from a 20-minute cancel to green in ~9 minutes. The matching consume emits no state downstream, so moving it is unobservable outside the operator. Semantics deliberately preserved: - the reserved-`table` collision raise stays (a user var named `table` would now be *silently shadowed* by the injected table — worse than before); - the "consumed" marker (`_loop_table`) is still set only by a **successful** `run_update`, so `condition()`'s short-circuit for pass-through-only Loop Ends is unchanged; - nested loops work by construction: the inner Loop Start's entry points at the outer Loop Start's output port, whose result doc is recreated per *outer* iteration but persists across *inner* iterations (the jump rewinds to the inner level only). Wins: no ~33% base64 bloat, no JSON-column size ceiling on the table (large-table loops become viable), strictly less I/O for any non-empty loop body (one read per iteration replaces N state-doc writes+reads per hop), and one Arrow codec instead of two. Note: this deepens the read-side use of `storagePairs.head._1` — the same shared upstream URI as the known back-edge fan-out design discussion; if that ever moves to a per-loop private doc, this read moves with it. ### Any related issues, documentation, discussions? Builds on #5900 (State columns) and #6661 (envelope through JVM hops). Related design context: #6660. ### How was this PR tested? - **Unit** — `test_loop_operators.py` rewritten for the attach-based flow plus new pins: the produced state carries no `table`; attaching alone does **not** mark the loop consumed; `run_update` fails loud when no table was attached. `test_main_loop.py` pins the base-URI derivation for the back-edge write (`state_uri(base)`), the missing-config fail-loud, and that the matching consume stashes the state without touching storage, with the read + update happening once at EndChannel. `test_initialize_executor_handler.py` covers the renamed proto field. 237 tests green locally (the only failures in a full sweep are pre-existing environment ones, identical on unmodified main). - **Scala** — full test-compile (proto regen included), `scalafmtCheckAll`, `scalafixAll --check`, and the worker/descriptor spec suites (`WorkerSpec`, `WorkflowWorkerSpec`, `SerializationManagerSpec`, `WorkflowExecutionManagerSpec`, `LoopStartOpDescSpec`, `LoopEndOpDescSpec`) all pass on Java 17. - **E2E** — the four `LoopIntegrationSpec` cases (single, nested 3×3, JVM chain, nested JVM chain) exercise the full read path in the `amber-integration` CI job (both jobs green in ~9 min); the nested cases specifically cover the inner-loop read against the outer Loop Start's per-outer-iteration output doc. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Fable 5)

The documentation is generated from this commit.

2026-07-30

fix(amber, v1.2): order worker state by version, not timestamp (#7106) ### What changes were proposed in this PR? Backport of #6011 to `release/v1.2`, cherry-picked from f5217504c3cfd8ac1d56f6617d7849895af17f59. Follows the Direct Backport Push convention; opened as a PR (rather than a direct push) as part of a backport-coverage audit for fixes merged to `main` since early June that were never labeled for backport. ### Any related issues, documentation, discussions? Backport of #6011. Originally linked #6010. ### How was this PR tested? Release-branch CI runs once the conflicts are resolved and this PR is marked ready for review. The cherry-pick **conflicted** and was committed with conflict markers. ### Was this PR authored or co-authored using generative AI tooling? Yes — backport prepared with Claude Code (mechanical cherry-pick; conflicts left as markers for the original author to resolve; the change itself is #6011 by its original author). Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

fix(amber, v1.2): advance region executions in a later coordinator round, not inside portCompleted (#7096) ### What changes were proposed in this PR? Automated backport of #6960 to `release/v1.2`. Source: 226614b527bf51a010355778440e1972ed3130a7 · [automation run](https://github.com/apache/texera/actions/runs/30506845994) ### Any related issues, documentation, discussions? Backport of #6960. ### How was this PR tested? Release-branch CI runs on this branch once the conflicts are resolved and this PR is marked ready for review. ### Was this PR authored or co-authored using generative AI tooling? No. Co-authored-by: Xinyuan Lin <xinyual3@uci.edu>

fix(amber): advance region executions in a later coordinator round, not inside portCompleted (#6960) ### What changes were proposed in this PR? **Root cause of the `LoopIntegrationSpec` retry storm.** `portCompleted`'s handler advanced region executions inline in its own continuation. That advance terminates a completed region, which sends `EndWorker` to each of its workers — so `EndWorker` went out *before* the reply to the very `portCompleted` being handled, and both travel the same FIFO control channel to that worker: ``` Before — all inside one handler round: advanceRegionExecutions() ──> EndWorker to W (seq N on W's control channel) return EmptyReturn() ──> ReturnInvocation to W (seq N+1, same channel) W processes EndWorker while the reply is still queued behind it: EndHandler: queue not empty => IllegalStateException("worker still has unprocessed messages") => attempt 1 of 150 fails => 2 ERRORs + 2 WARNs + 2 stack traces => 200 ms retry succeeds ``` The check's premise — that `EndWorker` is the last message a worker receives — was being violated by the coordinator itself on every teardown. Whether the worker actually observed the trailing reply was a microsecond race, hence the ~1-in-10 flake. **Fix: advance in a later coordinator round instead of inline.** A new coordinator-to-coordinator RPC carries the advance: ``` After: portCompleted round: bookkeeping; send CoordinatorInitiateAdvanceRegionExecutions to self return EmptyReturn() ──> ReturnInvocation to W later round: advance runs ──> EndWorker to W ``` A message the coordinator addresses to itself is transmitted and then received before it is handled, so it lands a whole message behind every reply the requesting round still owed — `EndWorker` can no longer overtake a reply, and the first termination attempt succeeds. **Second half: the worker tolerates a reply it cannot be ordered against.** Deferring orders the replies the coordinator has *already produced*, but region completion keys only on **port** completion, so the advance fires as soon as the last `portCompleted` is booked — while the same worker's `workerExecutionCompleted` (emitted right after it) may still be queued at the coordinator. Since `NetworkInputGateway.tryPickControlChannel` selects channels out of a `HashMap`, there is no cross-channel order: the advance can run first and that reply be emitted afterwards, landing behind `EndWorker`. Ordering cannot close this, so `EndWorker` no longer counts a queued `ReturnInvocation` as unprocessed work — processing one only fulfills a promise for a request the worker already issued, and all four worker→coordinator calls discard their futures, so nothing is pending on it. Every other queued element still fails the request, so the coordinator retries and the worker drains it first. This PR is JVM-only: the Python worker keeps its current check, which fails `EndWorker` whenever anything is queued, and the coordinator's retry covers it exactly as it does today. Bringing it to parity needs a way to classify what is queued without inspecting the next message, which means changing `InternalQueue` — left for a follow-up rather than bundled here. | | Before | After | |---|---|---| | Coordinator send order | `EndWorker`, then the `portCompleted` reply | reply, then `EndWorker` | | First `EndWorker` attempt | fails on a benign race (storm ~1 run in 10) | succeeds | | Normal teardown logs | 2 ERROR + 2 WARN + 2 stack traces | none | | Worker-side check | any queued message fails | a queued **`ReturnInvocation`** no longer fails; all real work still does | | Bookkeeping failure | `ControlError` to the reporting worker | unchanged | | Advance failure | `FatalError` to the client | unchanged | Changes: - **`coordinatorservice.proto`**: adds `CoordinatorInitiateAdvanceRegionExecutions`, alongside the existing self-directed `CoordinatorInitiateQueryStatistics`. - **`AdvanceRegionExecutionsHandler`** (new): handles that self-addressed request by running the advance, and reports failures to the client exactly as the inline call did. `advanceRegionExecutions`'s scaladoc now tells callers handling a worker-initiated request to go through it. - **`PortCompletedHandler`**: keeps its bookkeeping and its reply position; only the advance moves. - **`RegionExecutionManager`**: corrects the scaladoc that claimed `EndWorker` "will be the last message each worker receives". `StartWorkflowHandler` still advances directly: its caller is the client, it awaits the future to answer with the workflow state, and no `EndWorker` is involved at startup. ### Any related issues, documentation, discussions? #6891. ### How was this PR tested? - **`PortCompletedHandlerSpec` (new)**: drives a real `CoordinatorProcessor` through the real RPC server with a captured output gateway, over a single-region schedule already in flight. Pins that the advance leaves as a separate coordinator-addressed control message; that the *only* thing sent to the reporting worker while handling `portCompleted` is its reply (the direct regression guard — on `main` this list also contains `EndWorker`); that the port is recorded completed before the request goes out, since the advance now reads that state in a later round; that a port belonging to no executing region requests nothing; and that a failed continuation still returns a `ControlError` to the reporting worker. - Baseline check: 3 of those tests fail against `main`'s handler, while the `ControlError` test passes on **both** `main` and this branch — which is what verifies the worker-facing error contract is preserved rather than merely asserted. - Untouched and green, as evidence termination semantics are unchanged: `RegionExecutionManagerSpec`, `WorkflowExecutionManagerSpec`, `EndHandlerSpec`, `CoordinatorSpec`, `TrivialControlSpec`, `AsyncRPCClientSpec`. - Lint: `scalafmtCheck`, `scalafixAll --check` pass. - Loop confirmation is CI-side and statistical: `amber-integration` teardown logs should contain zero `Failed to terminate region ... on attempt 1` lines (runs without this fix show them repeatedly). ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Fable 5)

2026-07-26

fix(amber): order worker state by version, not timestamp (#6011) ### What changes were proposed in this PR? A fast **source** operator stayed **orange (RUNNING)** in the editor after the run finished (issue #6010). **Why it's a bug.** The controller reconstructs each worker's state from several unordered channels (source `RUNNING` via the `startWorker` response, non-source `RUNNING` via `workerStateUpdated`, `COMPLETED`/`PAUSED` via `queryStatistics`/`pauseWorker` responses) and reconciled them with **last-`System.nanoTime()`-wins** in `WorkerExecution`. Worker state, however, is single-writer and strictly ordered causally. For a tiny source the run finishes almost instantly, so the `startWorker` response — carrying the stale `RUNNING` it sampled at launch — can reach the controller *after* `COMPLETED` was recorded; its later receipt timestamp wins and clobbers `COMPLETED`. Results render fine (separate path), so only the border is stuck. This PR orders worker state **causally** instead of by wall clock: | Before | After | | --- | --- | | `WorkerExecution.update(nanoTime, state)` — last-write-wins by receipt time | `updateState(version, state)` — newest **logical version** wins | | stale late RUNNING clobbers COMPLETED | RUNNING (v2) < COMPLETED (v3) ⇒ COMPLETED kept | | — | terminal states (COMPLETED/TERMINATED) are absorbing | - `WorkerStateManager` bumps a monotonic `stateVersion` on every `transitTo` (its state-machine logical clock). - The version rides on every state report: `WorkerStateResponse`, `WorkerStateUpdatedRequest`, `WorkerMetrics` (3 new proto fields). - The controller applies a state only when its version is newer. Stats stay timestamp-ordered (monotonic snapshots can share a state version). - A single per-worker counter ⇒ no cross-process clock-sync assumptions (why not worker timestamps). - **pyamber parity:** the Python worker also reports state (start/pause/resume + query-statistics). Its `StateManager` now bumps the same monotonic version and the four handlers include it; otherwise version-0 reports would be dropped after the first and reconfiguration (RUNNING→PAUSED→RUNNING) would hang. ### Any related issues, documentation, discussions? Closes #6010. The timestamp-based `update` was introduced in #3557. ### How was this PR tested? JDK 17. Scala unit + Scala/Python integration + Python unit: ``` sbt "WorkflowExecutionService/testOnly \ *WorkerExecutionSpec *WorkerStateManagerSpec *OperatorExecutionSpec \ *RegionExecutionCoordinatorSpec *WorkflowExecutionCoordinatorSpec \ *RegionExecutionSpec *WorkflowExecutionSpec" # Scala-Python reconfiguration end-to-end (spawns Python UDF workers): sbt "WorkflowExecutionService/testOnly *ReconfigurationIntegrationSpec" # 3 passed # Python unit: cd amber && pytest src/test/python/core/architecture/managers/test_state_manager.py # 9 passed ``` - `WorkerExecutionSpec`: version ordering (positive + stale/equal-version negatives), terminal-state absorption, `forceTerminate`, independent stats-vs-state ordering, and a named regression for #6010 (`COMPLETED` survives a late `RUNNING`). Verified the regression goes **red** when `updateState` is reverted to last-write-wins. - `WorkerStateManagerSpec` / `test_state_manager.py`: version starts at 0, bumps per successful transition, no bump on no-op self-transition or rejected transition. - `ReconfigurationIntegrationSpec` reproduced the failure (timeout) before the pyamber fix and passes after. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Opus 4.8 (1M context), via Claude Code --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

2026-07-18

feat: add loop operators (#5700) > _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 [#4206](https://github.com/apache/texera/pull/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 #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 #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 #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 #4442. Builds on #4490 (cross-region state materialization) and #5085 (`DocumentFactory.documentExists`). Prerequisites split out of this PR per @Xiao-zhen-Liu's request, all merged: #5706 (worker-id helper), #5707 (state test harness), **#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>

2026-07-05

refactor(engine): remove dead CacheSourceOpExec and OpExecSource plumbing (#6126) ### What changes were proposed in this PR? Remove the dead `CacheSourceOpExec` executor and its `OpExecSource` plumbing. Cache source operators were replaced by input-port materialization reader threads in #3425 (the symmetric follow-up to sink-operator removal in #3312). The scheduler no longer inserts a cache-source read op for a materialized link — it tags the downstream input port with the URI directly: ``` Before (pre-#3425): upstream OutputPort(URI) ──► [CacheSourceOpExec] ──► downstream InputPort After (#3425+): upstream OutputPort(URI) ─────────────────────────► downstream InputPort(URI) ``` Since then `CacheSourceOpExec` has been unreachable: `SpecialPhysicalOpFactory.newSourcePhysicalOp` — the only producer of an `OpExecSource` init-info, which is what constructed `CacheSourceOpExec` in `DataProcessorRPCHandlerInitializer.setupExecutor` — has no production caller (its only references were its own spec). Removed: | Item | Location | | --- | --- | | `CacheSourceOpExec` | `common/workflow-operator/.../source/cache/CacheSourceOpExec.scala` | | `SpecialPhysicalOpFactory` (+ spec) | `common/workflow-operator/.../operator/SpecialPhysicalOpFactory.scala` | | `OpExecSource` match arm + now-unused imports | `amber/.../worker/DataProcessorRPCHandlerInitializer.scala` | | `OpExecSource` message + `opExecSource = 3` oneof field | `common/workflow-core/.../protobuf/.../executor.proto` | The `OpExecInitInfo` oneof is an internal controller↔worker wire format (built and consumed within the same deployment; `preserve_unknown_fields: false`), so dropping the field is safe. The now-unused `virtualidentity.proto` import is dropped. ### Any related issues, documentation, discussions? Closes #6125. The dead code was discovered during review of the test-coverage PR #6098. **Depends on #6132** (multi-region e2e integration test). Per review feedback, the multi-region path — which is exactly what the removed cache sources used to serve — gets end-to-end coverage **first** in #6132; this removal then rides on top, and its CI re-runs that test to prove the removal doesn't regress the path. Merge #6132 before this PR. ### How was this PR tested? No behavioral change — pure dead-code removal. - `sbt "WorkflowOperator/compile" "WorkflowExecutionService/compile"` — green (proto regenerates; the `OpExecInitInfo` match remains exhaustive without the `OpExecSource` case) - `sbt "WorkflowExecutionService/scalafmtCheck" "WorkflowExecutionService/scalafixAll --check"` — clean - The multi-region e2e test in #6132 exercises the materialization read/write path end-to-end; once that merges, this branch is rebased on it. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 4.8 [1M context])

refactor(amber): rename Controller to Coordinator (#6124) ### What changes were proposed in this PR? Renames the Amber master actor from **Controller** to **Coordinator** across the whole engine — classes, variables, comments, protobuf, and tests: | Area | Change | | --- | --- | | Scala package | `architecture.controller` → `architecture.coordinator` (dirs moved with `git mv`) | | Scala classes | `Controller` → `Coordinator`, `ControllerProcessor` → `CoordinatorProcessor`, `ControllerConfig` → `CoordinatorConfig`, `ControllerAsyncRPCHandlerInitializer`, `ControllerTimerService`, `ControllerSpec` → `Coordinator*` | | gRPC proto | `controllerservice.proto` → `coordinatorservice.proto`; `service ControllerService` → `CoordinatorService`; rpc `ControllerInitiateQueryStatistics` → `CoordinatorInitiateQueryStatistics`; scalapb `extends` option and proto comments | | Python runtime | `controller_interface` / `_controller_service_stub` / `controller_stub` → `coordinator_*`; generated betterproto bindings are gitignored and regenerate from the renamed proto (`bin/python-proto-gen.sh` verified) | | Actor identity | `CONTROLLER` ActorVirtualIdentity constant → `COORDINATOR` (Scala + Python) | | Location preference | `PreferController` → `PreferCoordinator` (workflow-core) | | Variables / comments | all camelCase & snake_case variants (`controllerConfig`, `controllerTimerService`, `controllerAddress`, …) and prose in scaladoc/docstrings | The rename is purely mechanical (case-preserving substring replacement, verified exhaustively): `grep -ri controller` over `amber/` and `common/` returns zero matches afterwards. Untouched on purpose: Angular's `HttpTestingController`, the Web `AbortController` in agent-service, and Envoy's `gatewayclass-controller` in the k8s templates. Follows #6123 (merged), which freed the Coordinator name by renaming the region-scheduling coordinators to managers. ### Any related issues, documentation, discussions? Closes #6122. ### How was this PR tested? Refactor with no behavior change — existing tests stay green with no assertion edits: ``` sbt "WorkflowCore/testOnly org.apache.texera.amber.core.workflow.WorkflowCoreTypesSpec org.apache.texera.amber.core.workflow.PhysicalOpSpec org.apache.texera.amber.util.VirtualIdentityUtilsSpec" sbt "WorkflowOperator/testOnly org.apache.texera.amber.operator.SpecialPhysicalOpFactorySpec" sbt "WorkflowExecutionService/testOnly org.apache.texera.amber.engine.architecture.coordinator.CoordinatorSpec org.apache.texera.amber.engine.architecture.coordinator.ClientEventSpec org.apache.texera.amber.engine.architecture.coordinator.WorkflowSchedulerSpec org.apache.texera.amber.engine.architecture.scheduling.WorkflowExecutionManagerSpec org.apache.texera.amber.engine.architecture.scheduling.RegionExecutionManagerSpec" cd amber && python -m pytest src/test/python/core/architecture/rpc/test_async_rpc_client.py src/test/python/core/util/test_virtual_identity.py src/test/python/core/runnables/test_console_message.py src/test/python/core/runnables/test_data_processor.py src/test/python/core/runnables/test_main_loop.py src/test/python/pytexera/storage/test_large_binary_manager.py ``` Full `Test/compile` of `WorkflowCore`, `WorkflowOperator`, and `WorkflowExecutionService` passes; Python bindings regenerated via `bin/python-proto-gen.sh` and 68 pytest cases pass. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Fable 5) Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

2026-06-17

feat(scheduling): reuse output storage across region re-executions (#5707) ### What changes were proposed in this PR? Adds an opt-in mechanism for an output port to **reuse** its storage when the owning operator's region re-executes, instead of recreating the document each time. Dormant and behavior-preserving — no operator sets the flag in this PR. - `OutputPort` gains a `reuseStorage: Boolean` proto field (alongside `blocking` / `mode`). It marks a port whose output accumulates across region re-executions — e.g. a Loop End port whose result builds up over the iterations of its own loop. - `DocumentFactory.createOrReuseDocument(uri, schema, reuseExisting, …)` is the create-or-reuse decision: when reuse is requested and a document already exists it opens and returns that one; otherwise it creates a fresh one. It always returns the document, so the call site does not branch. - `RegionExecutionCoordinator` reads each output port's `reuseStorage` flag while provisioning that port's result/state documents and routes through `createOrReuseDocument`. | port flag | region re-run behavior | |---|---| | `false` (every operator today) | recreate output/state documents — unchanged | | `true` (set by Loop End in the loop PR) | keep and reopen the existing documents | A runtime guard in `RegionExecutionCoordinator` asserts no port sets `reuseStorage` for now: the flag activates only with the loop operators, which are not yet on `main`. The guard keeps the dormant reuse path from being silently exercised before its consumer exists, and is removed when the loop operators land. ### Any related issues, documentation, discussions? Resolves #5709 (sub-issue of #4442 "Introduce for loop"). Split out of #5700 to keep that PR reviewable, per @Xiao-zhen-Liu's [review](https://github.com/apache/texera/pull/4206#pullrequestreview-4482667715). ### How was this PR tested? - `DocumentFactorySpec` — pins the create-or-reuse decision (the reuse × exists matrix plus the "no-reuse never probes existence" short-circuit) with injected document stubs, no iceberg backend. - `OutputPortReuseFlagSpec` — guards that no registered operator enables `reuseStorage` on any output port. - `WorkflowCore` / `WorkflowOperator` / `WorkflowExecutionService` compile; scalafmt + scalafix clean. ### Was this PR authored or co-authored using generative AI tooling? Co-authored with Claude Opus 4.8 in compliance with ASF.

2026-05-02

feat(engine): add jump-to-operator support (#4444) ### What changes were proposed in this PR? Add a generic controller-side primitive for jumping execution to the region containing a target operator (`JumpToOperatorRegion`). Design: - `Schedule` stays a minimal data class — a `Map[Int, Set[Region]]` keyed by contiguous `0..N-1` levels (enforced by a `require` invariant), an `initialLevelIndex` that seeds the iteration cursor on construction or `copy(...)`, and `Iterator[Set[Region]]` semantics so consumers walk it directly. No factory methods on `Schedule` produce other Schedules; `case class.copy(...)` is used at call sites instead. - `WorkflowExecutionCoordinator` exposes its current schedule as a public `var schedule` (no `getSchedule`/`replaceSchedule` wrappers). Normal execution mutates the cursor in place; a jump replaces the var with a new schedule via `copy(initialLevelIndex = targetLevel)`. - The jump computation lives in `JumpToOperatorRegionHandler`: it scans the coordinator's `schedule.levelSets` for the level containing the target operator and assigns a fresh schedule back to the coordinator. No precomputed operator→level map — each Schedule is single-use, so a map would cost the same as one linear scan. - `ControllerProcessor` constructs the coordinator with no schedule arg (the coordinator defaults to `Schedule(Map.empty)` until populated). `Controller.initState` assigns the real schedule into the coordinator immediately after `WorkflowScheduler.updateSchedule(physicalPlan)`. Out of scope (intentional): loop-operator logic, region-restart logic, state materialization. The primitive is generic so those features can be built on top. ### Any related issues, documentation, discussions? Closes #4443 Precursor test coverage for related modules: - #4562 — `ScheduleSpec` covering `Schedule` iterator semantics (open) - #4564 — `WorkflowSchedulerSpec` covering `WorkflowScheduler` contract (merged) ### How was this PR tested? - `sbt "WorkflowExecutionService/compile"` — passes. - `sbt "WorkflowExecutionService/testOnly org.apache.texera.amber.engine.architecture.scheduling.WorkflowExecutionCoordinatorSpec"` — 8/8 tests pass (1 pre-existing kill-sync test plus 7 added by this PR exercising jump behavior). ### Was this PR authored or co-authored using generative AI tooling? Generated-by: ChatGPT (Codex), Claude Code (Claude Opus 4.7) --------- Signed-off-by: Xinyuan Lin <xinyual3@uci.edu> Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

2026-04-25

feat(amber): Re-enable operator reconfiguration in Amber (#4220) ### What changes were proposed in this PR? Per the discussion in https://github.com/apache/texera/discussions/4016, we decided to bring the operator reconfiguration feature back to the Amber engine. This PR includes only the backend changes for this feature, but it is enabled on both the Java and Python sides. Since the code for the Fries Algorithm is still in the codebase, this feature is relatively straightforward to implement and maintain going forward. This PR allows source operators to be included in the reconfiguration scope (MCS), but it does not allow source operators themselves to be modified. First, under the current iterator-based interface, the state of a source operator is fully encapsulated within its iterator. Reading or manipulating the iterator state is already very difficult in both Scala and Python. Second, even if we could access the state, it would still be hard for users to clearly define the expected state transition semantics—e.g., whether to preserve the old state, reset it, or partially transfer it to the new operator. Due to the reasons above, we disable reconfiguration of source operators for now. If clear use cases emerge in the future, we can revisit this design decision. ### Any related issues, documentation, discussions? See https://github.com/apache/texera/discussions/4016. ### How was this PR tested? Introduced unit tests for this feature. This PR also updates scala CI to install python dependencies as we are using Python UDFs in our e2e tests. ### Was this PR authored or co-authored using generative AI tooling? No --------- Signed-off-by: Yicong Huang <17627829+Yicong-Huang@users.noreply.github.com> Signed-off-by: Shengquan Ni <13672781+shengquan-ni@users.noreply.github.com>

2026-04-11

feat: allow Multi-Link on Input Ports (#4342) ### What changes were proposed in this PR? Enabling multi-link support for input ports. | Before the change | After the change | | ------------- | ------------- | | <img width="651" height="491" alt="image" src="https://github.com/user-attachments/assets/6ebee08c-89ab-4731-bc4a-cfcd95ea8203" /> | <img width="661" height="498" alt="image" src="https://github.com/user-attachments/assets/bab2ca5c-be72-4643-8d21-5c56156e61a1" /> | ### Any related issues, documentation, discussions? Closes #4329 ### How was this PR tested? Tested manually. ### Was this PR authored or co-authored using generative AI tooling? No. --------- Signed-off-by: Xinyuan Lin <xinyual3@uci.edu> Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> Co-authored-by: Chen Li <chenli@gmail.com>

2026-02-24

feat: Separate Runtime Statistics Collection from UI Updates (#4205) <!-- Thanks for sending a pull request (PR)! Here are some tips for you: 1. If this is your first time, please read our contributor guidelines: [Contributing to Texera](https://github.com/apache/texera/blob/main/CONTRIBUTING.md) 2. Ensure you have added or run the appropriate tests for your PR 3. If the PR is work in progress, mark it a draft on GitHub. 4. Please write your PR title to summarize what this PR proposes, we are following Conventional Commits style for PR titles as well. 5. Be sure to keep the PR description updated to reflect all changes. --> ### What changes were proposed in this PR? <!-- Please clarify what changes you are proposing. The purpose of this section is to outline the changes. Here are some tips for you: 1. If you propose a new API, clarify the use case for a new API. 2. If you fix a bug, you can clarify why it is a bug. 3. If it is a refactoring, clarify what has been changed. 3. It would be helpful to include a before-and-after comparison using screenshots or GIFs. 4. Please consider writing useful notes for better and faster reviews. --> This PR introduces a new configuration parameter `runtime-statistics-persistence-interval` to independently control the frequency of runtime statistics persistence, separate from the UI update frequency (`status-update-interval`). Previously, both UI updates and runtime statistics persistence were controlled by a single parameter `status-update-interval`. This means frequent UI updates (e.g., 500ms) caused excessive statistics writes to storage. This change allows independent control: - `status-update-interval`: Controls how often the frontend UI refreshes (default: 500ms) - `runtime-statistics-persistence-interval`: Controls how often statistics are persisted to storage (default: 2000ms) #### Do two timers mean more frequent worker queries? No. The controller tracks the timestamp of the last completed full-graph worker query and uses `min(status-update-interval, runtime-statistics-persistence-interval)` as a freshness threshold. When a timer fires, if the elapsed time since the last query is within this threshold, the controller forwards stats from cache without querying workers — so the faster timer drives all real worker queries and the slower timer always reuses the result. If a query is already in-flight when the second timer fires, the controller serves stats from the previous completed query's cache. Cache reuse applies to timer-triggered queries only; event-triggered queries (e.g., from worker completion events) always proceed to real worker RPCs. #### Changes - Added `runtime-statistics-persistence-interval` parameter (default: 2000ms) in `application.conf` - Protobuf: Added `StatisticsUpdateTarget` enum (`UI_ONLY`, `PERSISTENCE_ONLY`, `BOTH_UI_AND_PERSISTENCE`) to `QueryStatisticsRequest` - Added `RuntimeStatisticsPersist` event for statistics-only updates; `ExecutionStatsUpdate` now handles UI-only updates - Added separate timer for runtime statistics persistence that runs independently from the UI update timer - Query Handling - Timer-triggered queries specify target: UI-only or persistence-only - Event-triggered queries (port/worker completion, pause, resume) send both UI and persistence updates to preserve original behavior - `QueryWorkerStatisticsHandler` routes to the appropriate event based on `StatisticsUpdateTarget` - Worker query deduplication in `QueryWorkerStatisticsHandler`: when the second timer fires, the controller checks whether worker stats were already fetched recently (within `min(status-update-interval, runtime-statistics-persistence-interval)`). If so, it forwards the cached stats to the appropriate sink (UI or persistence) without issuing any worker RPCs. If a query is already in-flight, cached stats from the previous completed query are forwarded. ### Any related issues, documentation, discussions? <!-- Please use this section to link other resources if not mentioned already. 1. If this PR fixes an issue, please include `Fixes #1234`, `Resolves #1234` or `Closes #1234`. If it is only related, simply mention the issue number. 2. If there is design documentation, please add the link. 3. If there is a discussion in the mailing list, please add the link. --> Closes #4204 ### How was this PR tested? <!-- If tests were added, say they were added here. Or simply mention that if the PR is tested with existing test cases. Make sure to include/update test cases that check the changes thoroughly including negative and positive cases if possible. If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future. If tests were not added, please describe why they were not added and/or why it was difficult to add. --> Tested with the following workflow and dataset, change the `runtime-statistics-persistence-interval` parameter to see if the runtime stats size reduces if we increase the parameter value. [Iris Dataset Analysis.json](https://github.com/user-attachments/files/25220000/Iris.Dataset.Analysis.json) [Iris.csv](https://github.com/user-attachments/files/25220003/Iris.csv) ### Was this PR authored or co-authored using generative AI tooling? <!-- If generative AI tooling has been used in the process of authoring this PR, please include the phrase: 'Generated-by: ' followed by the name of the tool and its version. If no, write 'No'. Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details. --> Generated-by: Claude-4.6 --------- Co-authored-by: Xiaozhen Liu <xiaozl3@uci.edu> Co-authored-by: Chen Li <chenli@gmail.com>

2025-12-01

chore: Move `apache.amber` to `apache.texera.amber` (#4094) ### What changes were proposed in this PR? This PR refactors the package structure by moving all Amber engine code from `org.apache.amber` to `org.apache.texera.amber`. This aligns the package naming with the Texera project organization and ensures all components are properly namespaced under the Apache Texera organization. **Key Changes:** 1. **Directory Structure Migration** - Moved all source directories: - Scala/Java sources: 8 modules moved - Protobuf definitions: 14 files moved - Python proto generated code: moved under new namespace - Frontend TypeScript proto: moved under new namespace 2. **Code Updates** - Updated across 707 files: - Package declarations in 576 Scala/Java files - Import statements across all Scala/Java files - 57 Python files updated for new proto imports - 14 Protobuf files updated with new Java package - 2 TypeScript files updated with new import paths - Configuration files (cluster.conf) - String literals containing class names for reflection/dynamic loading 3. **Package Namespace Changes:** ```diff - org.apache.amber.engine.common - org.apache.amber.operator.* - org.apache.amber.core.* - org.apache.amber.compiler.* + org.apache.texera.amber.engine.common + org.apache.texera.amber.operator.* + org.apache.texera.amber.core.* + org.apache.texera.amber.compiler.* ``` ### Any related issues, documentation, discussions? Closes #4003 ### How was this PR tested? CI ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Sonnet 4.5 (Cursor IDE)

2025-10-27

Add websocket event BigObjectUpdateEvent

2025-10-14

Big Object for Java Native Operators

2025-10-13

chore: rename `core` to `common` (#3882) According to the reorganization plan #3846. This PR renames `core` out to `common`. The folder will contain shared dependencies of services. resolves #3861 --------- Co-authored-by: Xinyuan Lin <xinyual3@uci.edu>

2025-10-12

chore: relocate amber (#3878) According to the reorganization plan #3846. This PR moves `core/amber` out to root. resolves #3865 --------- Signed-off-by: Yicong Huang <17627829+Yicong-Huang@users.noreply.github.com>

chore: relocate amber

2025-10-11

chore: rename packages to org.apache (#3848) Update package names in Java and Scala code to reflect the migration of the project to the Apache Software Foundation. Changes: 1. Replaced package references from `edu.uci.ics` to `org.apache`. 2. Moved source files from `./edu/uci/ics/` to `./org/apache/` fix #3792

2025-07-08

Merge branch 'master' into xinyuan-loop-mvp Signed-off-by: Xinyuan Lin <xinyual3@uci.edu>

2025-07-07

feat(amber): enable terminating workers in a region upon completion of region execution (#3468) This PR adds the ability for `RegionExecutionCoordinator`s to terminate its workers when it detects all the workers are completed. ## Design - A new control message `EndWorker` is added. - When a `RegionExecutionCoordinator` syncs its completion status with `RegionExecution` and detects it is completed, it sends an `EndWorker` control message to all the workers of this region and waits for their confirmations. - After receiving this final response from all the workers, `RegionExecutionCoordinator ` sends `gracefulStop` to all its workers, which terminates all the worker actors (PVMs are also killed automatically). ## Testing Use the same workflow and file in #3425 to test this PR. Expectations: - The workflow should execute successfully without any issues. - If you set a breakpoint in `WorkflowExecutionCoordinator`, line 86, and run the workflow, it will hit the break point each time a new region is launched. Then you can verify for each region whether only the PVMs for this region are running. - Check the logs of `ComputingUnitMaster` and see if `Region XX successfully terminated.` is logged for each region.

2025-07-02

Merge branch 'master' into xiaozhen-kill-regions # Conflicts: # core/gui/src/app/workspace/service/execute-workflow/execute-workflow.service.ts

2025-06-25

fix the control return

add table profile to the websocket event

merge the controller message with table profiler query

update the table profile definition

modify the profiler definition

add table profile related proto

Merge branch 'master' into xinyuan-loop-mvp

Rename ControlPayload Class to Direct Control Message (DCM) (#3492) In this PR, we will rename ControlPayload to Direct Control Message (DCM) to align with the terminology used in our discussions.

2025-06-24

fix

fix

2025-06-22

done!

2025-06-20

update

2025-06-18

address more comments.

Merge branch 'master' into xiaozhen-kill-regions # Conflicts: # core/amber/src/main/python/proto/edu/uci/ics/amber/engine/architecture/rpc/__init__.py

2025-06-17

Rename ChannelMarker Class to Embedded Control Message (ECM) (#3472)

update

reverse "TERMINATING"

minor refactoring.

fix fmt

Merge branch 'master' into xiaozhen-kill-regions # Conflicts: # core/amber/src/main/protobuf/edu/uci/ics/amber/engine/architecture/rpc/workerservice.proto # core/amber/src/main/python/core/architecture/rpc/async_rpc_handler_initializer.py # core/amber/src/main/python/proto/edu/uci/ics/amber/engine/architecture/rpc/__init__.py # core/amber/src/main/scala/edu/uci/ics/amber/engine/architecture/worker/DataProcessorRPCHandlerInitializer.scala

2025-06-14

update

init

Migrate existing markers into Control Message (#3432) Migrate existing markers — `StartChannel` and `EndChannel` — into `ControlMessage`. The `EndChannel` marker requires the following input alignment logic: Alignment of multiple `EndChannel` markers from different workers of the same port on the same operator. The alignment mechanism was originally implemented in the DP module. However, since `ChannelMarker` now handles alignment across multiple workers, the alignment logic (multi-worker alignment on the same port) has been moved to `ChannelMarker`. It is now called `PORT_ALIGNMENT`, as it is a general-purpose alignment strategy applicable to other types of `ChannelMarker`.

2025-06-13

Remove EndWorkerRequest.

Remove `EndWorkerResponse`

Change `Terminated` to be `Terminating`

2025-06-12

Add termination logic for Python worker.

2025-06-11

WIP for kill with control messages.

2025-06-06

Merge branch 'master' into xinyuan-basic-for-loop

2025-05-31

Rename to ALL_ALIGNMENT

2025-05-29

add PORT_ALIGNMENT

2025-05-28

fix the control return

add table profile to the websocket event

merge the controller message with table profiler query

update the table profile definition

modify the profiler definition

add table profile related proto

2025-05-27

Merge branch 'master' into xinyuan-migrate-marker2

Use Input Port Materialization Reader Threads to Replace Cache Source Operators (#3425) ## Motivation & Benefits In #3312 we removed sink operators. A natural symmetric next step is to remove cache source operators. Similar to #3312 and #3295, the main idea is to use input ports to read materialized storage. Beyond symmetry, doing has the following benefits: - By not using operators to implement materialized links, the scheduler will no longer modify a physical plan during schedule generation. This makes the design much simpler as the physical plan can be immutable, and truly separated from the scheduling. - Using input ports / output ports to indicate and implement materialization read/writes simplifies the future steps like clean separation of region resources, enabling caching, etc. ## Core Changes ### Scheduling #### Resource Config (`PortConfig`) `PortConfig` is updated with the two changes: - `PortConfig` now becomes the base class for two subclasses: `InputPortConfig` and `OutputPortConfig`. The two subclasses are the interfaces between the Scheduler and the Executor. - Each `InputPortConfig` contains a list of pairs of `URI` and `Partitioning`. - Each `OutputPortConfig` contains a single `URI`. - Another subclass `IntermediateInputPortConfig` is introduced, which is internal to the Scheduler (see docs for details) #### Scheduler Generators Both SchedulerGenerator Implementations (Cost-based & Greedy) have the following updates: - Materialized link implementation is updated: - Previously, a materialized link is implemented as an upstream output port with a URI + a cache source operator connecting to the downstream input port. - Now, it is simply an upstream output port with a URI + a downstream input port with this URI - Before resource allocation on a RegionDAG, the scheduler generates a RegionDAG with each region populated with a `ResourceConfig` that contains the appropriate URI information for each output port and **each input port** ##### For `ExpansionGreedyScheduleGenerator`, as it previously relies on links connected to each input port to enforce input port dependencies, and removing cache read operators would enable some input ports to have no input links at all, I updated the logic of `handleDependentLinks ` to use a combination of input ports and links instead. The assumption is, if an input port is not connected to any links (because it is the result of a materialization), then it will not cause any further cycles in the region graph. See the documentation about this method for details. ##### ScheduleGenerator base class Some shared methods are only used by ExpansionGreedyScheduleGenerator after this PR, so I moved them to `ExpansionGreedyScheduleGenerator` #### Resource Allocator - `DefaultResourceAllocator` is updated with the logic to assign `partitionings` to all the input ports of a region. - Each URI of each input port is treated like a worker with a `VirtualThreadActorId`. - Then the same logic we use to assign partitionings on each link is applied (`toPartitioning()`) ### Region Execution #### Region Added a `getStarterOperators ` method for a region, which is used by `StartWorker`. The workers of these operators will be started first in a region: - Source operators - Operators that have input port(s) that need to read from materialization. #### RegionExecutionCoordinator - The coordinator is updated based on the changes in `PortConfig` - PortConfigs of input ports are now used to send `AssignPort` requests. - `StartWorker` is sent to operators using `getStarterOperators `. - Duplicate URI check when assigning storage objects is no longer needed as an output port will never belong to two regions now (we fixed that in #3312). ### Worker Execution #### `InputPortMaterializationReaderThread` - This new thread class is added for both JVM workers and Python workers. - It sits inside InputManager. - It is assigned by `AssignPort` request, and started by `StartWorker` request. - It exits when it finishes reading all the inputs. - Its behavior mimics an upstream cacheSource operator, reusing the same logics as those in OutputManager. - It uses the same partitioning logic as in OutputManager, except it only sends to the worker it belongs to. - It puts the packaged messages to the DP internal queue, as if an upstream operator sent these messages. #### `OutputManager` As an operator with input port dependencies will belong to two regions, and is started in the dependee region first, to avoid the operator's workers being terminated before the depender region starts, I added an additional logic in `finalizeOutput` to avoid finalizing it when a worker does not have an output port. This is a side-product of this PR as it fixes an issue where the CartesianProduct does not produce results (because its worker is terminated too early) ## Limitations The main limitation of this PR is shuffling efficiency. See #3392 for details. ## Remaining TODO(s) This PR only separates input and output port configs on the scheduling level. Each worker still receive a unified `AssignPortRequest` for both input and output ports. We may need another PR to also separate the control commands for input and output ports. I did not include that as part of this PR as including it may be too complicated for a single PR. ## Additional Bug Fixes related to Python Partitioners Currently there is an issue with Python partitioner implementations. - `Hash`, `RoundRobin`, and `Range` partitioners currently use the index of a list of downstream worker ids to select downstream worker to send a tuple to. This list is `self.receivers`. - This **ordered** list `Partitioning` is created by the `ResourceAllocator` and sent to a Python worker by the controller. - In the current Python implementations of these partitioners, when receiving `Partitioning`, we are creating a intermediate `set` during the conversion of this `Partitioning` object into `self.receivers`. - Using this intermediate `set` structure WILL break the order of the list as using `{}` to create a `set` from an ordered list does not preserve the order of that list (because sets are unordered.) This causes `self.receivers` to be in a random order. - The fix is to use `dict.fromkeys(iterable)` to replace `{}` as "the insertion-order preservation nature of dict objects has been declared to be an official part of the Python language spec" since [Python 3.7](https://docs.python.org/3/whatsnew/3.7.html). I tested and can confirm this will fix the issues. - Note: For the sake of cleanness, I will create a separate PR to include these fixes. In this PR I am only including the changes to RoundRobin partitioner as it is required for replacing cache read operators. ## Testing The expected behavior of merging this PR is it should not have any change to the user experience or the behavior of workflow execution. Use this giant workflow to test it: <img width="2560" alt="image" src="https://github.com/user-attachments/assets/f7fc364f-1a91-4e5f-8a59-ca5a15567363" /> Workflow: [input-port-reader-test.json](https://github.com/user-attachments/files/20360881/input-port-reader-test.json) Dataset: [onehot-coding-for-multivalue-variable-sample-v1.zip](https://github.com/user-attachments/files/20067133/onehot-coding-for-multivalue-variable-sample-v1.zip) #### Expected outcome This workflow should execute "more correctly" as it does in the current master branch: - In the master branch there is a bug where the CartesianProduct will not display results. After this PR the CartesianProduct operator will behave correctly. - The Python HashPartitioner bugs will cause most of the HashJoin operators to have less output data than they should when the HashJoin's build input is connected to an upstream PythonUDF because the UDF is not doing partitioning correctly. After the next PR to fix those issues, all the HashJoin operators will behave correctly and **the last operator should have the same number of tuples as the input of the first operator**. Unfortunately they will not behave correctly in this PR (same as in the master branch). - Try different default number of workers and different UDF worker numbers to test if multi-worker input port reader is correct. - Try both CostBasedScheduleGenerator and ExpansionGreedyScheduleGenerator, and both should be able to execute this workflow correctly. - Try different number of input tuples by adjusting the input limit of the first operator.

2025-05-23

update

update

2025-05-19

fix the control return

add table profile to the websocket event

merge the controller message with table profiler query

2025-05-18

update the table profile definition

modify the profiler definition

2025-05-17

add table profile related proto

2025-05-15

update

2025-05-14

update

2025-05-05

Merge branch 'master' into xinyuan-migrate-marker1

Merge branch 'refs/heads/master' into xiaozhen-input-port-storage # Conflicts: # core/amber/src/main/python/core/architecture/packaging/input_manager.py # core/amber/src/test/scala/edu/uci/ics/amber/engine/architecture/worker/DPThreadSpec.scala # core/amber/src/test/scala/edu/uci/ics/amber/engine/architecture/worker/DataProcessorSpec.scala

2025-05-04

Add ASF header and RAT CI (#3415) Add the ASF header to all source code in the repo and introduce RAT CI. [Apache Release Audit Tool (RAT)](https://creadur.apache.org/rat) is a release audit tool focused on licenses. This PR adds the release audit tool to the GitHub Action workflow (CI). This would allow developers to quickly detect if new files during a PR submission or commit push were missing the license header. Add files to the optional .ratignore file if you want to exclude certain files and folders from being tested.

2025-05-02

init

2025-04-30

Add partitionings for port mat readers in inputManger of java side.

2025-04-28

init

2025-04-23

Xinyuan migrate marker (#3390)

2025-04-03

replace storageURI with storageURIs

2025-03-26

update

2025-03-19

update

2025-03-17

Merge branch 'master' into xinyuan-basic-for-loop

2025-03-15

Merge branch 'master' into xinyuan-migrate-marker

2025-03-12

Remove Sink Operator in the Backend (#3312) ### Contents of this PR 1. Completely removes sink operators in the backend, including physical sink operator, sink operator executor and related proto definition, and places that refer to the sink operator. 2. Replaces the logic of `getPorts` of a region by saving ports separately (previously we are using links to get the ports). 3. Fixes the timing of flushing storage buffer for port storage writer threads. 4. Refactors Python worker to send `PortCompleted` even if no links are connected to an output port. Relevant test cases are also modified to remove sink operators from the test cases. ### Regarding region completion and `getPorts`: Currently our implementation of region completion logic is not correct as it uses links of a region to get ports of the region and use the completion of these ports to indicate completion of the region. This hacky implementation worked in the past when we created sink operators explicitly but will not work in this PR. Ideally: - A region should be defined using only the operators and links in the region. - The completion of a region should be based on the completion of its operators, and the completion of an operator is defined by the completion of all of its ports. However, we currently have a hacky implementation of “hash-join”-like operators that have dependee input ports, and we include operator with a dependee input port in two regions, which prevents us from relying on operators to define the completion of a region. Our ultimate goal is to have a clean separation of regions, but that requires removing cache read operator first. To proceed with removing sink operator, in this PR, we implement a workaround in the region completion logic and do not rely on operators of a region to get the ports. After clean separation of regions is done, we can implement the clean definition of region completion logic and remove redundant port information in regions. ### Progressive Computation **Note: This PR removed ProgressiveSinkOp. We will remove ProgressiveUtils soon due to the high cost of implementing the retraction operation. Since we now use Iceberg for storage with an append-only default, removing a record requires a full scan, making retraction impractical. We will revisit it if there is more use cases requiring progressive computation.** ### TODOS that are not part of this PR - Rename "sinkStorageTTLInSecs" and "sinkStorageCleanUpCheckIntervalInSecs" in the AmberConfig. - Remove/rename mentions of sink in the frontend.

2025-03-10

Merge branch 'master' into xinyuan-basic-for-loop

2025-03-09

Use Output Ports of an Operator to Write Storage (#3295) Currently the Amber engine creates and uses sink operators to write the results of output ports of an operator. This design causes many problems, mainly because it alters the physical plan. Ideally a physical plan should not be changed once it is compiled. This PR updates the design to use output ports of an operator instead of sink operators to write port results to storage. #### Core Design Changes The changes implemented in this PR include: - The logic to add sink operators during compilation and scheduling are not removed in this PR. However, all the execution logic in the sink operator are removed. A sink operator will not do anything during execution. We will remove sink operators in the next PR. - `GlobalPortIdentity` is moved from `Region` to `workflow-core` so that it is accessible by all the modules. - The compiler does not create storage objects anymore. Instead, it produces a set of `GlobalPortIdentity` that need view results. This set is passed along to the scheduler as part of `WorkflowContext.workflowSettings`. In the future, this information will be directly produced by the frontend instead of by the compiler. - The scheduler combines the ports that need view result and materialized ports needed by the scheduler as part of a region. Ideally this information should be fixed once a region is created by the SchedulerGenerator. However, as the physical plan still needs to be changed currently (because of additional cache read operators), additional logic is implemented in the ScheduleGenerator to make sure this information is correct for all the regions. - The scheduler and resource allocator do not create storage objects directly and only assign storage URIs for each region as part of `resourceConfig`. - When a region is scheduled to execute, during the initialization of a region, these URIs are used to create storage objects. - `AssignPortRequest` is used to indicate whether an output port of a worker needs storage and to pass the storage URI information to a worker. Note this request is used for both input ports and output ports, and this PR only updates output ports. As a result, for input ports, empty storage URIs will be provided in `AssignPortRequest`. In the future, after we also use input ports to read storage, we will also update and use these storage URIs. - Note that since currently operators with dependee inputs belong to multiple regions, and AssignPortRequest is only used once for each worker, I had to implement additional logic in the to make sure all the regions that such an operator belongs to have the proper storage information (specifically, the output port connecting a dependee link belongs to two regions, and both regions need to have storageURI for this port) - Inside a worker (both Java and Python), the OutputManager is used to create writer threads for each output port that needs storage. The writing does not block the data processor, but will block the completion status of the operator/port. #### Relevant fixes This PR also contains a fix introduced by #3304, where the status of a workflow is not updated after running a workflow for the 2nd time. #### TODOs: - Completely remove sink op - Use `GlobalPortIdentity` for storage URIs - Remove the mention of "result" in the storage layer - Let the frontend specify view results on the port level - Remove cache read op and use input port for reading storage ### Important: Postgres Catalog Is Required After this PR This PR requires Postgres catalog to be set up because the Python Iceberg storage layer will be used (previously it was added in the codebase but not used), and Python Iceberg only works with Postgres catalog. If you have not switched to Postgres catalog, please refer to #3243 to set it up. We will soon make postgres catalog the default and possibly remove hadoop catalog. --------- Co-authored-by: Shengquan Ni <13672781+shengquan-ni@users.noreply.github.com>

2025-03-02

Merge branch 'master' into xinyuan-basic-for-loop

2025-02-28

init

2025-02-25

Revert "Revert "Enhance Runtime Statistics with Cumulative Metrics and Tuple Size Measurement (#3262)"" This reverts commit e11f1b5e20d76b30521eb452d93495682b9e6bd5.

Revert "Enhance Runtime Statistics with Cumulative Metrics and Tuple Size Measurement (#3262)" This reverts commit d3861cccdbcbe23499a8763672c10a5ee7c83635.

2025-02-19

Add ChannelMarker support on pyAmber (#3245) This PR: 1. Added ChannelMarkerManager to track and align ChannelMarkers on the Python side. 2. Now messages to Python also attach ChannelIdentity to distinguish different channels, previously we used ActorVirtualIdentity. 3. Due to the ChannelIdentity change, the generated proto message on Python side has to assign the `is_control` field explicitly, otherwise its hash is not consistent. To give an example: ``` channel_id = ChannelIdentity(from_worker_id = ..., to_worker_id = ...) new_channel_id = bytes(channel_id) assert hash(channel_id) == hash(new_channel_id) // this will fail. ``` 4. The InputQueue now also uses channel ID as a key, instead of using 2 fixed keys: CONTROL, DATA. --------- Co-authored-by: Xinyuan Lin <xinyual3@uci.edu>

Enhance Runtime Statistics with Cumulative Metrics and Tuple Size Measurement (#3262) This PR updates the runtime statistics by refining how metrics are stored and introducing new tuple size measurements. ### Changes 1. Cumulative Statistics: Instead of storing differential (diff) values for tuple counts and processing times, we now maintain cumulative values. This change simplifies performance tracking and analysis over time. 2. Tuple Size Metrics: Added metrics for both input and output tuple sizes. These sizes are computed using deep size calculations. The deep size calculation traverses the object graph starting from the target object, including all fields, array elements, and referenced objects. - Scala: Uses the `deepSizeOf` method. - Python: Uses the `pympler.asizeof` method. ### Migration To enable the new tuple size measurement in Python, please run: `pip install -r core/amber/requirements.txt` https://github.com/user-attachments/assets/1fc194bd-95d3-4946-b47a-e6e22b7b28a6

2025-02-13

Refactor codes

tuple size

2025-02-05

Merge branch 'shengquan-add-channel-marker-mgr-py' into iced-tea-vldb-submission

2025-02-04

WIP

2025-01-24

update

2025-01-23

Merge remote-tracking branch 'icedtea/master' into iced-tea-vldb-submission

2025-01-13

init