docs(pickle): user guide + Ray example + CI backstop + UDF.name

Companion to the pickle work in the previous commit. Ships the
discoverable surface a user would actually reach for when they hit
"how do I distribute these expressions":

* `docs/source/user-guide/io/distributing_expressions.rst` — end-to-end
  user guide covering the recommended `Pool(initializer=...)` pattern,
  the worker context shape, what does and does not survive the
  round-trip (scalar UDFs yes, UDAF/UDWF/FFI by name), Python 3.14
  start-method change, and the cloudpickle security note.
* `examples/ray_pickle_expr.py` — runnable Ray actor demo using
  `set_worker_ctx` from an actor `__init__`.
* `examples/README.md` — links to the Ray example.
* `docs/source/user-guide/io/index.rst` — adds the new page to the
  IO TOC.
* `.github/workflows/test.yml` — 30-minute `timeout-minutes` backstop
  on the test matrix so a hung multiprocessing worker (e.g. during a
  pickle regression) does not block CI indefinitely.
* `python/datafusion/user_defined.py` — `ScalarUDF` / `AggregateUDF` /
  `WindowUDF` get a `.name` property surfacing the registered name.
  Useful for tests asserting an expression carries a specific UDF
  reference, and for users debugging worker registrations.

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

Author: timsaucer

.github/workflows/test.yml

@@ -29,6 +29,9 @@ env:
 jobs:
   test-matrix:
     runs-on: ubuntu-latest
+    # Backstop: a hung multiprocessing worker (e.g. during a pickle regression)
+    # should not block CI longer than this.
+    timeout-minutes: 30
     strategy:
       fail-fast: false
       matrix:

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

