feat(amber): schedule loops with materialized boundaries instead of a fully materialized workflow

[aglinxinyuan]

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	semantics narrowed (name unchanged): forces every link incident to this operator to be materialized, instead of the whole plan. An operator-level flag is required because no port-level declaration can express forced materialization of input links (blocking exists only on OutputPort; dependencies needs a second port)
PhysicalPlan.getForcedMaterializedLinks (new)	getBlockingAndDependeeLinks ∪ links incident to requiresMaterializedExecution 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 (comments/doc); 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)

[github-actions]

Automated Reviewer Suggestions

Based on the git blame history of the changed files, we recommend the following reviewers:

• Contributors with relevant context: @Xiao-zhen-Liu
  You can notify them by mentioning @Xiao-zhen-Liu in a comment.

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 78.94%. Comparing base (f521750) to head (688c5d6).
⚠️ Report is 2 commits behind head on main.

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #6915      +/-   ##
============================================
+ Coverage     78.70%   78.94%   +0.24%     
- Complexity     3740     3794      +54     
============================================
  Files          1161     1161              
  Lines         46084    46082       -2     
  Branches       5110     5109       -1     
============================================
+ Hits          36269    36380     +111     
+ Misses         8206     8077     -129     
- Partials       1609     1625      +16     

Flag	Coverage Δ		*Carryforward flag
access-control-service	70.00% <ø> (ø)
agent-service	76.76% <ø> (ø)		Carriedforward from 69786fe
amber	72.19% <100.00%> (+0.66%)	⬆️
computing-unit-managing-service	20.49% <ø> (ø)
config-service	66.66% <ø> (ø)
file-service	67.21% <ø> (ø)
frontend	82.91% <ø> (ø)		Carriedforward from 69786fe
notebook-migration-service	78.94% <ø> (ø)
pyamber	95.38% <ø> (ø)		Carriedforward from 69786fe
workflow-compiling-service	55.14% <ø> (ø)

*This pull request uses carry forward flags. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[github-actions]

⚠️ Benchmark changes need a look

🟢 4 better · 🔴 1 worse · ⚪ 10 noise (<±5%) · 0 without baseline

Compared against main f521750 benchmarked on this same runner, so the delta is largely free of cross-runner hardware noise. The "7d avg" column still reflects the gh-pages dashboard. Treat <±5% as noise unless repeated.

Dashboard · Run

	config	throughput	MB/s	latency	max Δ latest / 7d
🔴	bs=10 sw=10 sl=64	398	0.243	25,677/28,351/28,351 us	🔴 +18.5% / 🔴 +102.9%
🟢	bs=100 sw=10 sl=64	793	0.484	125,245/143,514/143,514 us	🟢 -14.0% / 🔴 +31.3%
⚪	bs=1000 sw=10 sl=64	910	0.555	1,098,911/1,135,079/1,135,079 us	⚪ within ±5% / 🔴 +9.1%

Baseline details

Latest main f521750 from same runner

config	metric	PR	latest main	7d avg	Δ latest	Δ 7d
bs=10 sw=10 sl=64	throughput	398 tuples/sec	408 tuples/sec	759.72 tuples/sec	-2.5%	-47.6%
bs=10 sw=10 sl=64	MB/s	0.243 MB/s	0.249 MB/s	0.464 MB/s	-2.4%	-47.6%
bs=10 sw=10 sl=64	p50	25,677 us	21,672 us	12,653 us	+18.5%	+102.9%
bs=10 sw=10 sl=64	p95	28,351 us	34,548 us	16,371 us	-17.9%	+73.2%
bs=10 sw=10 sl=64	p99	28,351 us	34,548 us	19,941 us	-17.9%	+42.2%
bs=100 sw=10 sl=64	throughput	793 tuples/sec	810 tuples/sec	966.78 tuples/sec	-2.1%	-18.0%
bs=100 sw=10 sl=64	MB/s	0.484 MB/s	0.494 MB/s	0.59 MB/s	-2.0%	-18.0%
bs=100 sw=10 sl=64	p50	125,245 us	122,331 us	103,654 us	+2.4%	+20.8%
bs=100 sw=10 sl=64	p95	143,514 us	166,809 us	109,308 us	-14.0%	+31.3%
bs=100 sw=10 sl=64	p99	143,514 us	166,809 us	116,369 us	-14.0%	+23.3%
bs=1000 sw=10 sl=64	throughput	910 tuples/sec	916 tuples/sec	997.95 tuples/sec	-0.7%	-8.8%
bs=1000 sw=10 sl=64	MB/s	0.555 MB/s	0.559 MB/s	0.609 MB/s	-0.7%	-8.9%
bs=1000 sw=10 sl=64	p50	1,098,911 us	1,095,482 us	1,007,348 us	+0.3%	+9.1%
bs=1000 sw=10 sl=64	p95	1,135,079 us	1,126,158 us	1,050,149 us	+0.8%	+8.1%
bs=1000 sw=10 sl=64	p99	1,135,079 us	1,126,158 us	1,079,497 us	+0.8%	+5.1%

Raw CSV

config_idx,batch_size,schema_width,string_len,num_batches,total_ms,total_tuples,total_bytes,tuples_per_sec,mb_per_sec,lat_p50_us,lat_p95_us,lat_p99_us
0,10,10,64,20,502.79,200,128000,398,0.243,25677.15,28350.89,28350.89
1,100,10,64,20,2523.13,2000,1280000,793,0.484,125244.76,143513.86,143513.86
2,1000,10,64,20,21977.45,20000,12800000,910,0.555,1098910.55,1135079.43,1135079.43