feat(amber): carry cache-reuse status as a metrics flag

amber/src/main/protobuf/org/apache/texera/amber/engine/common/executionruntimestate.proto

@@ -82,6 +82,10 @@ message OperatorStatistics{
 message OperatorMetrics{
   architecture.rpc.WorkflowAggregatedState operator_state = 1 [(scalapb.field).no_box = true];
   OperatorStatistics operator_statistics = 2 [(scalapb.field).no_box = true];
+  // True when the operator's results were reused from the operator port cache
+  // instead of being computed by workers. Provenance of a completed operator,
+  // not a distinct state; the operator still reports COMPLETED.
+  bool reused_from_cache = 3;
 }
 
 message ExecutionStatsStore {

amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/ExecutionUtils.scala

@@ -77,7 +77,14 @@ object ExecutionUtils {
         dataProcessingTimeSum,
         controlProcessingTimeSum,
         idleTimeSum
-      )
+      ),
+      // Fully-reused semantics: partial reuse is possible (HashJoin's build and
+      // probe sit in different regions), and a partially reused operator reports
+      // false; per-port detail comes from the cache entries. The input holds the
+      // operators that currently have a region execution, so this shares the
+      // state field's transient window until all regions exist. Non-empty here,
+      // so the forall cannot hold vacuously.
+      reusedFromCache = metrics.forall(_.reusedFromCache)
     )
   }
 

amber/src/main/scala/org/apache/texera/web/model/websocket/event/OperatorStatisticsUpdateEvent.scala

@@ -30,7 +30,11 @@ case class OperatorAggregatedMetrics(
     numWorkers: Long,
     aggregatedDataProcessingTime: Long,
     aggregatedControlProcessingTime: Long,
-    aggregatedIdleTime: Long
+    aggregatedIdleTime: Long,
+    // Provenance: the operator completed by reusing cached results (no workers ran).
+    // Deliberately no default: a new construction site must decide the flag
+    // explicitly instead of silently sending false.
+    reusedFromCache: Boolean
 )
 
 case class OperatorStatisticsUpdateEvent(operatorStatistics: Map[String, OperatorAggregatedMetrics])

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

@@ -124,7 +124,8 @@ class ExecutionStatsService(
                 metrics.operatorStatistics.numWorkers,
                 metrics.operatorStatistics.dataProcessingTime,
                 metrics.operatorStatistics.controlProcessingTime,
-                metrics.operatorStatistics.idleTime
+                metrics.operatorStatistics.idleTime,
+                reusedFromCache = metrics.reusedFromCache
               )
               (x._1, res)
           })
@@ -236,23 +237,7 @@ class ExecutionStatsService(
     val updatedLastMetrics = lastPersistedMetrics ++ newKeys.map(_ -> defaultMetrics)
 
     // Combine new metrics with old metrics for keys that are no longer present
-    val completeMetricsMap = newMetrics ++ oldKeys.map(key => key -> updatedLastMetrics(key))
-
-    // Transform the complete metrics map to ensure consistent structure
-    completeMetricsMap.map {
-      case (key, metrics) =>
-        key -> OperatorMetrics(
-          metrics.operatorState,
-          OperatorStatistics(
-            metrics.operatorStatistics.inputMetrics,
-            metrics.operatorStatistics.outputMetrics,
-            metrics.operatorStatistics.numWorkers,
-            metrics.operatorStatistics.dataProcessingTime,
-            metrics.operatorStatistics.controlProcessingTime,
-            metrics.operatorStatistics.idleTime
-          )
-        )
-    }
+    newMetrics ++ oldKeys.map(key => key -> updatedLastMetrics(key))
   }
 
   private def storeRuntimeStatistics(

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

@@ -190,11 +190,13 @@ class ExecutionUtilsSpec extends AnyFlatSpec {
       numWorkers: Int = 0,
       dataTime: Long = 0,
       controlTime: Long = 0,
-      idleTime: Long = 0
+      idleTime: Long = 0,
+      reused: Boolean = false
   ): OperatorMetrics =
     OperatorMetrics(
       state,
-      OperatorStatistics(input, output, numWorkers, dataTime, controlTime, idleTime)
+      OperatorStatistics(input, output, numWorkers, dataTime, controlTime, idleTime),
+      reusedFromCache = reused
     )
 
   "ExecutionUtils.aggregateMetrics" should "return UNINITIALIZED defaults when given no metrics" in {
@@ -337,4 +339,22 @@ class ExecutionUtilsSpec extends AnyFlatSpec {
     assert(result.operatorStatistics.numWorkers == 3)
     assert(result.operatorStatistics.dataProcessingTime == 12)
   }
+
+  // -- aggregateMetrics: reused-from-cache provenance ----------------------
+
+  it should "report reusedFromCache only when every physical operator is reused" in {
+    val reusedA = metricsWith(WorkflowAggregatedState.COMPLETED, reused = true)
+    val reusedB = metricsWith(WorkflowAggregatedState.COMPLETED, reused = true)
+    val computed = metricsWith(WorkflowAggregatedState.COMPLETED)
+
+    assert(ExecutionUtils.aggregateMetrics(List(reusedA, reusedB)).reusedFromCache)
+    assert(!ExecutionUtils.aggregateMetrics(List(reusedA, computed)).reusedFromCache)
+    assert(!ExecutionUtils.aggregateMetrics(List(computed)).reusedFromCache)
+  }
+
+  it should "default reusedFromCache to false for empty input" in {
+    // Empty input takes the early-return path, whose default is false; metrics
+    // that no producer has marked keep that default too.
+    assert(!ExecutionUtils.aggregateMetrics(Iterable.empty).reusedFromCache)
+  }
 }

amber/src/test/scala/org/apache/texera/web/model/websocket/event/TexeraWebSocketEventSpec.scala

@@ -123,7 +123,10 @@ class TexeraWebSocketEventSpec extends AnyFlatSpec with Matchers {
     numWorkers = 17L,
     aggregatedDataProcessingTime = 18L,
     aggregatedControlProcessingTime = 19L,
-    aggregatedIdleTime = 20L
+    aggregatedIdleTime = 20L,
+    // Non-default on purpose: the symmetric round trip below only pins this
+    // field on the wire if a drop would change the value read back.
+    reusedFromCache = true
   )
 
   private val resultRow = objectMapper.createObjectNode().put("city", "Irvine")

frontend/src/app/workspace/types/execute-workflow.interface.ts

@@ -81,6 +81,8 @@ export enum OperatorState {
 export interface OperatorStatistics
   extends Readonly<{
     operatorState: OperatorState;
+    // Provenance: the operator completed by reusing cached results (no workers ran).
+    reusedFromCache?: boolean;
     aggregatedInputRowCount: number;
     aggregatedInputSize?: number;
     inputPortMetrics: Record<string, number>;