feat(amber): add cache-reused operator state and cached-region statistics handling

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;
 }
 
 message StartWorkflowResponse {

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)
     )
 
     def sumMetrics(
@@ -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) =>
+        WorkflowAggregatedState.CACHE_REUSED
+      case _ if states.exists(_ == runningState) => WorkflowAggregatedState.RUNNING
       case _ =>
         val unCompletedStates = states.filter(_ != completedState)
         if (unCompletedStates.forall(_ == uninitializedState)) {
@@ -117,10 +128,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
+          val totalSize = if (hasUnknown) -1L else mappings.map(_.tupleMetrics.size).sum
           PortTupleMetricsMapping(portId, TupleMetrics(totalCount, totalSize))
       }
       .toSeq
   }
+
+  /**
+    * Sums metric values while skipping negative sentinels (unknown values, e.g.
+    * from a cached input port), so an unknown does not corrupt operator totals.
+    */
+  def sumNonNegative(values: Iterable[Long]): Long = values.filter(_ >= 0).sum
 }

amber/src/main/scala/org/apache/texera/amber/engine/common/Utils.scala

@@ -103,6 +103,7 @@ object Utils extends LazyLogging {
       case WorkflowAggregatedState.FAILED        => "Failed"
       case WorkflowAggregatedState.KILLED        => "Killed"
       case WorkflowAggregatedState.UNKNOWN       => "Unknown"
+      case WorkflowAggregatedState.CACHE_REUSED  => "CacheReused"
       case WorkflowAggregatedState.Unrecognized(unrecognizedValue) =>
         s"Unrecognized($unrecognizedValue)"
     }
@@ -122,6 +123,7 @@ object Utils extends LazyLogging {
       case "killed"        => WorkflowAggregatedState.KILLED
       case "terminated"    => WorkflowAggregatedState.TERMINATED
       case "unknown"       => WorkflowAggregatedState.UNKNOWN
+      case "cachereused"   => WorkflowAggregatedState.CACHE_REUSED
       case other           => throw new IllegalArgumentException(s"Unrecognized state: $other")
     }
   }
@@ -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
       case other                                 => -1
     }
   }

amber/src/main/scala/org/apache/texera/web/service/ExecutionStatsService.scala

@@ -36,6 +36,7 @@ import org.apache.texera.amber.engine.architecture.coordinator.{
   WorkerAssignmentUpdate,
   WorkflowRecoveryStatus
 }
+import org.apache.texera.amber.engine.architecture.coordinator.execution.ExecutionUtils
 import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState
 import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState.{
   COMPLETED,
@@ -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)
+                ),
+                ExecutionUtils.sumNonNegative(
+                  metrics.operatorStatistics.inputMetrics.map(_.tupleMetrics.size)
+                ),
                 inMap,
-                metrics.operatorStatistics.outputMetrics.map(_.tupleMetrics.count).sum,
-                metrics.operatorStatistics.outputMetrics.map(_.tupleMetrics.size).sum,
+                ExecutionUtils.sumNonNegative(
+                  metrics.operatorStatistics.outputMetrics.map(_.tupleMetrics.count)
+                ),
+                ExecutionUtils.sumNonNegative(
+                  metrics.operatorStatistics.outputMetrics.map(_.tupleMetrics.size)
+                ),
                 outMap,
                 metrics.operatorStatistics.numWorkers,
                 metrics.operatorStatistics.dataProcessingTime,
