feat(amber): add cache-reused operator state and cached-region statistics handling#6729
feat(amber): add cache-reused operator state and cached-region statistics handling#6729Xiao-zhen-Liu wants to merge 2 commits into
Conversation
…tics handling Add a CACHE_REUSED WorkflowAggregatedState and the statistics handling for a cached region, so a later scheduler PR can report an operator whose result is reused from cache. - aggregateStates gains an optional cachedState; when every state is the cached one, the group rolls up to CACHE_REUSED. It defaults to None, so every existing caller is unchanged. - aggregatePortMetrics treats a negative count/size as unknown and keeps the aggregated port metrics unknown; sumNonNegative skips those sentinels when summing operator totals. - Utils maps CACHE_REUSED to its display string, parse case, and DB status code 6. With an empty cache nothing produces the cached state or a negative count, so the engine behaves identically to before.
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 | 366 | 0.223 | 24,873/42,575/42,575 us | 🔴 +19.7% / 🔴 +156.6% |
| ⚪ | bs=100 sw=10 sl=64 | 779 | 0.475 | 123,944/165,464/165,464 us | ⚪ within ±5% / 🔴 +51.7% |
| ⚪ | bs=1000 sw=10 sl=64 | 915 | 0.558 | 1,092,764/1,172,492/1,172,492 us | ⚪ within ±5% / 🔴 +12.0% |
Baseline details
Latest main d444fe0 from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---|---|---|---|---|
| bs=10 sw=10 sl=64 | throughput | 366 tuples/sec | 407 tuples/sec | 754.55 tuples/sec | -10.1% | -51.5% |
| bs=10 sw=10 sl=64 | MB/s | 0.223 MB/s | 0.248 MB/s | 0.461 MB/s | -10.1% | -51.6% |
| bs=10 sw=10 sl=64 | p50 | 24,873 us | 24,226 us | 12,816 us | +2.7% | +94.1% |
| bs=10 sw=10 sl=64 | p95 | 42,575 us | 35,570 us | 16,594 us | +19.7% | +156.6% |
| bs=10 sw=10 sl=64 | p99 | 42,575 us | 35,570 us | 19,806 us | +19.7% | +115.0% |
| bs=100 sw=10 sl=64 | throughput | 779 tuples/sec | 813 tuples/sec | 969.38 tuples/sec | -4.2% | -19.6% |
| bs=100 sw=10 sl=64 | MB/s | 0.475 MB/s | 0.496 MB/s | 0.592 MB/s | -4.2% | -19.7% |
| bs=100 sw=10 sl=64 | p50 | 123,944 us | 119,955 us | 103,584 us | +3.3% | +19.7% |
| bs=100 sw=10 sl=64 | p95 | 165,464 us | 160,898 us | 109,097 us | +2.8% | +51.7% |
| bs=100 sw=10 sl=64 | p99 | 165,464 us | 160,898 us | 117,304 us | +2.8% | +41.1% |
| bs=1000 sw=10 sl=64 | throughput | 915 tuples/sec | 905 tuples/sec | 1,004 tuples/sec | +1.1% | -8.8% |
| bs=1000 sw=10 sl=64 | MB/s | 0.558 MB/s | 0.552 MB/s | 0.613 MB/s | +1.1% | -8.9% |
| bs=1000 sw=10 sl=64 | p50 | 1,092,764 us | 1,108,000 us | 1,002,357 us | -1.4% | +9.0% |
| bs=1000 sw=10 sl=64 | p95 | 1,172,492 us | 1,182,839 us | 1,046,463 us | -0.9% | +12.0% |
| bs=1000 sw=10 sl=64 | p99 | 1,172,492 us | 1,182,839 us | 1,073,661 us | -0.9% | +9.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,547.18,200,128000,366,0.223,24872.80,42575.04,42575.04
1,100,10,64,20,2567.38,2000,1280000,779,0.475,123943.69,165464.40,165464.40
2,1000,10,64,20,21867.93,20000,12800000,915,0.558,1092763.67,1172491.51,1172491.51
| case WorkflowAggregatedState.COMPLETED => 3 | ||
| case WorkflowAggregatedState.FAILED => 4 | ||
| case WorkflowAggregatedState.KILLED => 5 | ||
| case WorkflowAggregatedState.CACHE_REUSED => 6 |
There was a problem hiding this comment.
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 frontendngbd-modal-workflow-executions.component.ts
*/
Please add the CacheReused case there too.
There was a problem hiding this comment.
Good catch, thanks. Mirrored code 6 in the frontend: added CacheReused to the ExecutionState enum and EXECUTION_STATUS_CODE, plus the statusMapping and the getExecutionStatus icon in the execution-history modal (a database icon in cyan #13c2c2), with a spec assertion. Happy to change the icon/color if you would prefer something else.
|
I will have to check in on weekend. |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #6729 +/- ##
============================================
+ Coverage 74.71% 75.51% +0.80%
- Complexity 3445 3521 +76
============================================
Files 1160 1160
Lines 45793 45815 +22
Branches 5069 5072 +3
============================================
+ Hits 34212 34598 +386
+ Misses 9929 9514 -415
- Partials 1652 1703 +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:
|
4 participants
What changes were proposed in this PR?
This adds the
CACHE_REUSEDworkflow-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.CACHE_REUSED = 11to theWorkflowAggregatedStateenum, mapped inUtilsto a display string (CacheReused), a parse case, and DB status code6.ExecutionUtils.aggregateStatestakes an optionalcachedState; when every state in the group equals it, the group rolls up toCACHE_REUSED. It defaults toNone, so every existing caller is unchanged. Only the operator-levelaggregateMetricspasses it.ExecutionUtils.aggregatePortMetricstreats a negative count/size as an unknown value and keeps the aggregated port metrics unknown;ExecutionUtils.sumNonNegativeskips those sentinels whenExecutionStatsServicesums 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_REUSEDor a negative tuple count, socachedStatestaysNone(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.
ExecutionUtilsSpeccovers thecachedStatebranch (all-cached rolls up toCACHE_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-1unknown-port sentinel inaggregatePortMetrics, its propagation throughaggregateMetrics, andsumNonNegative.UtilsSpeccovers the display and parse round-trip, the status code6, and that it is distinct fromCOMPLETED. The two specs pass (63 cases), andscalafmtCheckAllis clean.Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 4.8 (Claude Code)