feat(amber): add cache-reused operator state and cached-region statistics handling

[Xiao-zhen-Liu]

What changes were proposed in this PR?

This adds the CACHE_REUSED workflow-aggregated state and the statistics handling for a cached region. It is the state-and-stats slice of the operator output port result cache (MVP). It lands dormant and does not change behavior on an empty cache.

• Add CACHE_REUSED = 11 to the WorkflowAggregatedState enum, mapped in Utils to a display string (CacheReused), a parse case, and DB status code 6.
• ExecutionUtils.aggregateStates takes an optional cachedState; when every state in the group equals it, the group rolls up to CACHE_REUSED. It defaults to None, so every existing caller is unchanged. Only the operator-level aggregateMetrics passes it.
• ExecutionUtils.aggregatePortMetrics treats a negative count/size as an unknown value and keeps the aggregated port metrics unknown; ExecutionUtils.sumNonNegative skips those sentinels when ExecutionStatsService sums operator totals. A cached input port has no live tuple count, so it carries this unknown marker.

Empty-cache safety: with an empty cache nothing produces CACHE_REUSED or a negative tuple count, so cachedState stays None (the new branch is unreachable) and every count is non-negative (both sentinel paths are identity). The enum value is appended with no renumber, and the only code that matches on the enum (Utils) has catch-alls. The engine behaves identically to today until a later PR marks operators as reused from cache.

Any related issues, documentation, discussions?

Closes #5883. Part of #5881 (operator output port result cache, MVP). Design discussion: #5880.

How was this PR tested?

Added unit tests. ExecutionUtilsSpec covers the cachedState branch (all-cached rolls up to CACHE_REUSED, branch precedence over completed/terminated/running, and the empty-cache regression that the optional argument is byte-identical to the existing overload when no cached state is present), the -1 unknown-port sentinel in aggregatePortMetrics, its propagation through aggregateMetrics, and sumNonNegative. UtilsSpec covers the display and parse round-trip, the status code 6, and that it is distinct from COMPLETED. The two specs pass (63 cases), and scalafmtCheckAll is clean.

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

Generated-by: Claude Opus 4.8 (Claude Code)

[github-actions]

Automated Reviewer Suggestions

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

• Contributors with relevant context: @Yicong-Huang, @carloea2, @aglinxinyuan
  You can notify them by mentioning @Yicong-Huang, @carloea2, @aglinxinyuan in a comment.

[Xiao-zhen-Liu]

@carloea2 this is PR 2 of the operator output port result cache (MVP): the completed-from-cache operator state (CACHE_REUSED) plus cached-region statistics handling (#5883). Would appreciate your eyes on it. It lands dormant, with the empty-cache == main safety property held.

[Xiao-zhen-Liu]

@Yicong-Huang would you be able to review this one? (I couldn't add you via the reviewer field from a fork PR.) It's the cache state + stats slice of the operator output port result cache (#5883, under #5881), lands dormant with the empty-cache == main safety property.

[github-actions]

⚠️ Benchmark changes need a look

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

Compared against main d444fe0 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	565	0.345	17,269/22,200/22,200 us	🟢 -35.8% / 🔴 +40.4%
🔴	bs=100 sw=10 sl=64	1,192	0.727	83,405/114,541/114,541 us	🔴 +30.3% / 🟢 +18.1%
⚪	bs=1000 sw=10 sl=64	1,440	0.879	699,297/733,031/733,031 us	⚪ within ±5% / 🟢 +37.9%

Baseline details

Latest main d444fe0 from same runner

config	metric	PR	latest main	7d avg	Δ latest	Δ 7d
bs=10 sw=10 sl=64	throughput	565 tuples/sec	556 tuples/sec	769.76 tuples/sec	+1.6%	-26.6%
bs=10 sw=10 sl=64	MB/s	0.345 MB/s	0.339 MB/s	0.47 MB/s	+1.8%	-26.6%
bs=10 sw=10 sl=64	p50	17,269 us	16,738 us	12,663 us	+3.2%	+36.4%
bs=10 sw=10 sl=64	p95	22,200 us	34,590 us	15,813 us	-35.8%	+40.4%
bs=10 sw=10 sl=64	p99	22,200 us	34,590 us	19,001 us	-35.8%	+16.8%
bs=100 sw=10 sl=64	throughput	1,192 tuples/sec	1,288 tuples/sec	1,009 tuples/sec	-7.5%	+18.1%
bs=100 sw=10 sl=64	MB/s	0.727 MB/s	0.786 MB/s	0.616 MB/s	-7.5%	+18.0%
bs=100 sw=10 sl=64	p50	83,405 us	77,894 us	99,444 us	+7.1%	-16.1%
bs=100 sw=10 sl=64	p95	114,541 us	87,904 us	105,012 us	+30.3%	+9.1%
bs=100 sw=10 sl=64	p99	114,541 us	87,904 us	116,467 us	+30.3%	-1.7%
bs=1000 sw=10 sl=64	throughput	1,440 tuples/sec	1,434 tuples/sec	1,044 tuples/sec	+0.4%	+37.9%
bs=1000 sw=10 sl=64	MB/s	0.879 MB/s	0.875 MB/s	0.637 MB/s	+0.5%	+37.9%
bs=1000 sw=10 sl=64	p50	699,297 us	698,595 us	964,034 us	+0.1%	-27.5%
bs=1000 sw=10 sl=64	p95	733,031 us	730,914 us	1,007,497 us	+0.3%	-27.2%
bs=1000 sw=10 sl=64	p99	733,031 us	730,914 us	1,040,936 us	+0.3%	-29.6%

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,354.15,200,128000,565,0.345,17268.55,22199.82,22199.82
1,100,10,64,20,1677.99,2000,1280000,1192,0.727,83405.42,114541.37,114541.37
2,1000,10,64,20,13891.83,20000,12800000,1440,0.879,699297.10,733031.39,733031.39

[Ma77Ball]

New status code 6 is not mirrored to the frontend in ngbd-modal-workflow-executions.component.ts (mentioned in the doc comments above)

/**
* @param state indicates the workflow state
* @return code indicates the status of the execution in the DB it is 0 by default for any unused states.
* This code is stored in the DB and read in the frontend.
* If these codes are changed, they also have to be changed in the frontend ngbd-modal-workflow-executions.component.ts
*/

Please add the CacheReused case there too.

[Yicong-Huang]

I will have to check in on weekend.