feat(workflow-core): add operator output port cache key

[Xiao-zhen-Liu]

What changes were proposed in this PR?

Adds CacheKeyUtil (in workflow-core), which computes a deterministic cache key for
an output port from the upstream sub-DAG that produces it. The sub-DAG is the port's
operator plus its transitive upstream, obtained from a new
PhysicalPlan.getTransitiveUpstreamSubPlan(opId); computeCacheKey takes that sub-DAG.
For a target output port, the key is built from:

• the target output port's identity;
• every operator in the sub-DAG and, for each: its physical and logical operator id,
  its execution init info (the concrete execution definition the engine runs), and the
  schema of each of its output ports;
• the edges among those operators, including the specific ports each edge connects.

These are written to a JSON document with sorted keys and a sorted operator and edge
order, so the same upstream computation always serializes to the same bytes and hashes
(SHA-256) to the same key. Any upstream change (a parameter edit, a rewiring, a schema
change) produces a different key; changes elsewhere in the workflow do not.

Matching is collision-safe: isSameComputation compares the hash first and, on a hash
match, confirms with the full JSON, so a hash collision never reuses a result the port
was not computed from.

This is the first piece of the operator output port result cache. CacheKeyUtil is a
pure function with no database or storage dependency, and nothing calls it yet, so it
changes no existing behavior. The cache table (companion PR) and the service that uses
both land separately.

Any related issues, documentation, discussions?

Closes #5968. Part of the storage foundation #5882 (umbrella #5881). Design discussion: #5880.

How was this PR tested?

New unit tests in workflow-core (built from PhysicalPlans directly, no engine test
helpers, so the logic is covered in its own module): CacheKeyUtilSpec covers a stable
key, sensitivity to an upstream change, different ports, a source with no upstream,
upstream-only scope, and the three isSameComputation cases; the new
PhysicalPlan.getTransitiveUpstreamSubPlan is tested in WorkflowCoreTypesSpec. Run with
sbt "WorkflowCore/testOnly *CacheKeyUtilSpec *WorkflowCoreTypesSpec" — all pass.

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, @aglinxinyuan
  You can notify them by mentioning @Yicong-Huang, @aglinxinyuan in a comment.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 95.83333% with 2 lines in your changes missing coverage. Please review.
✅ Project coverage is 56.69%. Comparing base (a24d1d1) to head (5f0c570).
⚠️ Report is 32 commits behind head on main.

Files with missing lines	Patch %	Lines
...exera/amber/core/workflow/cache/CacheKeyUtil.scala	95.55%	1 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #5966      +/-   ##
============================================
+ Coverage     56.28%   56.69%   +0.40%     
- Complexity     2992     3024      +32     
============================================
  Files          1120     1122       +2     
  Lines         43217    43342     +125     
  Branches       4662     4670       +8     
============================================
+ Hits          24326    24574     +248     
+ Misses        17472    17326     -146     
- Partials       1419     1442      +23     

Flag	Coverage Δ		*Carryforward flag
access-control-service	70.00% <ø> (ø)
agent-service	44.95% <ø> (ø)		Carriedforward from 0207cab
amber	58.74% <95.83%> (+0.93%)	⬆️
computing-unit-managing-service	0.00% <ø> (ø)
config-service	52.30% <ø> (+0.74%)	⬆️
file-service	62.81% <ø> (+3.79%)	⬆️
frontend	49.33% <ø> (ø)		Carriedforward from 0207cab
notebook-migration-service	78.57% <ø> (ø)
pyamber	90.20% <ø> (ø)		Carriedforward from 0207cab
python	90.76% <ø> (ø)		Carriedforward from 0207cab
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.

[Xiao-zhen-Liu]

@carloea2 could you help review this one when you get a chance? Thanks!

[github-actions]

⚠️ Benchmark changes need a look

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

Compared against main a24d1d1 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	370	0.226	24,541/38,381/38,381 us	🔴 -9.6% / 🔴 +154.7%
⚪	bs=100 sw=10 sl=64	815	0.497	120,954/141,566/141,566 us	⚪ within ±5% / 🔴 +31.6%
⚪	bs=1000 sw=10 sl=64	938	0.572	1,069,318/1,105,838/1,105,838 us	⚪ within ±5% / 🔴 +8.3%

Baseline details

Latest main a24d1d1 from same runner

