Yicong-Huang commented on code in PR #6729:
URL: https://github.com/apache/texera/pull/6729#discussion_r3718794245
##########
frontend/src/app/workspace/types/execute-workflow.interface.ts:
##########
@@ -159,6 +159,7 @@ export enum ExecutionState {
Terminated = "Terminated",
Failed = "Failed",
Killed = "Killed",
+ CacheReused = "CacheReused",
Review Comment:
This breaks the frontend build. `ExecutionStateInfo` just below (lines
165-186) enumerates its members explicitly, so `handleExecutionEvent`'s
`default: return { state: newState };` (execute-workflow.service.ts:156) now
assigns a union containing `CacheReused` to a type that excludes it — TS2322,
failing on ubuntu, macos and windows.
`Terminated` is the precedent: it lives in the terminal variant at line 180.
Adding `CacheReused` there compiles, and matches what the state means.
##########
amber/src/main/protobuf/org/apache/texera/amber/engine/architecture/rpc/controlreturns.proto:
##########
@@ -126,6 +126,7 @@ enum WorkflowAggregatedState {
UNKNOWN = 8;
KILLED = 9;
TERMINATED = 10;
+ CACHE_REUSED = 11;
Review Comment:
Before the mirrors and the status code — does this value need to exist at
all?
A cached region already reports COMPLETED at both the region and the
workflow level, by construction. `RegionExecution.getState`
(RegionExecution.scala:122-134) is a pure function of port completion and
returns only COMPLETED or RUNNING. `WorkflowExecution.getState` (:161) is
`regionStates.forall(_ == COMPLETED)`. #5884's plan already says the skip path
"records a completed-from-cache result with no workers" and that "a skipped
region completes at once with no workers" — so marking the ports completed is
enough for both layers. The only level where a distinct value is observable is
the operator badge, which is #5886's scope, and cached-ness there can ride the
`Region.cached` flag #5884 adds anyway.
I went looking for where reusing COMPLETED was weighed and could not find
it. #5880 carries it as a single merge-plan bullet ("a 'completed from cache'
operator state and the matching statistics handling"); the token COMPLETED
appears nowhere in that discussion, nor in #4346. #5881, #5883, #5884, this
description, and the code comments all take the state as given.
The cost is concrete in this diff. Eight surfaces have to stay in sync —
this enum, the three `Utils` maps, the DB status byte, `ExecutionState`,
`ExecutionStateInfo`, `OperatorState`, and `EXECUTION_STATUS_CODE` plus the
modal's icon and search maps — and today two mirrors sit on a path that cannot
receive the value, one is missing on the path that can, and one breaks the
build. `ComputingUnitMaster.scala:189` is a further hazard: it rewrites any
stored status other than COMPLETED to FAILED when it sweeps expired executions
after a restart, so a successful cache reuse would eventually display as Failed.
The same question applies to the statistics half. A cached port's counts
have three candidate answers — zero, an unknown marker, or the result's real
row count — and only the middle one is implemented. The real count looks the
most useful (a re-run shows the same numbers as a real run) and it would delete
the sentinel machinery entirely. Does #5882 record the row count next to the
result location? If it does, I would drop both the new state and the sentinel,
and let a cached region be COMPLETED with its real numbers.
I am not asking you to redo the slice on my word — but this deserves an
answer in #5880 before the mirrors get finished, because the four findings
below all dissolve if the answer is "reuse COMPLETED".
##########
amber/src/main/scala/org/apache/texera/amber/engine/common/Utils.scala:
##########
@@ -166,6 +168,7 @@ object Utils extends LazyLogging {
case WorkflowAggregatedState.COMPLETED => 3
case WorkflowAggregatedState.FAILED => 4
case WorkflowAggregatedState.KILLED => 5
+ case WorkflowAggregatedState.CACHE_REUSED => 6
Review Comment:
Nothing can write this byte. `execution.setStatus(maptoStatusCode(state))`
(ExecutionStateStore.scala:44) is the only writer, and its `state` comes only
from `WorkflowExecution.getState` (WorkflowExecution.scala:156-181) or
`StartWorkflowHandler.scala:49,52`. That `getState` is a hand-rolled parallel
copy of `aggregateStates` which this PR does not extend, so it can only return
UNINITIALIZED, COMPLETED, RUNNING, PAUSED, READY or UNKNOWN.
So the three dashboard mirrors — `EXECUTION_STATUS_CODE[6]`,
`getExecutionStatus(6)`, and the `statusMapping` entry — are unreachable. This
confirms @Ma77Ball's instinct that this code path needed attention, just not in
the direction it went.
`TERMINATED` shows what to do instead: it gets no status code at all,
precisely because it is not a workflow-level state. I would follow that — drop
this case and the three frontend mirrors, and put the mirror on
`OperatorState`. Making `WorkflowExecution.getState` delegate to
`aggregateStates` so the workflow level can report the state too is the better
long-term shape, but it is a bigger change than this slice and it runs into the
restart-sweep problem noted on the proto line.
##########
amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/ExecutionUtils.scala:
##########
@@ -81,26 +82,50 @@ object ExecutionUtils {
)
}
+ /**
+ * Rolls a group of execution states up into one workflow-level state.
+ *
+ * When `cachedState` is provided and every state equals it, the group is
+ * reported as CACHE_REUSED. A cached state is otherwise terminal, like
+ * completed and terminated: it counts toward the all-terminal case and is
+ * dropped before the remaining states are classified. `cachedState`
+ * defaults to None, so an empty cache leaves this method byte-identical
+ * to before. When given, it must differ from the other six states.
+ */
def aggregateStates[T](
states: Iterable[T],
completedState: T,
terminatedState: T,
runningState: T,
uninitializedState: T,
pausedState: T,
- readyState: T
+ readyState: T,
+ cachedState: Option[T] = None
Review Comment:
This signature is now eight parameters of the same type, and the scaladoc
above carries a precondition the compiler cannot check ("it must differ from
the other six states"). A caller that swaps two arguments compiles silently.
Bundling the vocabulary into one case class would turn that into a compile
error. Not blocking, but this is the point where the next parameter should not
be positional.
##########
amber/src/main/scala/org/apache/texera/web/service/ExecutionStatsService.scala:
##########
@@ -115,11 +116,19 @@ class ExecutionStatsService(
val res = OperatorAggregatedMetrics(
Utils.aggregatedStateToString(metrics.operatorState),
-
metrics.operatorStatistics.inputMetrics.map(_.tupleMetrics.count).sum,
-
metrics.operatorStatistics.inputMetrics.map(_.tupleMetrics.size).sum,
+ ExecutionUtils.sumNonNegative(
+
metrics.operatorStatistics.inputMetrics.map(_.tupleMetrics.count)
Review Comment:
The guard stops two lines short of where it matters. `inMap`/`outMap` (lines
110-115) still map `pm.tupleMetrics.count` straight through, so this one event
carries guarded totals next to an unguarded per-port `-1`.
On the frontend, `inputMetrics[numericSuffix] ?? 0`
(joint-ui.service.ts:399, and :409 for outputs) only substitutes for null and
undefined, so `-1` survives to `toLocaleString()` and shows up as a literal
`-1` on the operator's port label. Unreachable today, but guaranteed once #5884
marks operators as reused — and the port label is exactly the surface
"cached-region statistics handling" covers.
Of the two places to fix it, I would teach the TS render to treat a negative
count as unknown rather than filter the port out of the map here: dropping the
entry makes a cached port indistinguishable from an operator that has no such
port.
##########
amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/ExecutionUtils.scala:
##########
@@ -81,26 +82,50 @@ object ExecutionUtils {
)
}
+ /**
+ * Rolls a group of execution states up into one workflow-level state.
+ *
+ * When `cachedState` is provided and every state equals it, the group is
+ * reported as CACHE_REUSED. A cached state is otherwise terminal, like
+ * completed and terminated: it counts toward the all-terminal case and is
+ * dropped before the remaining states are classified. `cachedState`
+ * defaults to None, so an empty cache leaves this method byte-identical
+ * to before. When given, it must differ from the other six states.
Review Comment:
"byte-identical" claims the source is unchanged, but the source did change —
what is preserved is behaviour. Same wording in the PR description.
```suggestion
* defaults to None, so an empty cache leaves this method behaviorally
* identical to before. When given, it must differ from the other six
states.
```
##########
amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/ExecutionUtils.scala:
##########
@@ -121,10 +146,20 @@ object ExecutionUtils {
.view
.map {
case (portId, mappings) =>
- val totalCount = mappings.map(_.tupleMetrics.count).sum
- val totalSize = mappings.map(_.tupleMetrics.size).sum
+ // A negative count/size marks an unknown value (e.g. a cached input
+ // port), so any unknown keeps the aggregated port metrics unknown.
+ val hasUnknown =
+ mappings.exists(m => m.tupleMetrics.count < 0 ||
m.tupleMetrics.size < 0)
+ val totalCount = if (hasUnknown) -1L else
mappings.map(_.tupleMetrics.count).sum
Review Comment:
If the sentinel survives the direction question, give it a name. `-1` is a
bare literal here and on the next line, and it sits inside the legal range of
`TupleMetrics.count` (`int64`) — so the sentinel overlaps values a producer
could legitimately report. A named constant in this object, used by
`sumNonNegative`'s predicate too, makes the contract greppable from the call
sites that have to honour it.
##########
frontend/src/app/workspace/types/execute-workflow.interface.ts:
##########
@@ -159,6 +159,7 @@ export enum ExecutionState {
Terminated = "Terminated",
Failed = "Failed",
Killed = "Killed",
+ CacheReused = "CacheReused",
Review Comment:
Separately: this mirror is on the workflow-level enum, but the state is
operator-level — `aggregateMetrics` (ExecutionUtils.scala:54) is the only
caller passing `Some(CACHE_REUSED)`.
`OperatorState` (lines 69-79) has no `CacheReused`, and
`changeOperatorState` (joint-ui.service.ts:443-459) has no case for it, so a
cache-reused operator renders in the gray unknown colour with the raw text
"CacheReused". `Terminated` is also absent from `OperatorState`, but harmlessly
— `OperatorExecution.getState` can never return it. Here the omission is
load-bearing, so please add the member plus a `case OperatorState.CacheReused`
alongside `Completed`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
