tdcmeehan commented on code in PR #59111:
URL: https://github.com/apache/spark/pull/59111#discussion_r4223124129
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/ExecuteExternalUDFExecSuite.scala:
##########
@@ -396,6 +434,91 @@ class ExecuteExternalUDFExecSuite extends QueryTest with
SharedSparkSession {
parameters = Map.empty)
}
+ gridTest("external UDF records typed terminal metrics from")(
+ Seq("finish", "cancel")) { terminal =>
+ val reported = ExecutionMetrics.newBuilder()
+ .setBytesIn(1)
+ .setBytesOut(2)
+ .setRowsIn(3)
+ .setRowsOut(4)
+ .setBatchesIn(5)
+ .setBatchesOut(6)
+ .setInitWallNanos(7)
+ .setProcessingWallNanos(8)
+ .setReceiveWallNanos(9)
+ .setSendWallNanos(10)
+ .setWorkWallNanos(11)
+ .setWorkCpuNanos(12)
+ .setFinishWallNanos(13)
+ .setMaxWorkWallNanos(14)
+ .setMaxWorkCpuNanos(15)
+ .build()
+ val execution = testExecution(
+ EchoResponses,
+ rowCount = 5L,
+ terminalMetrics = Some(reported),
+ cancelOnClose = terminal == "cancel")
+ val plan = execution.plan.asInstanceOf[TestExecuteExternalUDFExec]
+
+ terminal match {
+ case "finish" => plan.executeCollect()
+ case "cancel" => plan.executeTake(1)
+ case other => fail(s"unexpected terminal: $other")
+ }
+
+ Seq(
+ "bytesIn", "bytesOut", "rowsIn", "rowsOut", "batchesIn", "batchesOut",
+ "initWallNanos", "processingWallNanos", "receiveWallNanos",
"sendWallNanos",
+ "workWallNanos", "workCpuNanos", "finishWallNanos", "maxWorkWallNanos",
+ "maxWorkCpuNanos"
+ ).zipWithIndex.foreach { case (name, index) =>
+ assert(plan.metrics(name).value === index + 1L)
+ }
+ }
+
+ test("external UDF keeps the max, not the sum, of peak metrics across
sessions in a task") {
+ val metrics = ExternalUDFMetrics.create(spark.sparkContext)
+ def snapshot(workWallNanos: Long, maxWall: Long, maxCpu: Long):
ExecutionMetrics =
+ ExecutionMetrics.newBuilder()
+ .setWorkWallNanos(workWallNanos)
+ .setMaxWorkWallNanos(maxWall)
+ .setMaxWorkCpuNanos(maxCpu)
+ .build()
+
+ ExternalUDFMetrics.update(metrics, snapshot(workWallNanos = 10, maxWall =
10, maxCpu = 4))
+ ExternalUDFMetrics.update(metrics, snapshot(workWallNanos = 20, maxWall =
5, maxCpu = 8))
+
+ assert(metrics("workWallNanos").value === 30L)
+ assert(metrics("maxWorkWallNanos").value === 10L)
+ assert(metrics("maxWorkCpuNanos").value === 8L)
+ }
+
+ test("external UDF preserves missing and measured zero terminal metrics") {
+ val reported = ExecutionMetrics.newBuilder()
+ .setBytesIn(0)
Review Comment:
Good catch -- that branch was untested. Covered in 8107481: the
measured-zero test now also reports `max_work_cpu_nanos = 0` and asserts the
metric is `!isZero` with value 0, while `max_work_wall_nanos` stays unset for
the keep-max absence case. Mutation-checked: dropping the `metric.isZero ||`
disjunct makes it fail with `maxWorkCpuNanos.isZero was true` -- exactly your
verification.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