config	metric	PR	latest main	7d avg	Δ latest	Δ 7d
bs=10 sw=10 sl=64	throughput	370 tuples/sec	409 tuples/sec	777.62 tuples/sec	-9.5%	-52.4%
bs=10 sw=10 sl=64	MB/s	0.226 MB/s	0.25 MB/s	0.475 MB/s	-9.6%	-52.4%
bs=10 sw=10 sl=64	p50	24,541 us	23,928 us	12,612 us	+2.6%	+94.6%
bs=10 sw=10 sl=64	p95	38,381 us	37,635 us	15,070 us	+2.0%	+154.7%
bs=10 sw=10 sl=64	p99	38,381 us	37,635 us	18,360 us	+2.0%	+109.0%
bs=100 sw=10 sl=64	throughput	815 tuples/sec	833 tuples/sec	988.31 tuples/sec	-2.2%	-17.5%
bs=100 sw=10 sl=64	MB/s	0.497 MB/s	0.508 MB/s	0.603 MB/s	-2.2%	-17.6%
bs=100 sw=10 sl=64	p50	120,954 us	118,143 us	101,066 us	+2.4%	+19.7%
bs=100 sw=10 sl=64	p95	141,566 us	145,038 us	107,594 us	-2.4%	+31.6%
bs=100 sw=10 sl=64	p99	141,566 us	145,038 us	115,830 us	-2.4%	+22.2%
bs=1000 sw=10 sl=64	throughput	938 tuples/sec	938 tuples/sec	1,019 tuples/sec	0.0%	-8.0%
bs=1000 sw=10 sl=64	MB/s	0.572 MB/s	0.573 MB/s	0.622 MB/s	-0.2%	-8.1%
bs=1000 sw=10 sl=64	p50	1,069,318 us	1,061,623 us	986,982 us	+0.7%	+8.3%
bs=1000 sw=10 sl=64	p95	1,105,838 us	1,113,786 us	1,028,491 us	-0.7%	+7.5%
bs=1000 sw=10 sl=64	p99	1,105,838 us	1,113,786 us	1,058,493 us	-0.7%	+4.5%

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,540.55,200,128000,370,0.226,24541.19,38380.85,38380.85
1,100,10,64,20,2455.43,2000,1280000,815,0.497,120953.52,141565.63,141565.63
2,1000,10,64,20,21322.96,20000,12800000,938,0.572,1069317.58,1105838.07,1105838.07

[Yicong-Huang]

Two questions:

1. how will cache key be used? is testing reflecting its intended usage?
2. how come the coverage still says 0%?

[Yicong-Huang]

Thanks! Please see my comments above. Also left inline comments

[Xiao-zhen-Liu]

Thanks @Yicong-Huang — pushed changes addressing these (replies inline). Summary: renamed to StorageCacheKey / isSameComputation / buildJSONPayload; moved the upstream collection into PhysicalPlan.getTransitiveUpstreamSubPlan(opId) and dropped SubDag so computeCacheKey takes a sub-DAG; the test now lives in workflow-core (so the code is covered in its own module). Two I kept, with reasoning inline: deriving the JSON mapper from JSONUtils.objectMapper rather than reusing it directly (determinism needs ORDER_MAP_ENTRIES_BY_KEYS), and hashing opExecInitInfo rather than the Desc. Retitled to feat(workflow-core) since nothing remains in amber. Re-requesting your review.

[Xiao-zhen-Liu]

@Yicong-Huang on your two questions:

1. How will the cache key be used, and does the test reflect that?
The key identifies the computation behind an output port, so a re-run can decide whether to reuse the port's saved result. The cache service (#5885) stores a row keyed by the cache key when a port's result is materialized, and on a re-run computes the current key, looks it up, and on a hit confirms with isSameComputation (full-JSON compare) before reusing the stored result instead of recomputing. The unit tests reflect exactly that matching behavior: the same sub-DAG produces the same key (a re-run reuses), any upstream change produces a different key (it recomputes, never reusing a stale result), only upstream operators affect the key, and isSameComputation rejects a fabricated hash collision via the full JSON. The DB store/lookup wiring itself is exercised in #5885, where the consumer lives — this PR is just the pure key function.

2. Why did coverage say 0%?
It was a cross-module artifact: coverage is measured per module (JaCoCo), CacheKeyUtil is in workflow-core, but the test had been in the amber module — so the code ran but the coverage was attributed to amber, leaving the workflow-core patch at 0%. I moved the test into workflow-core (it builds PhysicalPlans directly, no engine helpers), so it's now covered in its own module — patch coverage is 95.8%.

[Yicong-Huang]

LGTM with two changes needed

[carloea2]

Comment 1

Could we add a CacheKeyUtil test with a non-empty OpExecInitInfo, such as OpExecWithClassName(..., descString) or OpExecWithCode(...)?

Comment 2

Should computeCacheKey validate that target is an output port and belongs to the provided sub-DAG, or is that expected to be enforced by callers?

Comment 3

Should fields like OutputPort.mode, blocking, or reuseStorage be included in the cache identity, or should we add a test documenting that they are intentionally ignored?

[Xiao-zhen-Liu]

@carloea2 thanks, addressed all three:

1. Non-empty OpExecInitInfo test — added: two operators differing only in their OpExecWithCode(code, "python") produce different cache keys, which confirms exec info is part of the identity and exercises the non-empty serializeOpExec path.

2. Validating target — kept it a pure internal helper rather than a runtime check. Its contract is that the caller passes an output port and the sub-DAG that produces it (plan.getTransitiveUpstreamSubPlan(target.opId)); I documented that in the scaladoc.

3. OutputPort.mode / blocking / reuseStorage — none are in the cache identity, and I added a test documenting that. mode only controls how stored tuples are presented to the UI (snapshot vs delta vs single-snapshot); every tuple is written to storage identically regardless of mode, and it's fixed by the operator type, so it's already implied by the operator identity in the hash. blocking (scheduling) and reuseStorage (storage) likewise don't change the materialized data.

[carloea2]

Thanks @Xiao-zhen-Liu

One more question.

I don't know if it is possible, but if an operator can access those properties like mode, blocking, etc., and change its computation based on that, shouldn't they be considered? I know this is an extreme case anyways.

[Yicong-Huang]

Thanks @Xiao-zhen-Liu

One more question.

I don't know if it is possible, but if an operator can access those properties like mode, blocking, etc., and change its computation based on that, shouldn't they be considered? I know this is an extreme case anyways.

Good question, but I think mode blocking those are execution configs, and they won't affect result.

However, I wonder if change of these invalidates the cache/creates a new storage cache key.