feat(execution): add PhysicalOp.requiresMaterializedExecution, consumed by the scheduler

[aglinxinyuan]

What changes were proposed in this PR?

Lets an operator declare it can only run under a fully-materialized schedule, and has the scheduler honor it:

• PhysicalOp gains requiresMaterializedExecution: Boolean = false (+ a withRequiresMaterializedExecution builder). It is a physical-execution property, so it lives on the physical op.
• CostBasedScheduleGenerator consumes it: when any physical op requires materialized execution it forces a fully-materialized schedule regardless of the requested execution mode; otherwise the existing PIPELINED/MATERIALIZED logic runs unchanged.

Default false ⇒ dormant and behavior-preserving: no operator requires it yet, so the scheduler's effective mode is unchanged today. The loop operators set the flag on their physical op.

Any related issues, documentation, discussions?

Resolves #5719 (sub-issue of #4442 "Introduce for loop"). Split out of #5700. Reflects the review discussion with @Yicong-Huang: the property belongs on PhysicalOp, and it is consumed by the scheduler.

How was this PR tested?

WorkflowCoreTypesSpec covers the PhysicalOp.requiresMaterializedExecution default + builder. WorkflowExecutionService/Test/compile, scalafixAll --check, and scalafmtCheckAll pass locally. The scheduler consumer is exercised end-to-end by the loop integration tests once the loop operators (which set the flag) land.

Was this PR authored or co-authored using generative AI tooling?

Co-authored with Claude Opus 4.8 in compliance with ASF.

[Copilot]

Pull request overview

Adds an operator-declared capability flag to enforce execution-mode constraints at workflow submission time, preventing workflows that require materialization from being run in PIPELINED mode (avoiding UI/engine mode mismatches).

Changes:

• Added LogicalOp.requiresMaterializedExecution: Boolean = false for operators to declare a MATERIALIZED-only requirement.
• Introduced WorkflowExecutionService.validateExecutionMode(operators, settings) to reject incompatible submissions with an actionable error message.
• Added ScalaTest coverage for the default flag value and validate/reject/accept paths.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated no comments.

File	Description
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/LogicalOp.scala	Adds requiresMaterializedExecution flag (default false) for operator-declared mode requirements.
amber/src/main/scala/org/apache/texera/web/service/WorkflowExecutionService.scala	Implements and invokes submission-time execution mode validation keyed off the new flag.
amber/src/test/scala/org/apache/texera/web/service/WorkflowExecutionServiceSpec.scala	Adds unit tests covering the flag default and validation behavior.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 53.91%. Comparing base (44df4f7) to head (8cbc747).
✅ All tests successful. No failed tests found.

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #5720      +/-   ##
============================================
- Coverage     53.96%   53.91%   -0.06%     
- Complexity     2782     2783       +1     
============================================
  Files          1100     1099       -1     
  Lines         42588    42567      -21     
  Branches       4583     4578       -5     
============================================
- Hits          22984    22949      -35     
- Misses        18268    18285      +17     
+ Partials       1336     1333       -3     

Flag	Coverage Δ		*Carryforward flag
access-control-service	70.44% <ø> (ø)
agent-service	34.36% <ø> (ø)		Carriedforward from 9032260
amber	55.27% <100.00%> (+<0.01%)	⬆️
computing-unit-managing-service	1.65% <ø> (ø)
config-service	57.35% <ø> (ø)
file-service	57.06% <ø> (ø)
frontend	48.02% <ø> (-0.14%)	⬇️	Carriedforward from 9032260
pyamber	90.14% <ø> (-0.05%)	⬇️	Carriedforward from 9032260
python	90.80% <ø> (ø)		Carriedforward from 9032260
workflow-compiling-service	58.69% <ø> (ø)

*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

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

Compared against main 44df4f7 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	417	0.254	24,096/28,749/28,749 us	🟢 -7.6% / 🟢 -17.8%
🔴	bs=100 sw=10 sl=64	947	0.578	103,416/135,293/135,293 us	🔴 +5.9% / 🟢 -7.9%
⚪	bs=1000 sw=10 sl=64	1,107	0.676	906,134/959,592/959,592 us	⚪ within ±5% / 🟢 -6.8%

Baseline details

Latest main 44df4f7 from same runner

