-
Notifications
You must be signed in to change notification settings - Fork 185
Design and merge plan: operator output port result cache (MVP) #5880
Design and merge plan: operator output port result cache (MVP)
In current Texera main, the engine runs a workflow from the start every time, even when the user changed only one operator near the end. This proposal adds a result cache so that, on a re-run, an output port whose upstream computation logic is unchanged reads its saved result instead of recomputing it. The code is written and working on a prototype branch. This post describes the design and the plan to bring it into main as small PRs, so anyone can raise concerns before the PRs go up.
Matching results across executions
Each output port has a cache key built from its upstream operators, their parameters, their output schemas, and the wiring between them. Two ports with the same cache key produce the same result (output port equivalence). When a run saves a port's result, we record (workflow, port, cache key) -> result location. On a later run, a port whose cache key has a recorded result is a matched port, and that result can be reused. Any edit upstream of a port changes its cache key, so its old result is no longer matched.
Scope (MVP)
In scope: reuse the saved result at every matched port (full reuse), match by cache key, invalidate entries that no longer match after an edit, and read, write, and clear the cache from the UI.
Out of scope (future work, not in these PRs): choosing per port whether reuse is cheaper than recompute (cost-based reuse planning), and removing results under storage limits (eviction). The merged code always reuses a matched port's result.
How it fits the current system
Current main (figure below): the Workflow Compiler builds a physical plan, CostBasedScheduleGenerator builds a schedule of regions, and the executor runs the regions on workers that read and write tables in storage.
With the cache MVP (figure below), the engine includes additional modules:
- Cache service: at submission, find the matched ports for this workflow (cache-key lookup); during execution, record cache metadata.
- Skeleton generator: remove the operators and edges whose results are reused, leaving the run-skeleton, the part that still needs to run.
- Scheduler (
CostBasedScheduleGenerator): schedule the run-skeleton as it does today. The removed part becomes regions that are skipped, and operators that read from it use the saved result locations.
The executor saves results to the cached-result storage as ports finish.
op-port-cache-related-diagrams-0 execution components (cache) drawio
Nothing changes when there are no matched ports
On the first run of a workflow, or any run right after an upstream edit, there are no matched ports: the run-skeleton is the whole plan and the schedule is the same as today. The cache changes behavior only when a matched port exists, so the code can land inactive and turn on once results are saved.
Merge plan: five PRs
In dependency order; each has its own issue:
- Storage foundation (Add operator output port cache storage (table and cache key) #5882 ): the cache table, the code that reads and writes it, and the cache-key computation.
- Cache state and statistics (Carry operator cache-reuse status to the frontend and handle cached-region statistics #5883 ): a "completed from cache" operator state and the matching statistics handling.
- Scheduler (Add cache-reuse planning to the scheduler (skeleton generation and schedule assembly) #5884 ): the reuse planner (full reuse), skeleton generator, and the change that schedules the run-skeleton and combines it with the skipped regions.
- Turn the feature on (Wire cache lookup, result saving, REST endpoints, and cleanup into execution #5885 ): the submission-time cache-matcher lookup, saving results as ports finish, the cache endpoints, and cleanup on deletion.
- Frontend (Add cache panel and canvas display to the workflow editor #5886 ): the cache panel and the canvas display.
PRs 1 and 2 are independent. PR 3 needs 1 and 2; PR 4 needs 3; PR 5 needs 4.
All reactions
Replies: 4 comments 1 reply
@Xiao-zhen-Liu Thanks for the great summary. Please also describe our plan to manage the lifecycle of the cached results.
All reactions
@Xiao-zhen-Liu Please chime in.
All reactions
@chenlica, here is the plan for managing the lifecycle of cached results.
Lifecycle of a cached result
Creation. An entry is written when a port's result is materialized during a run, as part of saving that result. It records the cache key of the port's upstream computation, the result's storage location, and the source execution.
Reuse / invalidation. On a later run, a port whose upstream computation is unchanged matches by cache key and reads the stored result instead of recomputing; entries persist across runs. A changed upstream (parameter, wiring, schema, operator version) yields a new cache key and a new entry; old entries are never overwritten, they just stop being matched.
Where results live. Materialized results are stored in the shared result storage (Iceberg via the REST catalog plus object store), addressed by workflow/execution/operator/port. This storage is process-wide and independent of any computing unit, so cached results survive a computing unit being destroyed. (This supersedes the earlier assumption that results are tied to the computing unit; after the storage-from-compute separation, they are not.)
Removal. Cleanup is driven by the workflow lifecycle, not the computing unit. The direction is to reuse WorkflowLifecycleManager's idle cleanup, which already clears a workflow's resources once it has had no connected users for a timeout, and extend it to also drop the workflow's cache entries and their stored results. Two points are still open for discussion:
- the retention window for cached results (TBD, configurable): the current 30s timeout (
web-server.workflow-state-cleanup-in-seconds) is for a live session's transient results and is too short for a cache whose value is reuse across sessions; - whether cache cleanup should be tied to that idle timeout at all (with a much longer window), or decoupled from the session so entries persist until the workflow is deleted or invalidated.
Either way, cache entries are also removed when the workflow is deleted (the table's foreign key to workflow cascades).
Out of scope (MVP). No size-bounded or cost-based eviction; that is the cost-model work and is intentionally not in this merge.
All reactions
@Xiao-zhen-Liu The summary is very clear. Thank you.
All reactions
Does the cache need a new operator state? In the #6729 review, @Yicong-Huang asked whether the CACHE_REUSED state that PR introduced needs to exist at all. Answering here since it is a design decision rather than a PR detail.
It does not. The state was never the goal, only a means: the goal is showing the user which operators and ports were reused, and nobody disagrees with that. The question is only how to carry that information.
A state is the wrong carrier. Checking every consumer: nothing behaves differently for a reused operator than for a completed one, and every place that checks state expects a finished operator to be COMPLETED:
- Workflow level: the completion and recovery checks (ExecutionRuntimeService, ClusterListener, WorkflowService, and the ComputingUnitMaster restart sweep flagged in the review).
- Operator level: the sync API's target-ready check, the reconfiguration lock, and the result panel's finished flag.
So a reused operator reports COMPLETED, and the reuse information travels as one boolean:
reused_from_cacheon OperatorMetrics, carried by the existing stats event to the frontend. The producer lands with Add cache-reuse planning to the scheduler (skeleton generation and schedule assembly) #5884 .- The UI keeps the distinction (Add cache panel and canvas display to the workflow editor #5886 ): the badge and port marks render from the flag and the cache entries. The badge can even show completed and reused together.
- Counts: real numbers from operator_port_cache.tuple_count for cached output ports (feat(dao): add operator_port_cache table #5967 kept it for this).
#6729 is reworked down to exactly this: the flag and the code that carries it, 6 files. The diff there is the concrete proposal.