docs: extend the inline-UDF guarantee to aggregate + window UDFs

With aggregate UDFs and window UDFs now reconstructable from bytes
alone, the user-facing contract simplifies to:

* Built-in functions and **all** Python UDFs (scalar, aggregate,
  window) travel inside the shipped expression. No worker-side
  pre-registration.
* Only UDFs imported via the FFI capsule protocol travel by name and
  require pre-registration via `set_worker_ctx`.

Update each user-facing surface:

* `docs/source/user-guide/io/distributing_expressions.rst` — drop the
  "aggregate/window UDFs travel by name only" caveat; rename the
  practical-considerations entry that called out the limitation.
* `python/datafusion/ipc.py` module + `clear_worker_ctx` — explicitly
  list scalar, aggregate, and window as inline-portable.
* `python/datafusion/expr.py` — `to_bytes` and `__reduce__`
  docstrings updated.
* Test module docstrings updated.

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

Author: timsaucer

docs/source/user-guide/io/distributing_expressions.rst

@@ -69,20 +69,25 @@ What travels with the expression
 
 * **Built-in functions** (``abs``, ``length``, arithmetic, comparisons, etc.)
   — fully portable. Worker needs nothing pre-registered.
-* **Python scalar UDFs** (defined with :py:func:`datafusion.udf`) — fully
-  portable. The callable and its signature travel inside the pickled bytes
-  and are reconstructed on the worker automatically.
-* **Aggregate UDFs**, **window UDFs**, **UDFs imported via the FFI capsule
-  protocol** — travel **by name only**. The worker must already have a
-  matching registration on its :py:class:`SessionContext`. Without that
-  registration, evaluation raises an error.
+* **Python UDFs** — fully portable. The callable, its signature, and any
+  state captured in closures travel inside the pickled bytes and are
+  reconstructed on the worker automatically. Applies equally to:
+
+  * **scalar UDFs** (:py:func:`datafusion.udf`)
+  * **aggregate UDFs** (:py:func:`datafusion.udaf`)
+  * **window UDFs** (:py:func:`datafusion.udwf`)
+* **UDFs imported via the FFI capsule protocol** — travel **by name only**.
+  The worker must already have a matching registration on its
+  :py:class:`SessionContext`. Without that registration, evaluation raises
+  an error.
 
 Registering shared UDFs on workers
 ----------------------------------
 
-When an expression references something that travels by name only (aggregate
-UDF, window UDF, FFI UDF), set up the worker's :py:class:`SessionContext`
-once per process and install it as the *worker context*:
+When an expression references an FFI capsule UDF (or any UDF the worker
+must resolve from its registered functions), set up the worker's
+:py:class:`SessionContext` once per process and install it as the
+*worker context*:
 
 .. code-block:: python
 
@@ -92,7 +97,7 @@ once per process and install it as the *worker context*:
 
     def init_worker():
         ctx = SessionContext()
-        ctx.register_udaf(my_aggregate)
+        ctx.register_udaf(my_ffi_aggregate)
         set_worker_ctx(ctx)
 
 
@@ -104,8 +109,8 @@ once per process and install it as the *worker context*:
 Inside a worker, expressions reconstructed by :py:func:`pickle.loads` resolve
 their by-name references against the installed worker context. If no worker
 context is installed, a fresh empty :py:class:`SessionContext` is used —
-fine for expressions that only reference built-ins and Python scalar UDFs,
-but anything by-name-only will fail to resolve.
+fine for expressions that only reference built-ins and Python UDFs, but
+FFI-capsule-backed registrations will fail to resolve.
 
 Python 3.14 default change
 --------------------------
@@ -122,30 +127,25 @@ Practical considerations
 
 * **Pickled size scales with what travels inline.** A pickled expression of
   just built-ins is small (tens of bytes). An expression carrying a Python
-  scalar UDF is hundreds of bytes (the callable and its signature). When the
-  same UDF is shipped many times, pre-registering it on each worker via
-  :py:func:`~datafusion.ipc.set_worker_ctx` and referring to it by name
-  cuts the per-blob overhead.
-* **Closure capture.** When a Python scalar UDF closes over surrounding
-  state — local variables, module-level objects, file paths — that state
-  is captured at pickling time. Surprises are possible if the captured
-  state is large, mutable, or not portable to the worker's environment.
-* **Aggregate and window UDFs always travel by name.** Their Python state
-  is held inside opaque factory closures that cannot be reconstructed from
-  bytes alone. Use :py:func:`~datafusion.ipc.set_worker_ctx` to register
-  them on each worker.
+  UDF is hundreds of bytes (the callable and its signature). When the same
+  UDF is shipped many times, registering an equivalent FFI-capsule UDF on
+  each worker via :py:func:`~datafusion.ipc.set_worker_ctx` and referring
+  to it by name cuts the per-blob overhead.
+* **Closure capture.** When a Python UDF closes over surrounding state —
+  local variables, module-level objects, file paths — that state is
+  captured at pickling time. Surprises are possible if the captured state
+  is large, mutable, or not portable to the worker's environment.
 
 Security
 --------
 
 .. warning::
 
-   Reconstructing an expression containing a Python scalar UDF executes
-   arbitrary Python code on the receiver. Only :py:func:`pickle.loads`
-   expressions from trusted sources. For untrusted-source workflows,
-   restrict senders to built-in functions and pre-registered Rust-side
-   UDFs, and never feed externally supplied bytes through
-   :py:func:`pickle.loads`.
+   Reconstructing an expression containing a Python UDF executes arbitrary
+   Python code on the receiver. Only :py:func:`pickle.loads` expressions
+   from trusted sources. For untrusted-source workflows, restrict senders
+   to built-in functions and pre-registered Rust-side UDFs, and never feed
+   externally supplied bytes through :py:func:`pickle.loads`.
 
 See also
 --------

