This is an automated email from the ASF dual-hosted git repository.
gaogaotiantian pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 623dfc5914a1 [SPARK-59777][PYTHON] Fix timing metrics for reused
Python workers
623dfc5914a1 is described below
commit 623dfc5914a1684bc636ab0846686696aab33db1
Author: Spenser Sun <[email protected]>
AuthorDate: Wed Sep 30 11:35:11 2026 -0700
[SPARK-59777][PYTHON] Fix timing metrics for reused Python workers
### What changes were proposed in this pull request?
Use the later of the current task start and Python worker boot timestamps
as the initialization accounting boundary. Add regression coverage for an idle
reused worker's boot, initialization, and total SQL metrics.
### Why are the changes needed?
A reused worker records its boot timestamp before waiting for the next
task, so its idle time can be included in initialization even though total
worker time starts with the current task.
### Does this PR introduce _any_ user-facing change?
Yes. Python worker timing metrics exclude pre-task idle time for reused
workers, while fresh-worker calculations remain unchanged.
### How was this patch tested?
`PythonUDFSuite` passed all 17 tests, and the new regression test fails
with the old calculation when initialization exceeds total worker time;
Scalastyle and the pre-push Python checks also passed.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-6)
Closes #59025 from Spenserrrr/udf-timing-reuse-fix.
Authored-by: Spenser Sun <[email protected]>
Signed-off-by: Tian Gao <[email protected]>
(cherry picked from commit 67ba9c135b2aed690dadb86f5c6799eea0a243e7)
Signed-off-by: Tian Gao <[email protected]>
---
.../org/apache/spark/api/python/PythonRunner.scala | 7 +++--
.../sql/execution/python/PythonUDFSuite.scala | 30 ++++++++++++++++++++++
2 files changed, 35 insertions(+), 2 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala
b/core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala
index e643045134f4..4b9f5b3d7f1b 100644
--- a/core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala
+++ b/core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala
@@ -863,8 +863,11 @@ private[spark] abstract class BasePythonRunner[IN, OUT](
val initTime = stream.readLong()
val finishTime = stream.readLong()
val processingTimeMs = stream.readLong()
- val boot = bootTime - startTime
- val init = initTime - bootTime
+ // A reused Python worker records bootTime before waiting for this task,
so it can precede
+ // startTime. Use the later timestamp to exclude the worker's idle time
from initialization.
+ val pythonWorkerInitializationStartTime = math.max(startTime, bootTime)
+ val boot = pythonWorkerInitializationStartTime - startTime
+ val init = initTime - pythonWorkerInitializationStartTime
val finish = finishTime - initTime
val total = finishTime - startTime
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/python/PythonUDFSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/python/PythonUDFSuite.scala
index 88e1e7a9122b..96fb33055057 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/python/PythonUDFSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/python/PythonUDFSuite.scala
@@ -17,6 +17,8 @@
package org.apache.spark.sql.execution.python
+import scala.concurrent.duration._
+
import org.apache.spark.sql.{AnalysisException, IntegratedUDFTestUtils, Row}
import org.apache.spark.sql.functions.{aggregate, array, avg, col, count, lit,
transform}
import org.apache.spark.sql.internal.SQLConf
@@ -220,6 +222,34 @@ class PythonUDFSuite extends SharedSparkSession {
s"pythonTotalTime should be > 0, but was $pythonTotalTime")
}
+ test("reused Python worker initialization excludes pre-task idle time") {
+ assume(shouldTestPythonUDFs)
+
+ def executePythonUDF(): BatchEvalPythonExec = {
+ val result = spark.range(1).coalesce(1).select(pythonTestUDF(col("id")))
+ result.collect()
+ result.queryExecution.executedPlan.collectFirst {
+ case exec: BatchEvalPythonExec => exec
+ }.getOrElse {
+ fail("Expected BatchEvalPythonExec in executed plan")
+ }
+ }
+
+ executePythonUDF()
+
+ eventually(timeout(10.seconds), interval(100.milliseconds)) {
+ Thread.sleep(1000L)
+ val pythonExec = executePythonUDF()
+ val bootTime = pythonExec.metrics("pythonBootTime").value
+ val initTime = pythonExec.metrics("pythonInitTime").value
+ val totalTime = pythonExec.metrics("pythonTotalTime").value
+
+ assert(bootTime === 0L, s"expected a reused Python worker, but boot time
was $bootTime ms")
+ assert(initTime <= totalTime,
+ s"initialization time $initTime ms exceeded total worker time
$totalTime ms")
+ }
+ }
+
test("SPARK-55046:pythonProcessingTime metric for ArrowEvalPythonExec") {
assume(shouldTestPythonUDFs)
withSQLConf(SQLConf.ARROW_PYSPARK_EXECUTION_ENABLED.key -> "true") {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
