aglinxinyuan opened a new pull request, #6915:
URL: https://github.com/apache/texera/pull/6915
### What changes were proposed in this PR?
**Loops no longer force the whole workflow to be fully materialized — only
their boundary links.**
Today, a single loop operator anywhere in the plan makes
`CostBasedScheduleGenerator` coerce the requested execution mode to
`MATERIALIZED`: every link is materialized, every operator becomes its own
region, and the cost-based search never runs. The runtime, however, only needs
the links *incident to the loop operators* to be materialized:
| Runtime mechanism | What it actually needs |
|---|---|
| Loop-back write address (`loopStartStateUris`) | the single link **into
LoopStart** materialized (`InputPortConfig` with one storage pair) |
| LoopEnd result accumulation (`reuseStorage`) | the link **out of LoopEnd**
materialized |
| Region re-execution per iteration | the body separated from its neighbors
by materialized edges (links **out of LoopStart** / **into LoopEnd**) |
| Back-edge jump (`JumpToOperatorRegion`) | rewinds the schedule by *level*
— indifferent to region granularity |
This PR makes the scheduler force exactly those links and optimize the rest
of the plan under the requested mode:
```
before (any loop => whole plan materialized, no search)
scan ==M==> LS ==M==> f1 ==M==> f2 ==M==> LE ==M==> sink regions: 6
after (only loop-boundary links forced; body + rest optimized)
scan ==M==> LS ==M==> f1 --pipe--> f2 ==M==> LE ==M==> sink regions: 5
```
Mechanics:
| Piece | Change |
|---|---|
| `PhysicalOp.requiresMaterializedExecution` | renamed to
**`requiresMaterializedBoundary`**: forces every incident link to be
materialized (a property no port-level `blocking`/`dependencies` declaration
can express for *input* links) |
| `PhysicalPlan.getForcedMaterializedLinks` (new) |
`getBlockingAndDependeeLinks` ∪ links incident to boundary operators;
`getNonBridgeNonBlockingLinks` (the search candidates) now excludes them |
| `CostBasedScheduleGenerator` | all search paths union
`getForcedMaterializedLinks` instead of `getBlockingAndDependeeLinks`; the
whole-plan coercion (`effectiveExecutionMode`) is removed; user-requested
`MATERIALIZED` mode is unchanged |
| `WorkflowExecutionManager` | the `loopStartStateUris` guard message
updated (the input edge of a LoopStart must be materialized — always true now
by construction) |
Side effect: this also closes a latent hole — on search **timeout**, the
fallback `bottomUpSearch()` previously ignored the loop coercion entirely (a
loop workflow could be scheduled with un-materialized boundaries). The forced
links are now embedded in every search path, including the fallback.
With loop boundaries forced, a multi-operator loop body becomes **one
pipelined region** that is re-executed per iteration (workers respawn together;
the loop state crosses body hops in-band), and operators outside the loop
pipeline normally.
### Any related issues, documentation, discussions?
Follow-up to the loop feature #5700, which shipped whole-plan
materialization as the minimal correct behavior (the `PhysicalOp` field comment
explicitly called restricting it to the operator's own boundaries "a possible
future optimization" — this PR).
### How was this PR tested?
New tests (written first, failing on the old scheduler):
- `PhysicalPlanSpec`: `getForcedMaterializedLinks` includes exactly the
incident links of boundary operators; reduces to blocking+dependee links
otherwise; forced links leave the search-candidate set.
- `CostBasedScheduleGeneratorSpec`: a `csv → LoopStart → keyword → keyword2
→ LoopEnd → keyword3` plan under the search yields exactly 5 regions — the 4
boundary links materialized, the two body operators sharing one **pipelined**
region, `state` empty (no extra materializations needed).
- `LoopIntegrationSpec` (e2e, PIPELINED mode): the 3-iteration loop still
accumulates exactly 3 rows, and a `LoopStart → Limit → Sleep → LoopEnd` body
pipelines internally (one two-operator region re-executed per iteration) — both
previously only ran under the coerced fully-materialized schedule.
Updated: `WorkflowCoreTypesSpec`, `LoopStartOpDescSpec`, `LoopEndOpDescSpec`
(rename); the two `effectiveExecutionMode` tests are superseded by the
loop-plan test above.
Local runs: `WorkflowCore` 62/62, `WorkflowOperator` 27/27,
`CostBasedScheduleGeneratorSpec` 10/10, `LoopIntegrationSpec` 6/6 (4 existing
MATERIALIZED cases + 2 new PIPELINED cases).
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Fable 5)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