config	metric	PR	latest main	7d avg	Δ latest	Δ 7d
bs=10 sw=10 sl=64	throughput	417 tuples/sec	445 tuples/sec	410.82 tuples/sec	-6.3%	+1.5%
bs=10 sw=10 sl=64	MB/s	0.254 MB/s	0.272 MB/s	0.251 MB/s	-6.6%	+1.3%
bs=10 sw=10 sl=64	p50	24,096 us	23,608 us	23,785 us	+2.1%	+1.3%
bs=10 sw=10 sl=64	p95	28,749 us	31,112 us	34,980 us	-7.6%	-17.8%
bs=10 sw=10 sl=64	p99	28,749 us	31,112 us	34,980 us	-7.6%	-17.8%
bs=100 sw=10 sl=64	throughput	947 tuples/sec	961 tuples/sec	891.94 tuples/sec	-1.5%	+6.2%
bs=100 sw=10 sl=64	MB/s	0.578 MB/s	0.586 MB/s	0.544 MB/s	-1.4%	+6.2%
bs=100 sw=10 sl=64	p50	103,416 us	105,553 us	112,277 us	-2.0%	-7.9%
bs=100 sw=10 sl=64	p95	135,293 us	127,787 us	139,802 us	+5.9%	-3.2%
bs=100 sw=10 sl=64	p99	135,293 us	127,787 us	139,802 us	+5.9%	-3.2%
bs=1000 sw=10 sl=64	throughput	1,107 tuples/sec	1,116 tuples/sec	1,041 tuples/sec	-0.8%	+6.3%
bs=1000 sw=10 sl=64	MB/s	0.676 MB/s	0.681 MB/s	0.635 MB/s	-0.7%	+6.4%
bs=1000 sw=10 sl=64	p50	906,134 us	892,761 us	972,714 us	+1.5%	-6.8%
bs=1000 sw=10 sl=64	p95	959,592 us	937,723 us	1,023,057 us	+2.3%	-6.2%
bs=1000 sw=10 sl=64	p99	959,592 us	937,723 us	1,023,057 us	+2.3%	-6.2%

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,480.08,200,128000,417,0.254,24095.81,28748.53,28748.53
1,100,10,64,20,2111.39,2000,1280000,947,0.578,103415.58,135292.91,135292.91
2,1000,10,64,20,18062.18,20000,12800000,1107,0.676,906133.86,959591.51,959591.51

[Yicong-Huang]

had a discussion with @aglinxinyuan, two comments:

1. I think this property belongs to the physical execution and should be a property of a physical op, not logical.
2. From the PR's perspective, it is not iideal to introduce a property without having a consumer of the property. To keep the PR scope small, the consumer can be simple. For instance, I assume this property should be consumed by scheduler (?), so a logic like this

if any(physicalOp.requireMaterializedExecution):
    // TODO: require non-pipeline mode

 // original pipeline execution mode logic
   .....

would suffice. Note that this simple consumer can help reviewer to understand how this property will be used.

[aglinxinyuan]

Addressed both points in 01c8603:

1. Moved the flag to PhysicalOp (it's a physical-execution property) + a withRequiresMaterializedExecution builder; removed it from LogicalOp.
2. Real consumer in the scheduler: CostBasedScheduleGenerator now forces a fully-materialized schedule when any physical op requires it (otherwise the existing PIPELINED/MATERIALIZED logic runs unchanged). Dropped the submission-time WorkflowExecutionService.validateExecutionMode.

Default false, so it's dormant and behavior-preserving today; the loop operators set it on their PhysicalOp. PR is now 3 files: PhysicalOp, CostBasedScheduleGenerator, WorkflowCoreTypesSpec.

[Yicong-Huang]

LGTM, left one comment inline

[Yicong-Huang]

Thanks. Let's get @Xiao-zhen-Liu's approval before merging

[chenlica]

@Xiao-zhen-Liu Please chime in.

[Yicong-Huang]

@Xiao-zhen-Liu please chime in. this PR is getting old.

[Xiao-zhen-Liu]

LGTM in general, left a few minor comments.

[github-actions]

Automated Reviewer Suggestions

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

• Contributors with relevant context: @Ma77Ball, @SarahAsad23
  You can notify them by mentioning @Ma77Ball, @SarahAsad23 in a comment.