Xiao-zhen-Liu commented on code in PR #6729:
URL: https://github.com/apache/texera/pull/6729#discussion_r3755700707
##########
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:
Answered in #5880: agreed, and the state should not exist at the operator
level either, since three operator-level consumers compare against COMPLETED.
The PR is reworked: the state, its mirrors, and the aggregateStates changes are
all gone. What remains is a reused_from_cache boolean on OperatorMetrics,
carried to the stats event. Thanks for the two corrections, both confirmed.
##########
amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/ExecutionUtils.scala:
##########
@@ -50,7 +50,8 @@ object ExecutionUtils {
WorkflowAggregatedState.RUNNING,
WorkflowAggregatedState.UNINITIALIZED,
WorkflowAggregatedState.PAUSED,
- WorkflowAggregatedState.READY
+ WorkflowAggregatedState.READY,
+ Some(WorkflowAggregatedState.CACHE_REUSED)
Review Comment:
Right. The rework removes those branches. What remains is the proto field,
which #5884's producer (OperatorExecution.cachedMetrics) fills at exactly the
layer you pointed at.
##########
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 sentinel is out of this PR. If an unknown marker comes back with the
producer in #5884, we will fix it on the render side as you suggest, so an
unknown port shows as unknown.
##########
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:
Moved to #5884 with the rest of the unknown-count handling, if the marker
survives at all.
##########
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:
Confirmed the second writer, thanks. With the state gone this case goes too,
and a reused operator's runtime statistics row records COMPLETED. Whether that
row also carries the reused flag is a #5884 decision.
##########
amber/src/main/scala/org/apache/texera/amber/engine/common/Utils.scala:
##########
@@ -141,6 +143,7 @@ object Utils extends LazyLogging {
case WorkflowAggregatedState.COMPLETED => 3
case WorkflowAggregatedState.FAILED => 4
case WorkflowAggregatedState.KILLED => 5
+ case WorkflowAggregatedState.CACHE_REUSED => 6
Review Comment:
Per #5880 the status code and its mirrors are dropped together with the
state. Your comment here is what started that finding, thanks.
--
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]
