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 3 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 | 435 | 0.266 | 22,023/29,483/29,483 us | 🔴 +10.6% / 🔴 +84.4% |
| 🟢 | bs=100 sw=10 sl=64 | 936 | 0.571 | 107,907/119,798/119,798 us | 🟢 -14.2% / 🔴 +11.7% |
| 🔴 | bs=1000 sw=10 sl=64 | 1,070 | 0.653 | 929,387/1,037,529/1,037,529 us | 🔴 +8.7% / ⚪ within ±5% |
Baseline details
Latest main 561cd0e from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---|---|---|---|---|
| bs=10 sw=10 sl=64 | throughput | 435 tuples/sec | 464 tuples/sec | 779.07 tuples/sec | -6.3% | -44.2% |
| bs=10 sw=10 sl=64 | MB/s | 0.266 MB/s | 0.283 MB/s | 0.476 MB/s | -6.0% | -44.1% |
| bs=10 sw=10 sl=64 | p50 | 22,023 us | 19,905 us | 12,818 us | +10.6% | +71.8% |
| bs=10 sw=10 sl=64 | p95 | 29,483 us | 31,437 us | 15,986 us | -6.2% | +84.4% |
| bs=10 sw=10 sl=64 | p99 | 29,483 us | 31,437 us | 19,339 us | -6.2% | +52.5% |
| bs=100 sw=10 sl=64 | throughput | 936 tuples/sec | 929 tuples/sec | 1,011 tuples/sec | +0.8% | -7.4% |
| bs=100 sw=10 sl=64 | MB/s | 0.571 MB/s | 0.567 MB/s | 0.617 MB/s | +0.7% | -7.5% |
| bs=100 sw=10 sl=64 | p50 | 107,907 us | 103,950 us | 100,965 us | +3.8% | +6.9% |
| bs=100 sw=10 sl=64 | p95 | 119,798 us | 139,621 us | 107,295 us | -14.2% | +11.7% |
| bs=100 sw=10 sl=64 | p99 | 119,798 us | 139,621 us | 115,531 us | -14.2% | +3.7% |
| bs=1000 sw=10 sl=64 | throughput | 1,070 tuples/sec | 1,098 tuples/sec | 1,049 tuples/sec | -2.6% | +2.0% |
| bs=1000 sw=10 sl=64 | MB/s | 0.653 MB/s | 0.67 MB/s | 0.64 MB/s | -2.5% | +2.0% |
| bs=1000 sw=10 sl=64 | p50 | 929,387 us | 912,518 us | 978,248 us | +1.8% | -5.0% |
| bs=1000 sw=10 sl=64 | p95 | 1,037,529 us | 954,624 us | 1,021,881 us | +8.7% | +1.5% |
| bs=1000 sw=10 sl=64 | p99 | 1,037,529 us | 954,624 us | 1,050,075 us | +8.7% | -1.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,459.38,200,128000,435,0.266,22023.19,29482.65,29482.65
1,100,10,64,20,2137.15,2000,1280000,936,0.571,107907.12,119798.31,119798.31
2,1000,10,64,20,18692.97,20000,12800000,1070,0.653,929386.78,1037528.51,1037528.51
|
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% 90.12% +3.31%
- Complexity 4226 4429 +203
============================================
Files 1173 1173
Lines 46865 47131 +266
Branches 5231 5286 +55
============================================
+ Hits 40682 42477 +1795
+ Misses 4457 2979 -1478
+ Partials 1726 1675 -51
*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.
Review follow-ups on apache#6729: - OperatorAggregatedMetrics.reusedFromCache loses its default so a new construction site must decide the flag explicitly instead of silently sending false, the same shape as the deleted computeStatsDiff rebuild. - The aggregateMetrics comment claimed "every physical operator"; the input holds the operators that currently have a region execution, so it now says so and names the shared transient window.
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)