diff --git a/amber/src/main/protobuf/org/apache/texera/amber/engine/architecture/rpc/controlreturns.proto b/amber/src/main/protobuf/org/apache/texera/amber/engine/architecture/rpc/controlreturns.proto index 064a27b5723..a9d3a1e4691 100644 --- a/amber/src/main/protobuf/org/apache/texera/amber/engine/architecture/rpc/controlreturns.proto +++ b/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 { 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 e89da69d433..13bb8893b68 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 @@ -50,7 +50,8 @@ object ExecutionUtils { WorkflowAggregatedState.RUNNING, WorkflowAggregatedState.UNINITIALIZED, WorkflowAggregatedState.PAUSED, - WorkflowAggregatedState.READY + WorkflowAggregatedState.READY, + Some(WorkflowAggregatedState.CACHE_REUSED) ) def sumMetrics( @@ -81,6 +82,13 @@ 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, @@ -88,13 +96,16 @@ object ExecutionUtils { 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 } diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/common/Utils.scala b/amber/src/main/scala/org/apache/texera/amber/engine/common/Utils.scala index 079640317c2..40e60588685 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/common/Utils.scala +++ b/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 } } 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 f112f4f65da..afb1b4e5b06 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 @@ -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, 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 1a66e3fdbbc..10e64051a71 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 @@ -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) + } } diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/common/UtilsSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/common/UtilsSpec.scala index 13674139f17..85666f7658a 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/common/UtilsSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/common/UtilsSpec.scala @@ -40,7 +40,8 @@ class UtilsSpec extends AnyFlatSpec { WorkflowAggregatedState.TERMINATED, WorkflowAggregatedState.FAILED, WorkflowAggregatedState.KILLED, - WorkflowAggregatedState.UNKNOWN + WorkflowAggregatedState.UNKNOWN, + WorkflowAggregatedState.CACHE_REUSED ) namedStates.foreach { state => assert( @@ -55,6 +56,13 @@ class UtilsSpec extends AnyFlatSpec { assert(Utils.aggregatedStateToString(unrecognized) == "Unrecognized(99)") } + it should "render CACHE_REUSED as CacheReused" in { + assert( + Utils.aggregatedStateToString(WorkflowAggregatedState.CACHE_REUSED) == + "CacheReused" + ) + } + // -- stringToAggregatedState ---------------------------------------------- "Utils.stringToAggregatedState" should "be case-insensitive and tolerant of surrounding whitespace" in { @@ -74,6 +82,33 @@ class UtilsSpec extends AnyFlatSpec { } } + it should "parse CacheReused case-insensitively and with surrounding whitespace" in { + assert( + Utils.stringToAggregatedState("CacheReused") == + WorkflowAggregatedState.CACHE_REUSED + ) + assert( + Utils.stringToAggregatedState("cachereused") == + WorkflowAggregatedState.CACHE_REUSED + ) + assert( + Utils.stringToAggregatedState(" CacheReused ") == + WorkflowAggregatedState.CACHE_REUSED + ) + } + + it should "not normalize interior separators when parsing a state name" in { + // Only surrounding whitespace is trimmed; interior spaces are not removed. + assertThrows[IllegalArgumentException] { + Utils.stringToAggregatedState("cache reused") + } + } + + it should "throw for empty or whitespace-only input" in { + assertThrows[IllegalArgumentException] { Utils.stringToAggregatedState("") } + assertThrows[IllegalArgumentException] { Utils.stringToAggregatedState(" ") } + } + // -- maptoStatusCode ------------------------------------------------------ "Utils.maptoStatusCode" should "map known states to their documented byte codes" in { @@ -84,6 +119,14 @@ class UtilsSpec extends AnyFlatSpec { assert(Utils.maptoStatusCode(WorkflowAggregatedState.COMPLETED) == 3.toByte) assert(Utils.maptoStatusCode(WorkflowAggregatedState.FAILED) == 4.toByte) assert(Utils.maptoStatusCode(WorkflowAggregatedState.KILLED) == 5.toByte) + assert(Utils.maptoStatusCode(WorkflowAggregatedState.CACHE_REUSED) == 6.toByte) + } + + it should "store CACHE_REUSED with a code distinct from COMPLETED" in { + assert( + Utils.maptoStatusCode(WorkflowAggregatedState.CACHE_REUSED) != + Utils.maptoStatusCode(WorkflowAggregatedState.COMPLETED) + ) } it should "return -1 for states that have no documented code" in { diff --git a/frontend/src/app/dashboard/component/user/user-workflow/ngbd-modal-workflow-executions/workflow-execution-history.component.spec.ts b/frontend/src/app/dashboard/component/user/user-workflow/ngbd-modal-workflow-executions/workflow-execution-history.component.spec.ts index 76ea30b7d6f..659908d12bb 100644 --- a/frontend/src/app/dashboard/component/user/user-workflow/ngbd-modal-workflow-executions/workflow-execution-history.component.spec.ts +++ b/frontend/src/app/dashboard/component/user/user-workflow/ngbd-modal-workflow-executions/workflow-execution-history.component.spec.ts @@ -266,6 +266,7 @@ describe("WorkflowExecutionHistoryComponent", () => { expect(component.getExecutionStatus(3)).toEqual(["Completed", "check-circle", "green"]); expect(component.getExecutionStatus(4)).toEqual(["Failed", "exclamation-circle", "gray"]); expect(component.getExecutionStatus(5)).toEqual(["Killed", "minus-circle", "red"]); + expect(component.getExecutionStatus(6)).toEqual(["CacheReused", "database", "#13c2c2"]); }); it("falls back to a question-circle for unknown codes", () => { diff --git a/frontend/src/app/dashboard/component/user/user-workflow/ngbd-modal-workflow-executions/workflow-execution-history.component.ts b/frontend/src/app/dashboard/component/user/user-workflow/ngbd-modal-workflow-executions/workflow-execution-history.component.ts index 724d9b4de20..d8c7bfa521b 100644 --- a/frontend/src/app/dashboard/component/user/user-workflow/ngbd-modal-workflow-executions/workflow-execution-history.component.ts +++ b/frontend/src/app/dashboard/component/user/user-workflow/ngbd-modal-workflow-executions/workflow-execution-history.component.ts @@ -189,6 +189,7 @@ export class WorkflowExecutionHistoryComponent implements OnInit, AfterViewInit ["completed", 3], ["failed", 4], ["killed", 5], + ["cachereused", 6], ]); public showORhide: boolean[] = [false, false, false, false, true]; public avatarColors: { [key: string]: string } = {}; @@ -352,6 +353,8 @@ export class WorkflowExecutionHistoryComponent implements OnInit, AfterViewInit return [ExecutionState.Failed.toString(), "exclamation-circle", "gray"]; case 5: return [ExecutionState.Killed.toString(), "minus-circle", "red"]; + case 6: + return [ExecutionState.CacheReused.toString(), "database", "#13c2c2"]; } return ["", "question-circle", "gray"]; } diff --git a/frontend/src/app/dashboard/type/workflow-executions-entry.ts b/frontend/src/app/dashboard/type/workflow-executions-entry.ts index 4f2542ec5f3..22c67c1da9c 100644 --- a/frontend/src/app/dashboard/type/workflow-executions-entry.ts +++ b/frontend/src/app/dashboard/type/workflow-executions-entry.ts @@ -41,4 +41,5 @@ export const EXECUTION_STATUS_CODE: Record = { 3: ExecutionState.Completed, 4: ExecutionState.Failed, 5: ExecutionState.Killed, + 6: ExecutionState.CacheReused, }; diff --git a/frontend/src/app/workspace/types/execute-workflow.interface.ts b/frontend/src/app/workspace/types/execute-workflow.interface.ts index 1fc99e7a605..efd456e8cd1 100644 --- a/frontend/src/app/workspace/types/execute-workflow.interface.ts +++ b/frontend/src/app/workspace/types/execute-workflow.interface.ts @@ -159,6 +159,7 @@ export enum ExecutionState { Terminated = "Terminated", Failed = "Failed", Killed = "Killed", + CacheReused = "CacheReused", } export type ExecutionStateInfo = Readonly<