@@ -0,0 +1,141 @@
+.. Licensed to the Apache Software Foundation (ASF) under one
+.. or more contributor license agreements.  See the NOTICE file
+.. distributed with this work for additional information
+.. regarding copyright ownership.  The ASF licenses this file
+.. to you under the Apache License, Version 2.0 (the
+.. "License"); you may not use this file except in compliance
+.. with the License.  You may obtain a copy of the License at
+
+..   http://www.apache.org/licenses/LICENSE-2.0
+
+.. Unless required by applicable law or agreed to in writing,
+.. software distributed under the License is distributed on an
+.. "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+.. KIND, either express or implied.  See the License for the
+.. specific language governing permissions and limitations
+.. under the License.
+
+Distributing expressions across processes
+=========================================
+
+DataFusion expressions (:py:class:`~datafusion.Expr`) can be serialized and
+shipped across process boundaries — useful for distributing work over a
+``multiprocessing.Pool``, a Ray actor pool, or any framework that supports a
+per-worker initialization hook.
+
+Pickle support
+--------------
+
+:py:class:`~datafusion.Expr` implements the pickle protocol directly. Call
+:py:func:`pickle.dumps` on an expression and ship the bytes; the receiver
+calls :py:func:`pickle.loads`. Python *scalar UDFs* are cloudpickled into the
+proto wire format by a Rust-side codec (``PythonUDFCodec``), so the blob is
+self-contained — the receiver does not need to pre-register the UDF.
+
+.. code-block:: python
+
+    import multiprocessing as mp
+    import pickle
+
+    import pyarrow as pa
+    from datafusion import SessionContext, col, lit, udf
+
+    def init_worker():
+        # Optional: install a worker context for aggregate / window UDFs,
+        # table providers, or Rust-side function registrations. Not needed
+        # for built-ins or Python scalar UDFs.
+        pass
+
+    def evaluate(blob_and_batch):
+        blob, batch = blob_and_batch
+        expr = pickle.loads(blob)
+        ctx = SessionContext()
+        df = ctx.from_pydict({"a": batch})
+        return df.with_column("result", expr).select("result").to_pydict()["result"]
+
+    if __name__ == "__main__":
+        double = udf(
+            lambda arr: pa.array([(v.as_py() or 0) * 2 for v in arr]),
+            [pa.int64()], pa.int64(), volatility="immutable", name="double",
+        )
+        blob = pickle.dumps(double(col("a")))
+
+        mp_ctx = mp.get_context("forkserver")
+        with mp_ctx.Pool(processes=4) as pool:
+            results = pool.map(
+                evaluate,
+                [(blob, [1, 2, 3]), (blob, [10, 20, 30])],
+            )
+        print(results)  # [[2, 4, 6], [20, 40, 60]]
+
+Worker-scoped context
+---------------------
+
+For references the codec cannot inline — aggregate UDFs, window UDFs, FFI
+capsule UDFs, or anything resolved through the
+:class:`SessionContext`'s function registry — set a worker-scoped context
+once per process using :py:func:`datafusion.ipc.set_worker_ctx`:
+
+.. code-block:: python
+
+    from datafusion import SessionContext
+    from datafusion.ipc import set_worker_ctx
+
+    def init_worker():
+        ctx = SessionContext()
+        ctx.register_udaf(my_aggregate)  # if needed
+        set_worker_ctx(ctx)
+
+    with mp.get_context("forkserver").Pool(
+        processes=4, initializer=init_worker
+    ) as pool:
+        ...
+
+Without a worker context, unpickling falls back to a fresh
+:py:class:`SessionContext`. Built-in functions resolve; Python scalar UDFs
+ride along inside the blob via the codec. References to aggregate / window
+UDFs or other registry-only entries raise an informative error if not
+registered on the worker.
+
+Python 3.14 default change
+--------------------------
+
+Python 3.14 changed the POSIX default start method for
+:py:mod:`multiprocessing` from ``fork`` to ``forkserver``. With ``fork``, a
+context set in the parent was visible in workers via copy-on-write; with
+``forkserver`` and ``spawn`` it is not. The codec + worker-init pattern works
+on every start method — prefer it over relying on inherited state.
+
+Trade-offs of inline UDFs
+-------------------------
+
+* **Blob size** — cloudpickled callables add bytes per blob. A trivial
+  built-in expression is ~20 bytes; an expression referencing a Python scalar
+  UDF is hundreds of bytes (the cloudpickled callable + signature). Pre-register
+  shared UDFs on workers via :py:func:`~datafusion.ipc.set_worker_ctx` when
+  the same UDF is shipped many times and you want to avoid the overhead.
+* **Closure capture** — cloudpickle captures closure state. Surprises are
+  possible if the UDF closes over large objects, module-level mutable state,
+  or non-portable file paths.
+* **FFI scalar UDFs cannot be inlined** — PyCapsule-backed UDFs have no
+  Python callable to cloudpickle. The codec leaves their ``fun_definition``
+  empty; the receiver must have a matching registration.
+* **Aggregate and window UDFs cannot be inlined yet** — their Python state
+  is held inside opaque factory closures on the Rust side. Pre-register on
+  the worker.
+
+Security
+--------
+
+.. warning::
+
+   Pickle blobs containing inlined UDFs deserialize via :py:mod:`cloudpickle`,
+   which executes arbitrary code on the receiver. Only :py:func:`pickle.loads`
+   blobs from trusted sources. For untrusted-source workflows, restrict the
+   sender to built-in functions and pre-registered Rust-side UDFs.
+
+See also
+--------
+
+* :py:mod:`datafusion.ipc` — module-level API reference.
+* ``examples/ray_pickle_expr.py`` — runnable Ray actor example.

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

@@ -24,6 +24,7 @@ IO
    arrow
    avro
    csv
+   distributing_expressions
    json
    parquet
    table_provider

examples/README.md

@@ -44,6 +44,10 @@ Here is a direct link to the file used in the examples:
 - [Register a Python UDF with DataFusion](./python-udf.py)
 - [Register a Python UDAF with DataFusion](./python-udaf.py)
 
+### Distributing DataFusion expressions
+
+- [Pickle expressions and send them to Ray actors](./ray_pickle_expr.py)
+
 ### Substrait Support
 
 - [Serialize query plans using Substrait](./substrait.py)

examples/ray_pickle_expr.py

