Design and merge plan: operator output port result cache (MVP)

[Xiao-zhen-Liu]

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.

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:

1. 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.
2. Cache state and statistics (Carry operator cache-reuse status to the frontend #5883): a "completed from cache" operator state and the matching statistics handling.
3. Scheduler (Add cache-reuse planning and skipped-region execution to the scheduler #5884): the reuse planner (full reuse), skeleton generator, and the change that schedules the run-skeleton and combines it with the skipped regions.
4. 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.
5. 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.

[chenlica]

@Xiao-zhen-Liu Thanks for the great summary. Please also describe our plan to manage the lifecycle of the cached results.

[chenlica]

@Xiao-zhen-Liu Please chime in.

[Xiao-zhen-Liu]

@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.

[chenlica]

@Xiao-zhen-Liu The summary is very clear. Thank you.

[Xiao-zhen-Liu]

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_cache on OperatorMetrics, carried by the existing stats event to the frontend. The producer lands with Add cache-reuse planning and skipped-region execution to the scheduler #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.

[Xiao-zhen-Liu]

Design update: planning cache reuse before scheduling

The first three PRs of the plan above are merged: the cache key (#5966), the operator_port_cache table (#5967) and the reuse flag on operator metrics (#6729). The scheduler part (#5884) now goes up as four PRs:

1. feat(amber): decide what a run skips for cache reuse #8731
2. feat(amber): schedule saved-result reads for cache reuse #8732
3. feat(amber): schedule a run with cache reuse #8733, which sits on the other two
4. refactor(amber): use region links in Fries for cache reuse #8734, which needs none of the others and only has to land before the lookup PR

Nothing sets a run's matched results until the lookup PR, so all four leave every run on main unchanged. Below is the design they introduce, and what changed from the post above.

Important

Assumption: full reuse. The design always reuses a usable saved result. Every retained operator that reads a port with a usable saved result reads that saved result, and an operator is skipped whenever usable saved results cover every port the run needs from it. There is no cost-based choice between reading a saved result and computing it again, even where computing would be cheaper. That choice (cost-based reuse planning) stays future work, as in the post above.

Deciding what a run skips

A run's matched results are the saved results found for its output ports (WorkflowContext.matchedResults). Only the server sets them; they are not part of the run request. SkeletonGenerator turns them into the run's skeleton, which marks every operator and link of the workflow for this run:

• An output port is a required output if the run must store it (the user views its operator's results, or the operator has no outgoing link), or if it has no link.
• A port is needed if it is a required output or feeds an operator that runs.
• An operator is retained, meaning it runs in this execution, if it has no output ports or if one of its needed ports has no usable saved result. A retained operator computes all of its outputs.
• Every other operator is skipped, and every link into a skipped operator is skipped.
• Every link from a port with a usable saved result into a retained operator reads the saved result.

The example below applies the rules to a join. A, B and G are viewed; B and D have saved results. B and D are skipped, and C-Build and E read their saved results. A runs because its viewed result has no saved copy, and the link from A into the skipped B is skipped.

Two changes from the post above:

• The skeleton (RunSkeleton) now marks the whole workflow, the skipped part included. The post above described it as only the part that still needs to run; that part is now the skeleton's retained part, which the scheduler builds from the marks.
• A saved port is read from storage by every retained operator it feeds, even when its own operator runs for another port. This follows from full reuse. When the run must store that port (the user views it), the run still stores and shows its own result; only the operators the port feeds read the saved copy.

A saved result is unusable, and the run just does not reuse it, when its port is not in the plan or its location is not that port's in this workflow and this run's warehouse. A plan with a loop reuses nothing. An unusable result never fails a run.

Scheduling the retained part

• The scheduler hands CostBasedScheduleGenerator the retained part: the plan's own operators that run and the links between them, without the links that read a saved result. The search, the regions and the storage of materialized links work as today.
• The generator also gets, for each input port on such a link, the saved location it reads (CacheReadInputs). Its second pass adds these next to the locations from materialized links. The resource allocator partitions an operator that reads a saved result as if it read a materialized link, so the retained part is partitioned as it would be in the whole plan.
• Operators keep their own link lists, which still name the links that read a saved result. Code that works on one region (Region.getSourceOperators in feat(amber): schedule saved-result reads for cache reuse #8732, the reconfiguration scope in FriesReconfigurationAlgorithm in refactor(amber): use region links in Fries for cache reuse #8734) now counts only the region's own links. Without saved results their answers do not change.
• Skipped operators become skip regions (Region.skipped), one per connected group, all in a new first level ahead of the generated levels, with ids above the generated ones. Each skip region records the saved results its ports provide. A run that skips every operator has only that level. Running a skip region, which starts no workers, comes in a later PR.
• The scheduler keeps the full plan, so code that looks up an operator still finds a skipped one.

A schedule for the example (the generated regions are the ones the search would typically pick):

With no matched results, the scheduler runs today's code: the whole plan goes to the generator.

[Xiao-zhen-Liu]

Design discussion of October 1: the open PRs are on hold

On October 1, @chenlica, @Yicong-Huang, @parshimers, @sshiv012 and I went over the design update above. The outcome affects how the work goes on, so I am posting it here.

Under full reuse (what the open PRs build). We discussed what a skipped operator and a skip region should do, and whether they add new region states. As I understood it, @Yicong-Huang's main requirement was that skipping should not add new region states, which the current design already meets.

Beyond full reuse. The main question was whether the current two-layer design can be extended past full reuse in a way that is compatible with other work. In the two-layer design, the skeleton decides what runs and then CostBasedScheduleGenerator plans the retained part, so the reuse decision is made outside the generator. @Yicong-Huang proposed a one-layer alternative:

• Treat reading a saved result as one more resource allocation choice on a materialized link, ideally handled only in the resource allocator and the cost estimator.
• The resource allocator makes the reuse decision: reusing means allocating no workers.
• CostBasedScheduleGenerator makes no reuse decision and still only chooses between pipelining and materialization.

In my view, both designs explore the same space of plans. We agreed that the two-layer design cannot be extended into the one-layer form for the general setting.

Outcome. There was no consensus that the one-layer alternative has been investigated enough to rule it out. Until it has, #8731, #8732, #8733 and #8734 are on hold, and the rest of the series waits. I will find out whether the one-layer approach works and how much complexity it adds to the code base, and whether the two-layer design can be generalized in a way that is compatible with the other work. I will discuss both with @chenlica and post the findings here.

Background on how CostBasedScheduleGenerator forms regions: the Pasta paper.