feat(amber): carry cache-reuse status as a metrics flag - #6729
feat(amber): carry cache-reuse status as a metrics flag#6729Xiao-zhen-Liu wants to merge 2 commits into
Conversation
Automated Reviewer SuggestionsBased on the
|
|
@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. |
|
| config | throughput | MB/s | latency | max Δ latest / 7d | |
|---|---|---|---|---|---|
| 🔴 | bs=10 sw=10 sl=64 | 393 | 0.24 | 25,073/42,305/42,305 us | 🔴 +31.5% / 🔴 +159.4% |
| 🔴 | bs=100 sw=10 sl=64 | 925 | 0.565 | 104,079/185,630/185,630 us | 🔴 +33.1% / 🔴 +71.6% |
| 🟢 | bs=1000 sw=10 sl=64 | 1,105 | 0.674 | 905,803/936,500/936,500 us | 🟢 -5.6% / 🟢 -12.9% |
Baseline details
Latest main 561cd0e from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---|---|---|---|---|
| bs=10 sw=10 sl=64 | throughput | 393 tuples/sec | 466 tuples/sec | 775.33 tuples/sec | -15.7% | -49.3% |
| bs=10 sw=10 sl=64 | MB/s | 0.24 MB/s | 0.284 MB/s | 0.473 MB/s | -15.5% | -49.3% |
| bs=10 sw=10 sl=64 | p50 | 25,073 us | 20,285 us | 12,743 us | +23.6% | +96.8% |
| bs=10 sw=10 sl=64 | p95 | 42,305 us | 32,170 us | 16,310 us | +31.5% | +159.4% |
| bs=10 sw=10 sl=64 | p99 | 42,305 us | 32,170 us | 18,926 us | +31.5% | +123.5% |
| bs=100 sw=10 sl=64 | throughput | 925 tuples/sec | 961 tuples/sec | 1,001 tuples/sec | -3.7% | -7.6% |
| bs=100 sw=10 sl=64 | MB/s | 0.565 MB/s | 0.587 MB/s | 0.611 MB/s | -3.7% | -7.5% |
| bs=100 sw=10 sl=64 | p50 | 104,079 us | 99,542 us | 101,399 us | +4.6% | +2.6% |
| bs=100 sw=10 sl=64 | p95 | 185,630 us | 139,502 us | 108,206 us | +33.1% | +71.6% |
| bs=100 sw=10 sl=64 | p99 | 185,630 us | 139,502 us | 118,195 us | +33.1% | +57.1% |
| bs=1000 sw=10 sl=64 | throughput | 1,105 tuples/sec | 1,104 tuples/sec | 1,026 tuples/sec | +0.1% | +7.7% |
| bs=1000 sw=10 sl=64 | MB/s | 0.674 MB/s | 0.674 MB/s | 0.626 MB/s | 0.0% | +7.6% |
| bs=1000 sw=10 sl=64 | p50 | 905,803 us | 901,323 us | 996,304 us | +0.5% | -9.1% |
| bs=1000 sw=10 sl=64 | p95 | 936,500 us | 991,756 us | 1,042,531 us | -5.6% | -10.2% |
| bs=1000 sw=10 sl=64 | p99 | 936,500 us | 991,756 us | 1,074,934 us | -5.6% | -12.9% |
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,508.74,200,128000,393,0.240,25072.69,42305.28,42305.28
1,100,10,64,20,2161.49,2000,1280000,925,0.565,104078.54,185629.64,185629.64
2,1000,10,64,20,18099.58,20000,12800000,1105,0.674,905803.35,936500.47,936500.47
|
I will have to check in on weekend. |
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #6729 +/- ##
============================================
+ Coverage 86.80% 87.94% +1.14%
- Complexity 4226 4340 +114
============================================
Files 1173 1178 +5
Lines 46865 46921 +56
Branches 5231 5230 -1
============================================
+ Hits 40682 41266 +584
+ Misses 4457 3894 -563
- Partials 1726 1761 +35
*This pull request uses carry forward flags. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
4647cc1 to
8d3ba59
Compare
@Yicong-Huang Can you review this PR? Thanks. |
Yicong-Huang
left a comment
Yicong-Huang
left a comment
There was a problem hiding this comment.
🔴 7 must-fix · 4 advisory · 1 polish — the new state's necessity is never established anywhere in the feature's paper trail, and its fan-out produced most of the rest of these findings, including a frontend build break.
Design & architecture (3)
controlreturns.proto:129— does this enum value need to exist? Region and workflow level already derive COMPLETED for a cached region (must-fix, see inline)Utils.scala:171— status byte 6 has no producer, so the dashboard mirror added here is dead code (must-fix, see inline)execute-workflow.interface.ts:162— theOperatorStatemirror, the side that does produce this state, was not updated (must-fix, see inline)
If the first one lands as "reuse COMPLETED", the next four findings and both simplifications disappear — worth settling the direction before fixing them.
Correctness (3)
execute-workflow.interface.ts:162—CacheReusedis in noExecutionStateInfovariant; TS2322 fails the frontend build on all three platforms (must-fix, see inline)ExecutionStatsService.scala:120—inMap/outMapbypass the non-negative guard, so-1reaches the canvas port label (must-fix, see inline)Closes #5883covers two of that issue's three scope bullets. The third — the zero-worker guard that still triggers the final result update — has no counterpart in the diff, and none ofExecutionResultService.scala:367,ExecutionStatsService.scala:218-219,ExecutionRuntimeService.scala:78handlesCACHE_REUSED. Either add the guard or downgrade the keyword toPart ofand keep the issue open (must-fix)
Simplifications (2)
ExecutionUtils.scala:153—-1is a bare literal insideTupleMetrics.count's legal range; name the constant if the sentinel survives (advisory, see inline)ExecutionUtils.scala:103— eight same-typed positional parameters, with a precondition the compiler cannot enforce (advisory, see inline)
Conventions (3)
- Retitle → "feat(amber): add cache-reused state and unknown stats markers" — the current subject is 82 characters against AGENTS.md's ~60-char guidance.
featand theamberscope are both right (advisory) - Description: "How was this PR tested?" lists only the two Scala specs and
scalafmtCheckAll, but four frontend files changed and the frontend is where CI is red (must-fix) Utils.scala:160tells the next person to updatengbd-modal-workflow-executions.component.ts, which no longer exists; the real mirrors areworkflow-execution-history.component.tsandworkflow-executions-entry.ts(advisory)
Polish: 1 quick touch-up (see inline comments).
Verification trace
Traced state derivation at all three levels to check whether a cached region needs a distinct value. OperatorExecution.getState rolls worker states up through aggregateStates, but RegionExecution.getState (RegionExecution.scala:122-134) is a pure function of port completion returning only COMPLETED or RUNNING, and WorkflowExecution.getState (:161) is regionStates.forall(_ == COMPLETED). So CACHE_REUSED cannot appear above the operator level by construction — which is also why status byte 6 has no writer: ExecutionStateStore.scala:44 is the only setStatus(maptoStatusCode(...)) call site, and its input comes only from WorkflowExecution.getState or StartWorkflowHandler.scala:49,52.
Separately confirmed the empty-cache safety claim branch by branch: with cachedState = None every added cachedState.contains(s) term is false, and hasUnknown is false for non-negative counts, so both new paths are the identity. That part holds on the Scala side.
carloea2
left a comment
carloea2
left a comment
There was a problem hiding this comment.
Please fix the open issues before merge. The frontend does not build, and cache reused operators are not handled in the operator state or port counts.
Yicong-Huang
left a comment
Yicong-Huang
left a comment
There was a problem hiding this comment.
🔴 0 resolved · 12 open · 5 new (5 new = 0 newly introduced · 5 late catches)
No commits since my last review, so all twelve findings there still stand and @carloea2's changes-requested still applies. This round adds five late catches, and two of them correct recommendations I made last time.
Design & architecture (2)
ExecutionUtils.scala:54— nothing can putCACHE_REUSEDinto this input: a cached region has zero workers, so the operator rolls up toUNINITIALIZED(must-fix, see inline)Utils.scala:171— correction: keep this case.maptoStatusCodehas a second, operator-level writer I missed (advisory, see inline)
Correctness (1)
execute-workflow.interface.ts:162— correction: the one-line fix I proposed last round breaksexhaustiveGuardand does not compile (must-fix, see inline)
Conventions (1)
- Description: the diff changes a rendered icon and colour (the cyan
databaseglyph atworkflow-execution-history.component.ts:356) with no before/after screenshot (advisory)
Polish: 1 quick touch-up (see inline comments).
Verification trace
Traced whether anything can produce the two new shapes. aggregateMetrics (ExecutionUtils.scala:46-55) consumes OperatorMetrics built by OperatorExecution.getStats (:92-105), whose state comes from getState (:73-84) — and that call passes no cachedState, while WorkerState (statistics.proto:32-40) has no cached member to pass. With zero workers, the shape #5883 describes for a skipped region, workerStates is empty, so aggregateStates returns UNINITIALIZED from its states.isEmpty branch; getStats also emits no port mappings at all, so nothing can carry -1. Both new branches are unreachable from the case they were built for.
Separately re-walked maptoStatusCode's call graph, which is where my earlier comment went wrong: ExecutionStateStore.scala:44 is workflow-level and cannot reach 6, but ExecutionStatsService.scala:294 is operator-level and can.
Add a reused_from_cache boolean to OperatorMetrics so a later scheduler PR can report an operator whose results were reused from the operator port cache. Per the direction settled in apache#5880, reuse is provenance of a completed operator, not a distinct state: a reused operator reports COMPLETED, and this flag carries the distinction to the frontend for display. - executionruntimestate.proto: bool reused_from_cache on OperatorMetrics. - aggregateMetrics: a logical operator is reused only when every one of its physical operators is. - OperatorAggregatedMetrics and the statistics event carry the flag; the TS OperatorStatistics interface gains the matching optional field. Nothing sets the flag until the producer lands with apache#5884, so with an empty cache the engine behaves identically to before.
8d3ba59 to
44001a4
Compare
Xiao-zhen-Liu
changed the title
Yicong-Huang
left a comment
Yicong-Huang
left a comment
There was a problem hiding this comment.
🟡 12 resolved · 0 open · 4 new (4 new = 4 newly introduced · 0 late catches)
The rework does what #5880 concluded, and all twelve prior findings are verified gone from the tree rather than just claimed fixed — 43 lines where the old revision needed eight synchronized surfaces. Nothing below blocks merge.
Simplifications (1)
ExecutionStatsService.scala:128—computeStatsDiff's identity rebuild at:243-256silently drops the new field (advisory, see inline)
Conventions (2)
- The websocket leg is untested: 0% patch coverage on
:128, andTexeraWebSocketEventSpec's fixture (:115-127) omits the field, so the spec pinning this wire contract misses it (advisory) - No linked issue in the sidebar (0 linked, 7 mentions). Dropping
Closeswas right; a non-closing link, or a checklist entry on #5881, restores the trace (advisory)
Polish: 1 quick touch-up (see inline comments).
Verification trace
Traced whether the new field survives each consumer of OperatorMetrics. On the UI path it does: statsStore.withOperatorInfo(evt.operatorMetrics) (:192) stores the metrics unmodified, so :128 reads a live value. On the persistence path it does not — computeStatsDiff (:243-256) rebuilds from fields 1 and 2 only, invisible today because runtimeStatisticsSchema has no such column.
Confirmed the dormancy claim too: the only writers are the spec helper and the forall rollup, and a primitive Boolean is never dropped under NON_ABSENT, so every payload carries reusedFromCache: false with no frontend reader. The mirror is complete — the proto sits outside frontend-proto-gen.sh's set, making execute-workflow.interface.ts:85 the required hand-mirror — and no CACHE_REUSED token survives in the tree.
Review follow-ups on apache#6729: - computeStatsDiff rebuilt OperatorMetrics from its first two fields, an identity transform that silently reset reused_from_cache on the persistence path. Return the combined map as is. - Set the flag non-default in TexeraWebSocketEventSpec's fixture so the symmetric round trip pins it on the wire. - Reword a test comment that relied on PR-history vocabulary and drop a duplicate assertion.
Yicong-Huang
left a comment
Yicong-Huang
left a comment
There was a problem hiding this comment.
🟡 3 resolved · 1 open · 2 new (2 new = 0 newly introduced · 2 late catches)
Last round's two substantive findings are verified gone from the tree, not just claimed. The two below predate that review and are my own misses. Nothing blocks merge.
Correctness (1)
ExecutionUtils.scala:84— theforallinput is narrower than "every physical operator"; asking whether partial reuse is representable (advisory, see inline)
Simplifications (1)
OperatorStatisticsUpdateEvent.scala:35— the only one of twelve DTO fields with a default argument, and it is never taken (advisory, see inline)
Conventions (1)
- Still open: no linked issue in the sidebar (0 linked, 7 mentions). A non-closing link, or a checklist entry on #5881, restores the trace (advisory)
Verification trace
Re-traced the field's survival now that the rebuild is gone. computeStatsDiff (:240) returns the combined map directly, so storeRuntimeStatistics (:243-270) newly receives metrics retaining field 3 — harmless, since it projects 11 named values into runtimeStatisticsSchema and never enumerates the message. The websocket leg is now pinned: the round trip asserts full case-class equality (TexeraWebSocketEventSpec:262-267) on a fixture with the field non-default, so a NON_ABSENT drop would change the value read back and fail.
On the rollup I traced the input set rather than the call: aggregateMetrics' sole caller (WorkflowExecution.scala:94-110) flattens getAllRegionExecutions and groups by logicalOpId, so the set is the physical operators that currently have a region execution. Also confirmed the mirror is complete — the proto sits outside frontend-proto-gen.sh's set, making execute-workflow.interface.ts:85 the required hand-mirror — the TS ? matches the sibling optionals resetStatus relies on, and no CACHE_REUSED token survives in the tree.
| // A logical operator is reused from cache only when every one of its | ||
| // physical operators is. `metrics` is non-empty here, so this cannot | ||
| // hold vacuously. | ||
| reusedFromCache = metrics.forall(_.reusedFromCache) |
There was a problem hiding this comment.
The comment promises more than the input set delivers. The only caller (WorkflowExecution.scala:94-110) flattens getAllRegionExecutions and groups by logicalOpId, so metrics holds the physical operators that currently have a region execution. For a HashJoin, the probe operator contributes nothing until its region is created, so forall can return true with a later operator still to come. SyncExecutionResource.scala:189-200 documents that same window for the sibling state field, so this is the function's existing shape rather than something you introduced.
Can a logical operator ever be partially reused under #5884? If reuse is all-or-nothing, only the comment needs narrowing to "the physical operators reporting so far". If it can be partial, the flag wants the same late-region guard the state field got.
| aggregatedIdleTime: Long | ||
| aggregatedIdleTime: Long, | ||
| // Provenance: the operator completed by reusing cached results (no workers ran). | ||
| reusedFromCache: Boolean = false |
There was a problem hiding this comment.
This is the only one of the twelve fields with a default, and neither construction site takes it — ExecutionStatsService.scala:128 and TexeraWebSocketEventSpec.scala:129 both pass it explicitly. Its only live effect is on the next construction site, where it turns a forgotten flag into a silent false rather than a compile error. That is the shape of the computeStatsDiff rebuild we just deleted, so it seems worth closing off while it is free.
| reusedFromCache: Boolean = false | |
| reusedFromCache: Boolean |
5 participants
What changes were proposed in this PR?
This PR was reworked after the discussion in #5880. The old version added a new CACHE_REUSED state; the review threads below refer to that version. The discussion concluded that a reused operator should just report COMPLETED, because every place that checks state treats completed and reused the same way. What still needs to travel is one bit: whether the operator's results came from the cache.
So the PR now adds only that bit:
reused_from_cacheboolean onOperatorMetrics. A reused operator still reports COMPLETED.aggregateMetrics).OperatorStatisticstype carry the flag to the frontend. No UI changes here; that is Add cache panel and canvas display to the workflow editor #5886.Nothing sets the flag yet. The producer comes with #5884, so with an empty cache the engine behaves exactly like main.
Any related issues, documentation, discussions?
Part of #5881. Design discussion: #5880. Related: #5883 and #5884.
How was this PR tested?
New unit tests in ExecutionUtilsSpec cover the all-physical-operators rule and the flag staying false when nothing sets it. Existing specs pass unchanged, scalafmt is clean, and the frontend production build passes.
Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude (Claude Code)