@@ -0,0 +1,96 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Distribute DataFusion expressions to Ray actors.
+
+This example shows the worker-init pattern from the user guide adapted to
+Ray's actor model: each actor builds its own :class:`SessionContext`.
+Python scalar UDFs travel inside the pickle blob via the Rust-side
+``PythonUDFCodec`` — no actor-side pre-registration is required. The
+worker context (set via :func:`datafusion.ipc.set_worker_ctx`) is still
+useful for aggregate/window UDFs or other registry-only entries; we set
+one here to show the pattern.
+
+Prerequisites:
+    pip install ray
+
+Run:
+    python examples/ray_pickle_expr.py
+"""
+
+import pickle
+
+import pyarrow as pa
+import ray
+from datafusion import SessionContext, col, lit, udf
+from datafusion.ipc import set_worker_ctx
+
+
+def _build_double_udf():
+    """Return the demo UDF used by the actors."""
+    return udf(
+        lambda arr: pa.array([(v.as_py() or 0) * 2 for v in arr]),
+        [pa.int64()],
+        pa.int64(),
+        volatility="immutable",
+        name="double",
+    )
+
+
+@ray.remote
+class DataFusionWorker:
+    """A Ray actor with a private :class:`SessionContext`."""
+
+    def __init__(self) -> None:
+        ctx = SessionContext()
+        ctx.register_udf(_build_double_udf())
+        # The worker context is what Expr.__setstate__ consults when
+        # pickled expressions arrive at this actor.
+        set_worker_ctx(ctx)
+        self._ctx = ctx
+
+    def evaluate(self, expr_blob: bytes, batch_pylist: list[int]) -> list[int]:
+        """Unpickle an Expr, run it over an in-memory batch, return results."""
+        expr = pickle.loads(expr_blob)
+        df = self._ctx.from_pydict({"a": batch_pylist})
+        out = df.with_column("result", expr).select("result")
+        return out.to_pydict()["result"]
+
+
+def main() -> None:
+    ray.init(ignore_reinit_error=True)
+
+    sender = SessionContext()
+    sender.register_udf(_build_double_udf())
+    expr = _build_double_udf()(col("a")) + lit(1)
+    blob = pickle.dumps(expr)
+    print(f"pickled expression: {len(blob)} bytes")
+
+    workers = [DataFusionWorker.remote() for _ in range(2)]
+    batches = [[1, 2, 3], [10, 20, 30], [100, 200, 300]]
+    futures = [
+        workers[i % len(workers)].evaluate.remote(blob, batch)
+        for i, batch in enumerate(batches)
+    ]
+    for batch, result in zip(batches, ray.get(futures), strict=True):
+        print(f"input {batch} -> {result}")
+
+    ray.shutdown()
+
+
+if __name__ == "__main__":
+    main()

python/datafusion/user_defined.py

@@ -132,6 +132,7 @@ def __init__(
 
         See helper method :py:func:`udf` for argument details.
         """
+        self._name = name
         if hasattr(func, "__datafusion_scalar_udf__"):
             self._udf = df_internal.ScalarUDF.from_pycapsule(func)
             return
@@ -141,6 +142,11 @@ def __init__(
             name, func, input_fields, return_field, str(volatility)
         )
 
+    @property
+    def name(self) -> str:
+        """Return the registered name of this UDF."""
+        return self._name
+
     def __repr__(self) -> str:
         """Print a string representation of the Scalar UDF."""
         return self._udf.__repr__()
@@ -394,6 +400,7 @@ def __init__(
         See :py:func:`udaf` for a convenience function and argument
         descriptions.
         """
+        self._name = name
         if hasattr(accumulator, "__datafusion_aggregate_udf__"):
             self._udaf = df_internal.AggregateUDF.from_pycapsule(accumulator)
             return
@@ -418,6 +425,11 @@ def __init__(
             str(volatility),
         )
 
+    @property
+    def name(self) -> str:
+        """Return the registered name of this UDAF."""
+        return self._name
+
     def __repr__(self) -> str:
         """Print a string representation of the Aggregate UDF."""
         return self._udaf.__repr__()
@@ -821,13 +833,19 @@ def __init__(
         See :py:func:`udwf` for a convenience function and argument
         descriptions.
         """
+        self._name = name
         if hasattr(func, "__datafusion_window_udf__"):
             self._udwf = df_internal.WindowUDF.from_pycapsule(func)
             return
         self._udwf = df_internal.WindowUDF(
             name, func, input_types, return_type, str(volatility)
         )
 
+    @property
+    def name(self) -> str:
+        """Return the registered name of this UDWF."""
+        return self._name
+
     def __repr__(self) -> str:
         """Print a string representation of the Window UDF."""
         return self._udwf.__repr__()