@@ -266,10 +275,18 @@ class ExecutionStatsService(
             Array(
               operatorId,
               new java.sql.Timestamp(System.currentTimeMillis()),
-              stat.operatorStatistics.inputMetrics.map(_.tupleMetrics.count).sum,
-              stat.operatorStatistics.inputMetrics.map(_.tupleMetrics.size).sum,
-              stat.operatorStatistics.outputMetrics.map(_.tupleMetrics.count).sum,
-              stat.operatorStatistics.outputMetrics.map(_.tupleMetrics.size).sum,
+              ExecutionUtils.sumNonNegative(
+                stat.operatorStatistics.inputMetrics.map(_.tupleMetrics.count)
+              ),
+              ExecutionUtils.sumNonNegative(
+                stat.operatorStatistics.inputMetrics.map(_.tupleMetrics.size)
+              ),
+              ExecutionUtils.sumNonNegative(
+                stat.operatorStatistics.outputMetrics.map(_.tupleMetrics.count)
+              ),
+              ExecutionUtils.sumNonNegative(
+                stat.operatorStatistics.outputMetrics.map(_.tupleMetrics.size)
+              ),
               stat.operatorStatistics.dataProcessingTime,
               stat.operatorStatistics.controlProcessingTime,
               stat.operatorStatistics.idleTime,

amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/ExecutionUtilsSpec.scala

@@ -40,6 +40,7 @@ class ExecutionUtilsSpec extends AnyFlatSpec {
   private val Uninitialized = "uninitialized"
   private val Paused = "paused"
   private val Ready = "ready"
+  private val Cached = "cached"
 
   private def aggregate(states: String*): WorkflowAggregatedState =
     ExecutionUtils.aggregateStates(
@@ -52,6 +53,18 @@ class ExecutionUtilsSpec extends AnyFlatSpec {
       Ready
     )
 
+  private def aggregateWithCache(states: String*): WorkflowAggregatedState =
+    ExecutionUtils.aggregateStates(
+      states,
+      Completed,
+      Terminated,
+      Running,
+      Uninitialized,
+      Paused,
+      Ready,
+      Some(Cached)
+    )
+
   "ExecutionUtils.aggregateStates" should "return UNINITIALIZED for an empty input" in {
     assert(aggregate() == WorkflowAggregatedState.UNINITIALIZED)
   }
@@ -119,6 +132,66 @@ class ExecutionUtilsSpec extends AnyFlatSpec {
     assert(aggregate(Completed, "not-a-real-state") == WorkflowAggregatedState.UNKNOWN)
   }
 
+  // -- aggregateStates: cached-state handling -----------------------------
+
+  it should "return CACHE_REUSED when every state is the cached sentinel" in {
+    assert(aggregateWithCache(Cached) == WorkflowAggregatedState.CACHE_REUSED)
+    assert(aggregateWithCache(Cached, Cached) == WorkflowAggregatedState.CACHE_REUSED)
+  }
+
+  it should "give the completed and terminated branches precedence over the cached branch" in {
+    assert(aggregateWithCache(Completed, Completed) == WorkflowAggregatedState.COMPLETED)
+    assert(aggregateWithCache(Terminated, Terminated) == WorkflowAggregatedState.COMPLETED)
+  }
+
+  it should "give running precedence when a cached state is also present" in {
+    assert(aggregateWithCache(Cached, Running) == WorkflowAggregatedState.RUNNING)
+    assert(aggregateWithCache(Completed, Cached, Running) == WorkflowAggregatedState.RUNNING)
+  }
+
+  it should "return UNINITIALIZED for an empty input even when a cached sentinel is provided" in {
+    assert(aggregateWithCache() == WorkflowAggregatedState.UNINITIALIZED)
+  }
+
+  it should "return UNKNOWN when a cached state is mixed with any non-cached state" in {
+    // The fall-through strips only the completed sentinel, so a lone cached
+    // state alongside anything else is not reclassified. The terminal-state
+    // rework that would fold cached in with completed is a separate change.
+    assert(aggregateWithCache(Completed, Cached) == WorkflowAggregatedState.UNKNOWN)
+    assert(aggregateWithCache(Terminated, Cached) == WorkflowAggregatedState.UNKNOWN)
+    assert(aggregateWithCache(Cached, Uninitialized) == WorkflowAggregatedState.UNKNOWN)
+    assert(aggregateWithCache(Cached, Paused) == WorkflowAggregatedState.UNKNOWN)
+    // Contrast: an all-ready group maps to RUNNING, but cached + ready is UNKNOWN.
+    assert(aggregateWithCache(Cached, Ready) == WorkflowAggregatedState.UNKNOWN)
+  }
+
+  it should "match the no-cache overload exactly when no state is the cached sentinel" in {
+    // Empty-cache safety property: aggregateMetrics always passes
+    // Some(CACHE_REUSED), but with no cached state present the result
+    // must equal the cachedState = None classification.
+    val inputs = Seq(
+      Seq(Completed, Completed),
+      Seq(Terminated, Terminated),
+      Seq(Completed, Running),
+      Seq(Completed, Paused, Paused),
+      Seq(Completed, Uninitialized, Uninitialized),
+      Seq(Completed, Ready, Ready),
+      Seq(Completed, Paused, Ready),
+      Seq(Completed, Terminated)
+    )
+    inputs.foreach { states =>
+      assert(
+        aggregateWithCache(states: _*) == aggregate(states: _*),
+        s"cached overload diverged from the no-cache overload for $states"
+      )
+    }
+  }
+
+  it should "leave the cache branch unreachable when cachedState is None" in {
+    // Without a defined cachedState, every-state-cached is not reclassified.
+    assert(aggregate(Cached, Cached) == WorkflowAggregatedState.UNKNOWN)
+  }
+
   // -- aggregatePortMetrics -----------------------------------------------
 
   "ExecutionUtils.aggregatePortMetrics" should "return empty when given no mappings" in {
@@ -181,6 +254,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.
+    val mapping = PortTupleMetricsMapping(PortIdentity(0), TupleMetrics(50, -7))
+    assert(
+      ExecutionUtils.aggregatePortMetrics(List(mapping)) ==
+        Seq(PortTupleMetricsMapping(PortIdentity(0), TupleMetrics(-1, -1)))
+    )
+  }
+
+  it should "keep a port unknown when a known and an unknown mapping share it" in {
+    val portId = PortIdentity(0)
+    val known = PortTupleMetricsMapping(portId, TupleMetrics(5, 50))
+    val unknown = PortTupleMetricsMapping(portId, TupleMetrics(-1, -1))
+    assert(
+      ExecutionUtils.aggregatePortMetrics(List(known, unknown)) ==
+        Seq(PortTupleMetricsMapping(portId, TupleMetrics(-1, -1)))
+    )
+  }
+
+  it should "isolate the unknown marker per port while still summing a known port" in {
+    val known = PortIdentity(0)
+    val cached = PortIdentity(1)
+    val mappings = List(
+      PortTupleMetricsMapping(known, TupleMetrics(1, 10)),
+      PortTupleMetricsMapping(known, TupleMetrics(2, 20)),
+      PortTupleMetricsMapping(cached, TupleMetrics(-1, -1))
+    )
+    val result = ExecutionUtils.aggregatePortMetrics(mappings).toSet
+    assert(
+      result == Set(
+        PortTupleMetricsMapping(known, TupleMetrics(3, 30)),
+        PortTupleMetricsMapping(cached, TupleMetrics(-1, -1))
+      )
+    )
+  }
+
+  it should "treat the same numeric id with different internal flags as distinct ports" in {
+    // aggregatePortMetrics does not filter internal ports; it groups by the full
+    // PortIdentity, so the internal flag distinguishes the two ports.
+    val publicPort = PortIdentity(0)
+    val internalPort = PortIdentity(0, internal = true)
+    val mappings = List(
+      PortTupleMetricsMapping(publicPort, TupleMetrics(1, 10)),
+      PortTupleMetricsMapping(internalPort, TupleMetrics(2, 20))
+    )
+    val result = ExecutionUtils.aggregatePortMetrics(mappings).toSet
+    assert(
+      result == Set(
+        PortTupleMetricsMapping(publicPort, TupleMetrics(1, 10)),
+        PortTupleMetricsMapping(internalPort, TupleMetrics(2, 20))
+      )
+    )
+  }
+
   // -- aggregateMetrics ---------------------------------------------------
 
   private def metricsWith(
@@ -337,4 +465,83 @@ class ExecutionUtilsSpec extends AnyFlatSpec {
     assert(result.operatorStatistics.numWorkers == 3)
     assert(result.operatorStatistics.dataProcessingTime == 12)
   }
+
+  it should "report CACHE_REUSED when every operator is reused from cache" in {
+    val a = metricsWith(WorkflowAggregatedState.CACHE_REUSED, numWorkers = 1)
+    val b = metricsWith(WorkflowAggregatedState.CACHE_REUSED, numWorkers = 2)
+
+    val result = ExecutionUtils.aggregateMetrics(List(a, b))
+
+    assert(result.operatorState == WorkflowAggregatedState.CACHE_REUSED)
+    assert(result.operatorStatistics.numWorkers == 3)
+  }
+
+  it should "propagate an unknown (-1) cached input port through aggregateMetrics" in {
+    val cachedPort = PortIdentity(0)
+    val metrics = metricsWith(
+      WorkflowAggregatedState.CACHE_REUSED,
+      input = Seq(PortTupleMetricsMapping(cachedPort, TupleMetrics(-1, -1)))
+    )
+
+    val result = ExecutionUtils.aggregateMetrics(List(metrics))
+
+    assert(result.operatorState == WorkflowAggregatedState.CACHE_REUSED)
+    assert(
+      result.operatorStatistics.inputMetrics ==
+        Seq(PortTupleMetricsMapping(cachedPort, TupleMetrics(-1, -1)))
+    )
+  }
+
+  it should "filter an internal cached port so its -1 marker does not leak into the aggregate" in {
+    val publicPort = PortIdentity(0)
+    val internalCachedPort = PortIdentity(1, internal = true)
+    val metrics = metricsWith(
+      WorkflowAggregatedState.CACHE_REUSED,
+      input = Seq(
+        PortTupleMetricsMapping(publicPort, TupleMetrics(3, 30)),
+        PortTupleMetricsMapping(internalCachedPort, TupleMetrics(-1, -1))
+      )
+    )
+
+    val result = ExecutionUtils.aggregateMetrics(List(metrics))
+
+    assert(
+      result.operatorStatistics.inputMetrics ==
+        Seq(PortTupleMetricsMapping(publicPort, TupleMetrics(3, 30)))
+    )
+  }
+
+  it should "sum scalar statistics with a plain sum, not the non-negative guard" in {
+    // sumNonNegative guards only port tuple counts/sizes; scalar fields such as
+    // dataProcessingTime use a plain sum, so a negative contribution is summed.
+    val a = metricsWith(WorkflowAggregatedState.RUNNING, dataTime = 10)
+    val b = metricsWith(WorkflowAggregatedState.RUNNING, dataTime = -4)
+
+    val result = ExecutionUtils.aggregateMetrics(List(a, b))
+
+    assert(result.operatorStatistics.dataProcessingTime == 6)
+  }
+
+  // -- sumNonNegative -----------------------------------------------------
+
+  "ExecutionUtils.sumNonNegative" should "return 0 for an empty input" in {
+    assert(ExecutionUtils.sumNonNegative(Iterable.empty) == 0L)
+  }
+
+  it should "sum every value when they are all non-negative" in {
+    assert(ExecutionUtils.sumNonNegative(List(1L, 2L, 3L)) == 6L)
+    assert(ExecutionUtils.sumNonNegative(List(0L, 5L)) == 5L)
+  }
+
+  it should "skip negative sentinels and sum only the non-negative values" in {
+    assert(ExecutionUtils.sumNonNegative(List(5L, -1L, 3L)) == 8L)
+  }
+
+  it should "return 0 when every value is negative" in {
+    assert(ExecutionUtils.sumNonNegative(List(-1L, -7L)) == 0L)
+  }
+
+  it should "drop Long.MinValue but keep Long.MaxValue" in {
+    assert(ExecutionUtils.sumNonNegative(List(Long.MinValue, Long.MaxValue)) == Long.MaxValue)
+  }
 }