Xiao-zhen-Liu commented on code in PR #6729:
URL: https://github.com/apache/texera/pull/6729#discussion_r3755724674
##########
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:
Gone with the rework: the enum member no longer exists, so neither does this
build break. The frontend production build passes on the new head.
##########
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:
Confirmed, and it went moot the same way: with the enum member gone there is
no ExecutionStateInfo change and no guard case to add.
##########
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:
Agreed the operator level is where this information belongs. Per #5880 it
now travels as reused_from_cache on OperatorStatistics rather than a state
value, so OperatorState stays untouched. The badge rendering comes with #5886.
##########
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:
Gone with the state (#5880): this case and the three dashboard mirrors. A
reused operator's runtime statistics row now records COMPLETED.
##########
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:
The rework drops the cachedState parameter, so the signature is back to
main's seven. Agreed the next parameter added here should be bundled rather
than positional.
##########
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:
That scaladoc is gone with the rework, since aggregateStates is untouched
now. Point taken on the wording.
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/ExecutionUtilsSpec.scala:
##########
@@ -181,6 +274,61 @@ class ExecutionUtilsSpec extends AnyFlatSpec {
assert(ExecutionUtils.aggregatePortMetrics(List(mapping)) == Seq(mapping))
}
+ it should "mark a port unknown (-1 count and size) when a mapping has a
negative size" in {
+ // A negative in either field marks the whole port unknown, discarding a
+ // valid count. Size-only exercises the right operand of the check.
Review Comment:
That test went with the sentinel in the rework.
##########
amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/ExecutionUtils.scala:
##########
@@ -81,20 +82,30 @@ 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. `cachedState` defaults to None, so an
+ * empty cache leaves this method byte-identical to before.
+ */
def aggregateStates[T](
states: Iterable[T],
completedState: T,
terminatedState: T,
runningState: T,
uninitializedState: T,
pausedState: T,
- readyState: T
+ readyState: T,
+ cachedState: Option[T] = None
): WorkflowAggregatedState = {
states match {
case _ if states.isEmpty =>
WorkflowAggregatedState.UNINITIALIZED
case _ if states.forall(_ == completedState) =>
WorkflowAggregatedState.COMPLETED
case _ if states.forall(_ == terminatedState) =>
WorkflowAggregatedState.COMPLETED
- case _ if states.exists(_ == runningState) =>
WorkflowAggregatedState.RUNNING
+ case _ if cachedState.isDefined && states.forall(_ == cachedState.get) =>
Review Comment:
Following up: per #5880 the state is dropped entirely, so the terminal-state
handling we settled here went with it. Reuse now travels as a reused_from_cache
boolean on OperatorMetrics, and a reused operator reports 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]
