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 <[email protected]> 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>;
