docs: drop manual pickle.dumps/loads from worker examples

The Pool / Ray-actor examples called `pickle.dumps` on the sender
and `pickle.loads` on the worker explicitly. That's not what real
user code looks like — `multiprocessing.Pool.starmap`, Ray's
`@ray.remote`, and similar frameworks serialize their function
arguments automatically. Showing the manual wrapping makes the API
look more involved than it is and obscures the point: users hand a
DataFusion `Expr` to their distribution framework like any other
Python object, and it Just Works.

Rewrites:

* User guide worker-pool example switches from
  `pool.map(evaluate, [(blob, batch), ...])` (where `blob =
  pickle.dumps(expr)`) to `pool.starmap(evaluate, [(expr, batch),
  ...])`. `evaluate(expr, batch)` receives the reconstructed
  expression directly.
* Ray example drops the `pickle.dumps(expr)` / `pickle.loads(blob)`
  pair; `evaluate(expr, batch)` takes a typed `Expr`. Drops the
  unused `pickle` import.
* Worker-context narrative updated: "expressions reconstructed by
  pickle.loads" -> "expressions arriving from the driver".
* Security warning reworded to mention pickle as the underlying
  mechanism while still framing the contract in user terms (only
  accept expressions from trusted sources).

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

Author: timsaucer

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

@@ -24,27 +24,30 @@ worker processes (``multiprocessing.Pool``, a Ray actor pool, or any other
 framework with a per-worker initialization hook), and have each worker
 evaluate the expression against its own slice of data.
 
-DataFusion expressions support this directly: they can be sent through
-Python's standard `pickle <https://docs.python.org/3/library/pickle.html>`_
-module like any other Python object. Python UDFs — scalar, aggregate, and
-window — travel inside the pickled bytes; the receiver does not need to
-pre-register them.
+DataFusion expressions support this directly: pass one to a worker
+process and Python's standard
+`pickle <https://docs.python.org/3/library/pickle.html>`_ machinery
+serializes it transparently — the same machinery
+:py:meth:`multiprocessing.pool.Pool.map`, Ray's
+``@ray.remote``, and similar libraries already use to ship function
+arguments. Python UDFs — scalar, aggregate, and window — travel inside
+the serialized expression; the receiver does not need to pre-register
+them.
 
 Basic worker-pool example
 -------------------------
 
 .. code-block:: python
 
     import multiprocessing as mp
-    import pickle
 
     import pyarrow as pa
     from datafusion import SessionContext, col, udf
 
 
-    def evaluate(blob_and_batch):
-        blob, batch = blob_and_batch
-        expr = pickle.loads(blob)  # Python UDFs travel inside the bytes.
+    def evaluate(expr, batch):
+        # `expr` arrived here via the pool's automatic pickling —
+        # no manual serialization needed in user code.
         ctx = SessionContext()
         df = ctx.from_pydict({"a": batch})
         return df.with_column("result", expr).select("result").to_pydict()["result"]
@@ -55,13 +58,13 @@ Basic worker-pool example
             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")))
+        expr = double(col("a"))
 
         mp_ctx = mp.get_context("forkserver")
         with mp_ctx.Pool(processes=4) as pool:
-            results = pool.map(
+            results = pool.starmap(
                 evaluate,
-                [(blob, [1, 2, 3]), (blob, [10, 20, 30])],
+                [(expr, [1, 2, 3]), (expr, [10, 20, 30])],
             )
         print(results)  # [[2, 4, 6], [20, 40, 60]]
 
@@ -108,8 +111,8 @@ must resolve from its registered functions), set up the worker's
     ) as pool:
         ...
 
-Inside a worker, expressions reconstructed by :py:func:`pickle.loads` resolve
-their by-name references against the installed worker context. If no worker
+Inside a worker, expressions arriving from the driver 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 UDFs, but
 FFI-capsule-backed registrations will fail to resolve.
@@ -144,10 +147,10 @@ Security
 .. warning::
 
    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`.
+   Python code on the receiver — pickle is doing the work under the hood
+   and pickle is unsafe on untrusted input. Only accept expressions from
+   trusted sources. For untrusted-source workflows, restrict senders to
+   built-in functions and pre-registered Rust-side UDFs.
 
 See also
 --------

examples/ray_pickle_expr.py

@@ -30,11 +30,9 @@
     python examples/ray_pickle_expr.py
 """
 
-import pickle
-
 import pyarrow as pa
 import ray
-from datafusion import SessionContext, col, lit, udf
+from datafusion import Expr, SessionContext, col, lit, udf
 from datafusion.ipc import set_worker_ctx
 
 
@@ -62,9 +60,10 @@ def __init__(self) -> None:
         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)
+    def evaluate(self, expr: Expr, batch_pylist: list[int]) -> list[int]:
+        """Run the expression against an in-memory batch."""
+        # `expr` arrived here via Ray's automatic argument serialization —
+        # no manual pickle handling needed in user code.
         df = self._ctx.from_pydict({"a": batch_pylist})
         out = df.with_column("result", expr).select("result")
         return out.to_pydict()["result"]
@@ -76,13 +75,11 @@ def main() -> None:
     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)
+        workers[i % len(workers)].evaluate.remote(expr, batch)
         for i, batch in enumerate(batches)
     ]
     for batch, result in zip(batches, ray.get(futures), strict=True):