(spark) branch branch-4.2 updated: [SPARK-57355][PYTHON] Fix __module__ check in udf profiler

[ga...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

gaogaotiantian pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/branch-4.2 by this push:
     new 80c7f6b440f2 [SPARK-57355][PYTHON] Fix __module__ check in udf profiler
80c7f6b440f2 is described below

commit 80c7f6b440f29b414bbe6c4df8876500841abddc
Author: Tian Gao <ga...@hotmail.com>
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 <ga...@hotmail.com>
    Signed-off-by: Tian Gao <ga...@hotmail.com>
    (cherry picked from commit 9eb44b3ee9fe57c76cd6310270193aa0922d0944)
    Signed-off-by: Tian Gao <ga...@hotmail.com>
---
 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 32ea54171184..0da0e0ccc35c 100644
--- a/python/pyspark/worker.py
+++ b/python/pyspark/worker.py
@@ -1086,7 +1086,8 @@ def read_single_udf(pickleSer, infile, 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: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org