Proto commits in Texera/texera

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

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

The documentation is generated from this commit.

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

2024-12-31

Make OpExecInitInfo serializable (#3183) Previously, creating a physical operator during compilation also required the creation of its corresponding executor instances. To delay this process so that executor instances were created within workers, we used a lambda function (in `OpExecInitInfo`). However, the lambda approach had a critical limitation: it was not serializable and it is language dependent. This PR addresses this issue by replacing the lambda functions in `OpExecInitInfo` with fully serializable Protobuf entities. The serialized information now ensures compatibility with distributed environments and is language-independent. Two primary types of `OpExecInitInfo` are introduced: 1. **`OpExecWithClassName`**: - **Fields**: `className: String`, `descString: String`. - **Behavior**: The language compiler dynamically loads the class specified by `className` and uses `descString` as its initialization argument. 2. **`OpExecWithCode`**: - **Fields**: `code: String`, `language: String`. - **Behavior**: The language compiler compiles the provided `code` based on the specified `language`. The arguments are already pre-populated into the code string. ### Special Cases The `ProgressiveSink` and `CacheSource` executors are treated as special cases. These executors require additional unique information (e.g., `storageKey`, `workflowIdentity`, `outputMode`) to initialize their executor instances. While this PR preserves the handling of these special cases, these executors will eventually be refactored or removed as part of the plan to move storage management to the port layer.

Remove duplicated scalapb definition (#3182) The scalapb proto definition both present in workflow-core and amber. This PR removes the second copy.

2024-12-30

Move protobuf definitions under core package (#3181) This PR moves all definitions of protobuf under workflow-core to be within the core package name.

2024-12-19

Move output mode on port (#3169) This PR unifies the design of the output mode, and make it a property on an output port. Previously we had two modes: 1. SET_SNAPSHOT (return a snapshot of a table) 2. SET_DELTA (return a delta of tuples) And different chart types (e.g., HTML, bar chart, line chart). However, we are only using HTML type after switching to pyplot. Additionally, the output mode and chart type are associated with a logical operator, and passed along to the downstream sink operator, this does not support multiple output ports operators. ### New design 1. Move OutputMode onto an output port's property. 2. Unify to three modes: a. SET_SNAPSHOT (return a snapshot of a table) b. SET_DELTA (return a delta of tuples) c. SINGLE_SNAPSHOT (only used for visualizations to return a html) The SINGLE_SNAPSHOT is needed now as we need a way to differenciate a HTML output vs a normal data table output. This is due to the storage with mongo is limited by 16 mb, and HTMLs are usually larger than 16 mb. After we remove this limitation on the storage, we will remove the SINGLE_SNAPSHOT and fall back to SET_SNAPSHOT.

2024-12-04

Update amber to depend on sub projects (#3111) This PR removes two copies of the code for workflow-core and workflow-operator and makes amber directly depend on those projects. Important points: 1. downgraded `snakeyaml` in `workflow-core` from 2.0 to 1.30 due to a conflict in `amber` project. 2. override all operator definitions in `workflow-compiling-service` with the latest changes in `amber`. Then deleted all operator definitions in `amber`. 3. we still have 2 implementations in `workflow-compiling-service` and `amber` for compilation. We should soon remove the implementation in `amber` Co-authored-by: Jiadong Bai <bobbaicloudwithpants@gmail.com> Co-authored-by: Yicong Huang <17627829+Yicong-Huang@users.noreply.github.com>

Flatten micro-service folder (#3098) This PR flattens the micro-services folder into the core folder. Our next step will be to let the amber project depend on the micro-services so we can get rid of the two-copy issue ASAP. --------- Co-authored-by: Jiadong Bai <43344272+bobbai00@users.noreply.github.com>