Spenserrrr opened a new pull request, #59246: URL: https://github.com/apache/spark/pull/59246
### What changes were proposed in this pull request? - Add a task-local `WorkerMetrics` collector and reusable context-manager timers in `worker.py`. Elapsed intervals accumulate in nanoseconds and are converted to milliseconds when the worker report is written. - Instrument ordinary scalar pandas UDFs in the pandas eval handler, measuring Arrow-to-pandas input conversion, UDF callback execution, and pandas-to-Arrow output conversion separately. - Add `pythonNumTimingReports` and `pythonNumTimedBatches` to make timing coverage visible, including supported tasks with empty input. - Decode the worker's numeric JSON report into a generic `Map[String, Long]`. Keep lifecycle calculations, registered SQL metric updates, and spill accounting in `handleMetricsData`, so additional reported metric names do not require decoder or update-loop changes. - Register the three phase timings and two coverage counters on `ArrowEvalPythonExec` for scalar pandas UDF execution. ### Why are the changes needed? - The existing overall Python execution duration does not show how much time is spent converting input, executing UDF callbacks, or converting output. Separate phase measurements make that breakdown available without timing input fetching or output-consumer pauses as conversion work. - A generic numeric report and collector make it easier to add measurements while preserving a small Scala consumption path. New SQL metrics still need driver-side registration to define their names, display types, and accumulators. ### Does this PR introduce _any_ user-facing change? - Yes. Ordinary scalar pandas UDF operators expose three additional duration metrics and two coverage counters in the Spark SQL UI. - The measurements record observed elapsed time, including profiler wrapper work when profiling is enabled. Iterator UDF operators do not register these phase metrics. ### How was this patch tested? - Added deterministic timer tests for accumulation, nesting, reusable scopes, and exception propagation, plus pandas handler tests for phase boundaries and supported empty input. - Adapted the existing `BasePythonRunnerSuite` decoder tests for numeric maps, arbitrary field names, integer precision, validation, and framing. - Added Spark checks for reported phase values/task and batch counters across two queries, and for the absence of phase metrics on scalar iterator UDF plans. - Passed 17 focused Python timer/handler tests and 10 `BasePythonRunnerSuite` tests, plus Core/SQL compilation and Scala style checks. - Ran the changed Python test files with pytest. The initial combined run had 119 passes and 10 local setup failures; the worker file then passed all 35 cases in isolation, and the four variant failures plus the two new SQL metric checks passed after adding the rebuilt variant classes to the local classpath. - Ruff lint/format and the configured mypy checks passed. Local JVM checks used rebuilt classes with cached assembly dependencies and a session-only LZ4 1.11.3 override. ### Was this patch authored or co-authored using generative AI tooling? - Generated-by: Codex (GPT-6). -- 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]
