Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -77,7 +77,11 @@ object ExecutionUtils {
dataProcessingTimeSum,
controlProcessingTimeSum,
idleTimeSum
)
),
// A logical operator is reused from cache only when every one of its
// physical operators is. `metrics` is non-empty here, so this cannot
// hold vacuously.
reusedFromCache = metrics.forall(_.reusedFromCache)

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The comment promises more than the input set delivers. The only caller (WorkflowExecution.scala:94-110) flattens getAllRegionExecutions and groups by logicalOpId, so metrics holds the physical operators that currently have a region execution. For a HashJoin, the probe operator contributes nothing until its region is created, so forall can return true with a later operator still to come. SyncExecutionResource.scala:189-200 documents that same window for the sibling state field, so this is the function's existing shape rather than something you introduced.

Can a logical operator ever be partially reused under #5884? If reuse is all-or-nothing, only the comment needs narrowing to "the physical operators reporting so far". If it can be partial, the flag wants the same late-region guard the state field got.

)
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,9 @@ case class OperatorAggregatedMetrics(
numWorkers: Long,
aggregatedDataProcessingTime: Long,
aggregatedControlProcessingTime: Long,
aggregatedIdleTime: Long
aggregatedIdleTime: Long,
// Provenance: the operator completed by reusing cached results (no workers ran).
reusedFromCache: Boolean = false

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is the only one of the twelve fields with a default, and neither construction site takes it — ExecutionStatsService.scala:128 and TexeraWebSocketEventSpec.scala:129 both pass it explicitly. Its only live effect is on the next construction site, where it turns a forgotten flag into a silent false rather than a compile error. That is the shape of the computeStatsDiff rebuild we just deleted, so it seems worth closing off while it is free.

Suggested change
reusedFromCache: Boolean = false
reusedFromCache: Boolean

)

case class OperatorStatisticsUpdateEvent(operatorStatistics: Map[String, OperatorAggregatedMetrics])
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -124,7 +124,8 @@ class ExecutionStatsService(
metrics.operatorStatistics.numWorkers,
metrics.operatorStatistics.dataProcessingTime,
metrics.operatorStatistics.controlProcessingTime,
metrics.operatorStatistics.idleTime
metrics.operatorStatistics.idleTime,
reusedFromCache = metrics.reusedFromCache
Comment thread
Xiao-zhen-Liu marked this conversation as resolved.
)
(x._1, res)
})
Expand Down Expand Up @@ -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(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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)
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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>;
Expand Down
Loading