refactor: keyword-only inlining flag, skip GIL on prefix mismatch

- `SessionContext.with_python_udf_inlining` now keyword-only (`*, enabled`)
  to match the documented call style and the existing doctests/tests.
- `refuse_if_inline` and the three `try_decode_python_*` decoders short-
  circuit on a `starts_with(family)` check before `Python::attach`, so
  plans whose UDFs are not Python-defined no longer pay a GIL acquisition
  per decode call. Semantics preserved: `strip_wire_header` already
  returns `Ok(None)` when the prefix does not match.
- `datafusion.ipc` module docstring wraps the `set_sender_ctx` example in
  `try`/`finally` and notes that the thread-local holds a strong
  reference to the installed `SessionContext` until cleared.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: timsaucer

crates/core/src/codec.rs

@@ -393,7 +393,16 @@ impl LogicalExtensionCodec for PythonLogicalCodec {
 /// surface *their* diagnostic instead of the strict-mode message.
 /// The strict message implies sender intent ("inlining is disabled"),
 /// so it should fire only when the bytes really would have decoded.
+///
+/// Fast path: short-circuit on the family-magic prefix before
+/// acquiring the GIL. Plans with many non-Python UDFs would otherwise
+/// pay a GIL acquisition per decode call just to confirm "not a
+/// Python UDF". `read_framed_payload` itself rejects buffers that
+/// don't start with `family`, so this is purely an optimization.
 fn refuse_if_inline(buf: &[u8], family: &[u8], kind: &str, name: &str) -> Result<()> {
+    if !buf.starts_with(family) {
+        return Ok(());
+    }
     Python::attach(|py| match read_framed_payload(py, buf, family, kind)? {
         Some(_) => Err(refuse_inline_payload(kind, name)),
         None => Ok(()),
@@ -578,6 +587,9 @@ pub(crate) fn try_encode_python_scalar_udf(node: &ScalarUDF, buf: &mut Vec<u8>)
 /// the caller to delegate to its `inner` codec (and eventually the
 /// `FunctionRegistry`).
 pub(crate) fn try_decode_python_scalar_udf(buf: &[u8]) -> Result<Option<Arc<ScalarUDF>>> {
+    if !buf.starts_with(PY_SCALAR_UDF_FAMILY) {
+        return Ok(None);
+    }
     Python::attach(|py| -> Result<Option<Arc<ScalarUDF>>> {
         let Some(payload) = read_framed_payload(py, buf, PY_SCALAR_UDF_FAMILY, "scalar UDF")?
         else {
@@ -834,6 +846,9 @@ pub(crate) fn try_encode_python_udwf(node: &WindowUDF, buf: &mut Vec<u8>) -> Res
 }
 
 pub(crate) fn try_decode_python_udwf(buf: &[u8]) -> Result<Option<Arc<WindowUDF>>> {
+    if !buf.starts_with(PY_WINDOW_UDF_FAMILY) {
+        return Ok(None);
+    }
     Python::attach(|py| -> Result<Option<Arc<WindowUDF>>> {
         let Some(payload) = read_framed_payload(py, buf, PY_WINDOW_UDF_FAMILY, "window UDF")?
         else {
@@ -916,6 +931,9 @@ pub(crate) fn try_encode_python_udaf(node: &AggregateUDF, buf: &mut Vec<u8>) ->
 }
 
 pub(crate) fn try_decode_python_udaf(buf: &[u8]) -> Result<Option<Arc<AggregateUDF>>> {
+    if !buf.starts_with(PY_AGG_UDF_FAMILY) {
+        return Ok(None);
+    }
     Python::attach(|py| -> Result<Option<Arc<AggregateUDF>>> {
         let Some(payload) = read_framed_payload(py, buf, PY_AGG_UDF_FAMILY, "aggregate UDF")?
         else {

python/datafusion/context.py

@@ -1770,7 +1770,7 @@ def with_physical_extension_codec(self, codec: Any) -> SessionContext:
         new.ctx = new_internal
         return new
 
-    def with_python_udf_inlining(self, enabled: bool) -> SessionContext:
+    def with_python_udf_inlining(self, *, enabled: bool) -> SessionContext:
         """Control whether Python UDFs are embedded in serialized expressions.
 
         When ``enabled=True`` (the default), serialized expressions carry

python/datafusion/ipc.py

@@ -62,16 +62,27 @@ def init_worker():
 .. code-block:: python
 
     from datafusion import SessionContext
-    from datafusion.ipc import set_sender_ctx
+    from datafusion.ipc import clear_sender_ctx, set_sender_ctx
 
     driver_ctx = SessionContext().with_python_udf_inlining(enabled=False)
     set_sender_ctx(driver_ctx)
-    pickle.dumps(expr)  # encoded with inlining disabled
+    try:
+        pickle.dumps(expr)  # encoded with inlining disabled
+    finally:
+        clear_sender_ctx()
 
 Without a sender context the default codec is used (Python UDF
 inlining on). The sender context only affects pickle / ``to_bytes``
 encoding; explicit ``expr.to_bytes(ctx)`` calls still use the supplied
 ``ctx``.
+
+The thread-local holds a strong reference to the installed
+:class:`SessionContext` until :func:`clear_sender_ctx` is called or
+the thread exits. Long-running driver threads that install a sender
+context once and never clear it will retain that session for the
+lifetime of the thread; pair :func:`set_sender_ctx` with
+:func:`clear_sender_ctx` (e.g. in a ``try``/``finally``) when the
+sender context is only needed for a bounded scope.
 """
 
 from __future__ import annotations