(texera) branch gh-readonly-queue/main/pr-6729-fb9f4e26d8680ebeb3da28881b27707645d59acd created (now 17169154e4)

[gi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-6729-fb9f4e26d8680ebeb3da28881b27707645d59acd
in repository https://gitbox.apache.org/repos/asf/texera.git


      at 17169154e4 feat(amber): carry cache-reuse status as a metrics flag (#6729)

This branch includes the following new commits:

     new 17169154e4 feat(amber): carry cache-reuse status as a metrics flag (#6729)

The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails.  The revisions
listed as "add" were already present in the repository and have only
been added to this reference.

(texera) 01/01: feat(amber): carry cache-reuse status as a metrics flag (#6729)

[gi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6729-fb9f4e26d8680ebeb3da28881b27707645d59acd
in repository https://gitbox.apache.org/repos/asf/texera.git

commit 17169154e40b97509788748c4e6f0a6693f17a70
Author: Xiaozhen Liu <xi...@uci.edu>
AuthorDate: Tue Aug 18 08:06:42 2026 +0000

    feat(amber): carry cache-reuse status as a metrics flag (#6729)
    
    ### What changes were proposed in this PR?
    
    This PR was reworked after the discussion in #5880. The old version
    added a new CACHE_REUSED state; the review threads below refer to that
    version. The discussion concluded that a reused operator should just
    report COMPLETED, because every place that checks state treats completed
    and reused the same way. What still needs to travel is one bit: whether
    the operator's results came from the cache.
    
    So the PR now adds only that bit:
    
    - A `reused_from_cache` boolean on `OperatorMetrics`. A reused operator
    still reports COMPLETED.
    - A logical operator counts as reused only when all of its physical
    operators are (`aggregateMetrics`).
    - The statistics websocket event and the TS `OperatorStatistics` type
    carry the flag to the frontend. No UI changes here; that is #5886.
    
    Nothing sets the flag yet. The producer comes with #5884, so with an
    empty cache the engine behaves exactly like main.
    
    ### Any related issues, documentation, discussions?
    
    Part of #5881. Design discussion: #5880. Related: #5883 and #5884.
    
    ### How was this PR tested?
    
    New unit tests in ExecutionUtilsSpec cover the all-physical-operators
    rule and the flag staying false when nothing sets it. Existing specs
    pass unchanged, scalafmt is clean, and the frontend production build
    passes.
    
    ### Was this PR authored or co-authored using generative AI tooling?
    
    Generated-by: Claude (Claude Code)
---
 .../engine/common/executionruntimestate.proto      |  4 ++++
 .../coordinator/execution/ExecutionUtils.scala     |  9 +++++++-
 .../event/OperatorStatisticsUpdateEvent.scala      |  6 +++++-
 .../texera/web/service/ExecutionStatsService.scala | 21 +++----------------
 .../coordinator/execution/ExecutionUtilsSpec.scala | 24 ++++++++++++++++++++--
 .../websocket/event/TexeraWebSocketEventSpec.scala |  5 ++++-
 .../workspace/types/execute-workflow.interface.ts  |  2 ++
 7 files changed, 48 insertions(+), 23 deletions(-)

diff --git a/amber/src/main/protobuf/org/apache/texera/amber/engine/common/executionruntimestate.proto b/amber/src/main/protobuf/org/apache/texera/amber/engine/common/executionruntimestate.proto
index e712b3adc8..e384e80c80 100644
--- a/amber/src/main/protobuf/org/apache/texera/amber/engine/common/executionruntimestate.proto
+++ b/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 {
diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/ExecutionUtils.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/ExecutionUtils.scala
index 666aeece42..303fba9a1f 100644
--- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/ExecutionUtils.scala
+++ b/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)
     )
   }
 
diff --git a/amber/src/main/scala/org/apache/texera/web/model/websocket/event/OperatorStatisticsUpdateEvent.scala b/amber/src/main/scala/org/apache/texera/web/model/websocket/event/OperatorStatisticsUpdateEvent.scala
index d4aa6117c9..af91f65784 100644
--- a/amber/src/main/scala/org/apache/texera/web/model/websocket/event/OperatorStatisticsUpdateEvent.scala
+++ b/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])
diff --git a/amber/src/main/scala/org/apache/texera/web/service/ExecutionStatsService.scala b/amber/src/main/scala/org/apache/texera/web/service/ExecutionStatsService.scala
index ac33478451..2c173c25a8 100644
--- a/amber/src/main/scala/org/apache/texera/web/service/ExecutionStatsService.scala
+++ b/amber/src/main/scala/org/apache/texera/web/service/ExecutionStatsService.scala
@@ -125,7 +125,8 @@ class ExecutionStatsService(
                 metrics.operatorStatistics.numWorkers,
                 metrics.operatorStatistics.dataProcessingTime,
                 metrics.operatorStatistics.controlProcessingTime,
-                metrics.operatorStatistics.idleTime
+                metrics.operatorStatistics.idleTime,
+                reusedFromCache = metrics.reusedFromCache
               )
               (x._1, res)
           })
@@ -237,23 +238,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(
diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/ExecutionUtilsSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/ExecutionUtilsSpec.scala
index 237936fa06..3bdb6c2e86 100644
--- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/ExecutionUtilsSpec.scala
+++ b/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)
+  }
 }
diff --git a/amber/src/test/scala/org/apache/texera/web/model/websocket/event/TexeraWebSocketEventSpec.scala b/amber/src/test/scala/org/apache/texera/web/model/websocket/event/TexeraWebSocketEventSpec.scala
index 6a1e063d59..eeeb7f89a9 100644
--- a/amber/src/test/scala/org/apache/texera/web/model/websocket/event/TexeraWebSocketEventSpec.scala
+++ b/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")
diff --git a/frontend/src/app/workspace/types/execute-workflow.interface.ts b/frontend/src/app/workspace/types/execute-workflow.interface.ts
index 8bb7696edf..54d1a2d6ff 100644
--- a/frontend/src/app/workspace/types/execute-workflow.interface.ts
+++ b/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>;