This is an automated email from the ASF dual-hosted git repository.
gaogaotiantian pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 9eb44b3ee9fe [SPARK-57355][PYTHON] Fix __module__ check in udf profiler
9eb44b3ee9fe is described below
commit 9eb44b3ee9fe57c76cd6310270193aa0922d0944
Author: Tian Gao <[email protected]>
AuthorDate: Wed Jun 10 13:48:18 2026 -0700
[SPARK-57355][PYTHON] Fix __module__ check in udf profiler
### What changes were proposed in this pull request?
For `getattr(chained_func, "__module__",
"").startswith("pyspark.sql.worker.")`, if `changed_func.__module__` is `None`,
it will raise an exception. So we check the attribute.
### Why are the changes needed?
In some rare cases, `chained_func.__module__` can be `None` - for example,
if the function is created with `compile` or `exec`. We need to deal with that.
### Does this PR introduce _any_ user-facing change?
Yes. This fixed a bug when users use profilers against UDFs without
`__module__`.
### How was this patch tested?
A regression test is added.
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #56416 from gaogaotiantian/fix-datasource-profiler.
Authored-by: Tian Gao <[email protected]>
Signed-off-by: Tian Gao <[email protected]>
---
python/pyspark/sql/tests/test_udf_profiler.py | 13 +++++++++++++
python/pyspark/worker.py | 3 ++-
2 files changed, 15 insertions(+), 1 deletion(-)
diff --git a/python/pyspark/sql/tests/test_udf_profiler.py
b/python/pyspark/sql/tests/test_udf_profiler.py
index 2900860d43d2..feda3958224c 100644
--- a/python/pyspark/sql/tests/test_udf_profiler.py
+++ b/python/pyspark/sql/tests/test_udf_profiler.py
@@ -300,6 +300,19 @@ class UDFProfiler2TestsMixin:
for id in self.profile_results:
self.assert_udf_profile_present(udf_id=id,
expected_line_count_prefix=10)
+ def test_perf_profiler_udf_without_module(self):
+ @udf("long")
+ def add1(x):
+ return x + 1
+
+ add1.__module__ = None
+
+ with self.sql_conf({"spark.sql.pyspark.udf.profiler": "perf"}):
+ df = self.spark.range(10, numPartitions=2).select(add1("id"))
+ df.collect()
+
+ self.assertEqual(1, len(self.profile_results),
str(self.profile_results.keys()))
+
@unittest.skipIf(
not have_pandas or not have_pyarrow,
pandas_requirement_message or pyarrow_requirement_message,
diff --git a/python/pyspark/worker.py b/python/pyspark/worker.py
index 136e438f5354..47a80173f86e 100644
--- a/python/pyspark/worker.py
+++ b/python/pyspark/worker.py
@@ -937,7 +937,8 @@ def read_single_udf(pickleSer, udf_info, eval_type,
runner_conf, udf_index):
# If chained_func is from pyspark.sql.worker, it is to read/write data
source.
# In this case, we check the data_source_profiler config.
- if getattr(chained_func, "__module__",
"").startswith("pyspark.sql.worker."):
+ module = getattr(chained_func, "__module__", "")
+ if isinstance(module, str) and module.startswith("pyspark.sql.worker."):
profiler = runner_conf.data_source_profiler
else:
profiler = runner_conf.udf_profiler
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
