GitHub user Xiao-zhen-Liu added a comment to the discussion: Design and merge plan: operator output port result cache (MVP)
**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 three PRs: 1. #8731 2. #8732 3. #8733, which sits on the other two Nothing sets a run's matched results until the lookup PR, so all three 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. <img width="1714" height="1225" alt="Example: logical plan, physical plan with saved results, and the skeleton" src="https://github.com/user-attachments/assets/d24c77f8-cacd-4e49-bbc8-6e49fff0dcc9"; /> 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. <img width="1454" height="566" alt="Split runs for B, while A reads Split's saved port 0" src="https://github.com/user-attachments/assets/bbbbfdf4-1f94-49d6-a2ed-8994632cc6d6"; /> 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 <img width="1923" height="1400" alt="How the scheduler uses the skeleton" src="https://github.com/user-attachments/assets/4f3ec0a0-81b3-441e-935d-07e51adbba05"; /> - 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`, the reconfiguration scope in `FriesReconfigurationAlgorithm`) 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): <img width="2387" height="1000" alt="Schedule for the example: skip regions in level 0, generated regions after" src="https://github.com/user-attachments/assets/4e0ac199-352b-4a13-82b2-75ebd360f7fe"; /> With no matched results, the scheduler runs today's code: the whole plan goes to the generator. GitHub link: https://github.com/apache/texera/discussions/5880#discussioncomment-18646157 ---- This is an automatically sent email for [email protected]. To unsubscribe, please send an email to: [email protected]
