apache / apache/texera

Add cache-reuse planning and skipped-region execution to the scheduler

Open
#5,884 0 comments 0 reactions 0 assignees View on GitHub
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.