Yicong-Huang commented on code in PR #59111:
URL: https://github.com/apache/spark/pull/59111#discussion_r4222514831
##########
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:
**Nit (P3):** This test checks measured-zero only for `bytesIn`, which goes
through `addIfPresent`. The two peak metrics go through `maxIfPresent`, and
there the `metric.isZero` branch is the only path that records a reported 0.
The nano-timing metric starts at -1 and `value` reads 0 while unset, so `0 >
metric.value` alone is false. No test reports `max_work_wall_nanos` or
`max_work_cpu_nanos` as 0. If that branch were dropped, a measured-zero peak
would show as unreported and every assertion here would still pass. Could this
snapshot also set `.setMaxWorkCpuNanos(0)` (and/or the wall peak), with an
assertion that the metric is `!isZero` and has value 0?
**Verification:**
- **Regression:** A snapshot whose max_work_* field is set to 0 leaves the
peak metric non-isZero with value 0, and the assertion fails if the keep-max
logic records a value only when it exceeds the current value.
--
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]
