refactor(udf): add _from_internal classmethod to UDF wrappers

`SessionContext.udf` / `udaf` / `udwf` previously constructed wrapper
objects by calling `__new__` directly and writing the private `_udf`
/ `_udaf` / `_udwf` attribute from outside the owning module. Three
near-identical blocks coupled `context.py` to wrapper internals.

Add a `_from_internal` classmethod on each wrapper that takes an
already-constructed `df_internal` handle and returns a wrapper
without re-running `__init__`. The lookup methods now collapse to a
single call, the `__new__` bypass is documented on the wrapper class
itself, and renaming the private field is a one-spot edit.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: timsaucer

python/datafusion/context.py

@@ -1354,9 +1354,7 @@ def udf(self, name: str) -> ScalarUDF:
         """
         from datafusion.user_defined import ScalarUDF as _ScalarUDF  # noqa: PLC0415
 
-        wrapper = _ScalarUDF.__new__(_ScalarUDF)
-        wrapper._udf = self.ctx.udf(name)
-        return wrapper
+        return _ScalarUDF._from_internal(self.ctx.udf(name))
 
     def udaf(self, name: str) -> AggregateUDF:
         """Look up a registered aggregate UDF by name.
@@ -1387,9 +1385,7 @@ def udaf(self, name: str) -> AggregateUDF:
             AggregateUDF as _AggregateUDF,
         )
 
-        wrapper = _AggregateUDF.__new__(_AggregateUDF)
-        wrapper._udaf = self.ctx.udaf(name)
-        return wrapper
+        return _AggregateUDF._from_internal(self.ctx.udaf(name))
 
     def udwf(self, name: str) -> WindowUDF:
         """Look up a registered window UDF by name.
@@ -1418,9 +1414,7 @@ def udwf(self, name: str) -> WindowUDF:
         """
         from datafusion.user_defined import WindowUDF as _WindowUDF  # noqa: PLC0415
 
-        wrapper = _WindowUDF.__new__(_WindowUDF)
-        wrapper._udwf = self.ctx.udwf(name)
-        return wrapper
+        return _WindowUDF._from_internal(self.ctx.udwf(name))
 
     def udfs(self) -> list[str]:
         """Return the sorted names of all registered scalar UDFs.

python/datafusion/user_defined.py

@@ -153,6 +153,18 @@ def __init__(
             name, func, input_fields, return_field, str(volatility)
         )
 
+    @classmethod
+    def _from_internal(cls, internal: df_internal.ScalarUDF) -> ScalarUDF:
+        """Wrap an already-constructed internal ``ScalarUDF`` handle.
+
+        Used by :py:meth:`SessionContext.udf` to surface a function looked
+        up from the session's function registry without re-running
+        :py:meth:`__init__`.
+        """
+        wrapper = cls.__new__(cls)
+        wrapper._udf = internal
+        return wrapper
+
     @property
     def name(self) -> str:
         """Return the registered name of this UDF.
@@ -453,6 +465,18 @@ def __init__(
             str(volatility),
         )
 
+    @classmethod
+    def _from_internal(cls, internal: df_internal.AggregateUDF) -> AggregateUDF:
+        """Wrap an already-constructed internal ``AggregateUDF`` handle.
+
+        Used by :py:meth:`SessionContext.udaf` to surface a function looked
+        up from the session's function registry without re-running
+        :py:meth:`__init__`.
+        """
+        wrapper = cls.__new__(cls)
+        wrapper._udaf = internal
+        return wrapper
+
     @property
     def name(self) -> str:
         """Return the registered name of this UDAF.
@@ -873,6 +897,18 @@ def __init__(
             name, func, input_types, return_type, str(volatility)
         )
 
+    @classmethod
+    def _from_internal(cls, internal: df_internal.WindowUDF) -> WindowUDF:
+        """Wrap an already-constructed internal ``WindowUDF`` handle.
+
+        Used by :py:meth:`SessionContext.udwf` to surface a function looked
+        up from the session's function registry without re-running
+        :py:meth:`__init__`.
+        """
+        wrapper = cls.__new__(cls)
+        wrapper._udwf = internal
+        return wrapper
+
     @property
     def name(self) -> str:
         """Return the registered name of this UDWF.