Add cache-reuse planning and skipped-region execution to the scheduler
- Dominant language
- Scala
- Stars
- 314
- Forks
- 187
- Avg merge
- 1d 21h
- Merged PRs (30d)
- 214
Description
### Task Summary
Part of #5881 (design: #5880).
Let the scheduler reuse matched ports. Before `CostBasedScheduleGenerator` runs, remove the operators whose results are reused, leaving the run-skeleton; schedule the run-skeleton as today; then combine it with the skipped regions into one schedule. When the run starts, a skipped region completes at once without workers and reports statistics from the saved results. This is the core of the feature and needs the most review.
**Scope**
Planning:
- `WorkflowSettings.cachedOutputs`: the matched ports for a run and their saved result locations, empty by default. Only the server's cache lookup (#5885) may fill it. `WorkflowSettings` arrives from the client with the execute request, so a value sent by the client must never be used: it names storage locations the run would read.
- `CacheReusePreSchedulingStep`: the reuse planner and skeleton generator. Starting from the outputs the run needs (sinks plus any ports the user asked to view), it keeps an operator if any of its needed outputs is not a matched port; otherwise the saved results supply its outputs and the operator is removed. The result is the run-skeleton, the skipped regions, and the saved result locations for matched inputs and outputs.
- `PreSchedulingHints`: a small carrier for those result-location overrides; empty by default.
- `CostBasedScheduleGenerator`: two new optional inputs (the hints and the skipped regions), both empty by default. When empty, the schedule is built exactly as today. When present, it schedules the run-skeleton and places the skipped regions ahead of the regions that run.
- `Region` gains a `cached` flag and the output port config gains the saved tuple count, both defaulting to today's behavior.
Running a skipped region:
- `RegionExecutionManager` runs the normal path for a regular region. For a skipped region it starts no workers, records its operators as COMPLETED with `reused_from_cache` set (#5883), and passes the saved result locations downstream.
- Statistics: an output port of a skipped operator reports the saved tuple count; an input port reports unknown, not zero. The statistics are persisted like any other run's, so a fully reused run still shows them in execution history.
- Completion: a fully reused run has no workers, so no worker ever reports completion. The completion signal goes through `WorkflowExecutionManager`, so it is sent exactly once whether or not workers ran, and the final result update still runs when a run finishes before the periodic result update has started.
**Why this is safe when there are no matched ports.** When `cachedOutputs` is empty, `CacheReusePreSchedulingStep` returns early: the run-skeleton is the whole plan, the hints are empty, and there are no skipped regions. `CostBasedScheduleGenerator` then builds the schedule as it does today, and every region has `cached = false`, so `RegionExecutionManager` runs the normal path.
The planning diagrams in the design Discussion show a cost estimator inside the scheduler; here the cost-based choice is not used (every matched port is reused). It is the existing estimator, run only on the run-skeleton.
**Depends on.** #5883 (the `reused_from_cache` flag), for running a skipped region; planning depends on nothing. Can be done while inactive, since nothing writes cache entries until #5885.
**Out of scope.** No cost-based decision about whether to reuse (full reuse only). Reading and writing cache entries at submission and completion (#5885). Showing reuse in the UI (#5886).
### Task Type
- [ ] Refactor / Cleanup
- [ ] DevOps / Deployment / CI
- [ ] Testing / QA
- [ ] Documentation
- [ ] Performance
- [x] Other
Contributor guide
Research direction
Start by reading CacheReusePreSchedulingStep, PreSchedulingHints, CostBasedScheduleGenerator, Region, RegionExecutionManager, and WorkflowExecutionManager, along with dependency #5883. Trace the no-cache path first, then the fully reused path and its statistics and completion behavior. Done means empty cache inputs preserve today's scheduling and execution, while matched outputs produce skipped regions, saved-result statistics, downstream locations, and exactly one completion signal.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- scala
- Domain
- backend, distributed-systems
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100