python/datafusion/expr.py

@@ -439,11 +439,11 @@ def to_bytes(self, ctx: SessionContext | None = None) -> bytes:
         Use this — or :func:`pickle.dumps` — to send an expression to a
         worker process for distributed evaluation.
 
-        Built-in functions and Python scalar UDFs travel inside the
-        returned bytes; the worker does not need to pre-register them.
-        Aggregate UDFs, window UDFs, and UDFs imported via the FFI
-        capsule protocol travel by name only and must be registered on
-        the worker. See :doc:`/user-guide/io/distributing_expressions`.
+        Built-in functions and Python UDFs (scalar, aggregate, window)
+        travel inside the returned bytes; the worker does not need to
+        pre-register them. UDFs imported via the FFI capsule protocol
+        travel by name only and must be registered on the worker. See
+        :doc:`/user-guide/io/distributing_expressions`.
         """
         ctx_arg = ctx.ctx if ctx is not None else None
         return bytes(self.expr.to_bytes(ctx_arg))
@@ -470,8 +470,10 @@ def __reduce__(self) -> tuple:
         """Pickle protocol hook.
 
         Lets expressions be shipped to worker processes via
-        :func:`pickle.dumps` / :func:`pickle.loads`. The worker's
-        :class:`SessionContext` for resolving by-name references is
+        :func:`pickle.dumps` / :func:`pickle.loads`. Built-in functions
+        and Python UDFs travel inside the pickle bytes; only FFI-capsule
+        UDFs require pre-registration on the worker. The worker's
+        :class:`SessionContext` for resolving those references is
         looked up via :func:`datafusion.ipc.set_worker_ctx`, falling
         back to a fresh empty :class:`SessionContext` if none has been
         installed on the worker.

python/datafusion/ipc.py

@@ -20,22 +20,23 @@
 When a :class:`Expr` is shipped to a worker process (e.g. through
 :func:`multiprocessing.Pool` or a Ray actor), the worker reconstructs the
 expression against a :class:`SessionContext`. If the expression references
-aggregate UDFs, window UDFs, table providers, or UDFs imported via the FFI
-capsule protocol — anything the worker would otherwise resolve from its
-registered functions — install a configured :class:`SessionContext` once
-per worker:
+UDFs imported via the FFI capsule protocol — or any UDF the worker would
+otherwise resolve from its registered functions rather than from inside
+the shipped expression — install a configured :class:`SessionContext`
+once per worker:
 
 >>> # doctest: +SKIP
 >>> from datafusion import SessionContext
 >>> from datafusion.ipc import set_worker_ctx
 >>>
 >>> def init_worker():
 ...     ctx = SessionContext()
-...     ctx.register_udaf(my_aggregate)
+...     ctx.register_udaf(my_ffi_aggregate)
 ...     set_worker_ctx(ctx)
 
-Built-in functions and Python scalar UDFs travel inside the shipped
-expression itself and do not need pre-registration on the worker.
+Built-in functions and Python UDFs (scalar, aggregate, window) travel
+inside the shipped expression itself and do not need pre-registration
+on the worker.
 
 See :doc:`/user-guide/io/distributing_expressions` for the full pattern.
 """
@@ -75,8 +76,8 @@ def clear_worker_ctx() -> None:
 
     After clearing, expressions reconstructed in this worker fall back to a
     fresh empty :class:`SessionContext` — adequate for built-ins and Python
-    scalar UDFs, but anything that travels by name only (aggregate UDFs,
-    window UDFs, FFI UDFs) will fail to resolve.
+    UDFs (scalar, aggregate, window), but anything imported via the FFI
+    capsule protocol will fail to resolve.
     """
     if hasattr(_local, "ctx"):
         del _local.ctx

python/tests/test_pickle_expr.py

@@ -17,11 +17,10 @@
 
 """In-process pickle round-trip tests for :class:`Expr`.
 
-Built-in functions and Python scalar UDFs travel with the pickled
-expression and do not need worker-side pre-registration. The worker
-context (:mod:`datafusion.ipc`) is only consulted for references that
-travel by name — aggregate UDFs, window UDFs, UDFs imported via the FFI
-capsule protocol.
+Built-in functions and Python UDFs (scalar, aggregate, window) travel
+with the pickled expression and do not need worker-side pre-registration.
+The worker context (:mod:`datafusion.ipc`) is only consulted for UDFs
+imported via the FFI capsule protocol.
 
 Cross-process tests live in ``test_pickle_multiprocessing.py``.
 """

python/tests/test_pickle_multiprocessing.py

@@ -18,11 +18,11 @@
 """Cross-process pickle tests for :class:`Expr`.
 
 Workers run with each :mod:`multiprocessing` start method (``fork``,
-``forkserver``, ``spawn``). Python scalar UDFs travel with the pickled
-expression and need no worker-side pre-registration. Worker-side helpers
-live in ``_pickle_multiprocessing_helpers`` — the underscore prefix
-avoids pytest collection so the module imports under its real name in
-worker subprocesses.
+``forkserver``, ``spawn``). Python UDFs (scalar, aggregate, window) travel
+with the pickled expression and need no worker-side pre-registration.
+Worker-side helpers live in ``_pickle_multiprocessing_helpers`` — the
+underscore prefix avoids pytest collection so the module imports under
+its real name in worker subprocesses.
 """
 
 from __future__ import annotations