refactor: tighten PR1 scope to codec plumbing only

Review feedback pass. PR1 is now strictly the composable codec layer +
session routing + class-method serialization API. Anything that
touches actual Python UDF inline encoding or Python expression
wrapping moves to PR2 alongside the pickle work.

Dropped:

* `encode_python_scalar_udf` / `decode_python_scalar_udf` helpers
  from `crates/core/src/codec.rs`, along with cloudpickle and pyarrow
  imports. The wrapper codecs now delegate every method to `inner`.
  `DFPYUDF1` magic constant is kept (marked `dead_code` for now) as a
  reservation so PR2 has a single definition site.
* `udf.rs` reverted to pre-PR1 shape. The codec no longer needs
  `func()` / `input_fields()` / `volatility()` / `from_parts()`
  accessors. Re-added by PR2 when scalar-UDF inlining lands.
* `PyPhysicalExpr` class + Python wrapper + `__init__` export +
  `MyPhysicalExpr` FFI fixture + `_test_physical_expr.py`. No
  consumer in PR1 or PR2 plan documents; symmetry with
  `PyExecutionPlan` is not enough to justify the surface area.
* Rust-side `PyLogicalPlan::to_proto` / `from_proto` and
  `PyExecutionPlan::to_proto` / `from_proto` deprecated wrappers.
  The deprecation lives entirely in the Python wrapper layer, which
  emits `DeprecationWarning` and forwards to `to_bytes` /
  `from_bytes`. Less Rust duplication.
* `PythonLogicalCodec::with_default_inner` /
  `PythonPhysicalCodec::with_default_inner` — redundant with
  `impl Default`. Logic moved into `Default::default`.
* `PySessionContext::default_logical_codec` /
  `default_physical_codec` helpers. Inlined as
  `Arc::new(PythonLogicalCodec::default())` at the three call sites.

Tests (root: 1076, FFI example: 36) all green after the cuts.

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

Author: timsaucer

crates/core/src/codec.rs

@@ -18,68 +18,60 @@
 //! Python-aware extension codecs.
 //!
 //! [`PythonLogicalCodec`] wraps a user-supplied (or default)
-//! [`LogicalExtensionCodec`] and adds in-band encoding of Python-defined
-//! scalar UDFs via cloudpickle. [`PythonPhysicalCodec`] is the symmetric
-//! wrapper around [`PhysicalExtensionCodec`]; both reuse the same payload
-//! framing so a Python `ScalarUDF` round-trips identically through either
-//! codec layer.
+//! [`LogicalExtensionCodec`] and is the codec datafusion-python parks
+//! on every `SessionContext`. [`PythonPhysicalCodec`] is the symmetric
+//! wrapper around [`PhysicalExtensionCodec`].
 //!
-//! ## Wire-format magic prefix registry
+//! In PR1 both codecs delegate every call to their `inner` codec. The
+//! types exist so that follow-up work (pickle support, Python scalar
+//! UDF inline encoding) can add in-band Python payloads without
+//! re-plumbing the session field.
 //!
-//! Each Python-inline payload begins with an 8-byte magic prefix so the
-//! decoder can distinguish it from arbitrary `fun_definition` bytes (e.g.
-//! produced by a user FFI codec or left empty by the default codec).
+//! ## Wire-format magic prefix registry
 //!
-//! | Layer + kind                  | Magic prefix | Owner          |
-//! | ----------------------------- | ------------ | -------------- |
-//! | `PythonLogicalCodec` scalar   | `DFPYUDF1`   | in use         |
-//! | `PythonLogicalCodec` agg      | `DFPYUDA1`   | reserved       |
-//! | `PythonLogicalCodec` window   | `DFPYUDW1`   | reserved       |
-//! | `PythonPhysicalCodec` scalar  | `DFPYUDF1`   | in use (shared with logical) |
-//! | `PythonPhysicalCodec` agg     | `DFPYUDA1`   | reserved       |
-//! | `PythonPhysicalCodec` window  | `DFPYUDW1`   | reserved       |
-//! | `PythonPhysicalCodec` expr    | `DFPYPE1`    | reserved       |
-//! | User FFI extension codec      | user-chosen  | downstream     |
-//! | Default codec                 | (none)       | upstream       |
+//! Future in-band Python payloads will be prefixed with an 8-byte
+//! magic so the decoder can distinguish them from arbitrary
+//! `fun_definition` bytes produced by the default codec or a user FFI
+//! codec.
 //!
-//! Dispatch precedence inside the Python codecs: **Python-inline payload
-//! (magic prefix match) → `inner` codec → caller's registry fallback.**
-//! When a payload does not match a Python magic prefix the call delegates
-//! to `inner`, which is typically `DefaultLogicalExtensionCodec` /
-//! `DefaultPhysicalExtensionCodec` but may be a user-supplied FFI codec
-//! installed via `SessionContext.with_logical_extension_codec(...)` /
-//! `with_physical_extension_codec(...)`.
+//! | Layer + kind                  | Magic prefix | Status        |
+//! | ----------------------------- | ------------ | ------------- |
+//! | `PythonLogicalCodec` scalar   | `DFPYUDF1`   | reserved (PR2)|
+//! | `PythonLogicalCodec` agg      | `DFPYUDA1`   | reserved      |
+//! | `PythonLogicalCodec` window   | `DFPYUDW1`   | reserved      |
+//! | `PythonPhysicalCodec` scalar  | `DFPYUDF1`   | reserved (PR2)|
+//! | `PythonPhysicalCodec` agg     | `DFPYUDA1`   | reserved      |
+//! | `PythonPhysicalCodec` window  | `DFPYUDW1`   | reserved      |
+//! | `PythonPhysicalCodec` expr    | `DFPYPE1`    | reserved      |
+//! | User FFI extension codec      | user-chosen  | downstream    |
+//! | Default codec                 | (none)       | upstream      |
 //!
-//! User FFI codecs should pick non-colliding prefixes (recommend a `DF`
-//! namespace plus a crate-specific suffix).
+//! Dispatch precedence once in-band payloads land: **Python-inline
+//! payload (magic prefix match) → `inner` codec → caller's registry
+//! fallback.** User FFI codecs should pick non-colliding prefixes
+//! (recommend a `DF` namespace plus a crate-specific suffix).
 
 use std::sync::Arc;
 
-use arrow::datatypes::{Field, Schema};
-use arrow::pyarrow::ToPyArrow;
-use datafusion::arrow::datatypes::SchemaRef;
-use datafusion::arrow::pyarrow::FromPyArrow;
+use arrow::datatypes::SchemaRef;
 use datafusion::common::{Result, TableReference};
 use datafusion::datasource::TableProvider;
 use datafusion::execution::TaskContext;
-use datafusion::logical_expr::{Extension, LogicalPlan, ScalarUDF, ScalarUDFImpl};
+use datafusion::logical_expr::{Extension, LogicalPlan, ScalarUDF};
 use datafusion::physical_plan::ExecutionPlan;
 use datafusion_proto::logical_plan::{DefaultLogicalExtensionCodec, LogicalExtensionCodec};
 use datafusion_proto::physical_plan::{DefaultPhysicalExtensionCodec, PhysicalExtensionCodec};
-use pyo3::BoundObject;
-use pyo3::prelude::*;
-use pyo3::types::{PyBytes, PyTuple};
-
-use crate::udf::PythonFunctionScalarUDF;
 
-/// Magic prefix for an inlined Python scalar UDF payload. Shared between
-/// logical and physical codec layers — the cloudpickled tuple shape is
-/// identical, so a `ScalarUDF` reaching either codec encodes the same way.
+/// Reserved magic prefix for an inlined Python scalar UDF payload.
+/// Not produced or consumed by PR1; the constant is reserved here so
+/// follow-up work has a single definition site.
+#[allow(dead_code)]
 pub(crate) const PY_SCALAR_UDF_MAGIC: &[u8] = b"DFPYUDF1";
 
-/// `LogicalExtensionCodec` that serializes Python scalar UDFs inline and
-/// delegates every other call to a composable `inner` codec. See module
-/// docs for the dispatch precedence.
+/// `LogicalExtensionCodec` parked on every `SessionContext`. Wraps a
+/// composable `inner` codec; PR1 delegates every method straight
+/// through. The wrapper exists so follow-up patches can add Python
+/// in-band encoding without changing every serializer.
 #[derive(Debug)]
 pub struct PythonLogicalCodec {
     inner: Arc<dyn LogicalExtensionCodec>,
@@ -90,19 +82,14 @@ impl PythonLogicalCodec {
         Self { inner }
     }
 
-    /// Convenience constructor wrapping `DefaultLogicalExtensionCodec`.
-    pub fn with_default_inner() -> Self {
-        Self::new(Arc::new(DefaultLogicalExtensionCodec {}))
-    }
-
     pub fn inner(&self) -> &Arc<dyn LogicalExtensionCodec> {
         &self.inner
     }
 }
 
 impl Default for PythonLogicalCodec {
     fn default() -> Self {
-        Self::with_default_inner()
+        Self::new(Arc::new(DefaultLogicalExtensionCodec {}))
     }
 }
 
@@ -141,29 +128,17 @@ impl LogicalExtensionCodec for PythonLogicalCodec {
     }
 
     fn try_encode_udf(&self, node: &ScalarUDF, buf: &mut Vec<u8>) -> Result<()> {
-        if try_encode_python_scalar_udf(node, buf)? {
-            return Ok(());
-        }
         self.inner.try_encode_udf(node, buf)
     }
 
     fn try_decode_udf(&self, name: &str, buf: &[u8]) -> Result<Arc<ScalarUDF>> {
-        match try_decode_python_scalar_udf(buf)? {
-            Some(udf) => Ok(udf),
-            None => self.inner.try_decode_udf(name, buf),
-        }
+        self.inner.try_decode_udf(name, buf)
     }
 }
 
-/// `PhysicalExtensionCodec` mirror of [`PythonLogicalCodec`]. Scalar-UDF
-/// payloads use the shared `DFPYUDF1` framing so an `ExecutionPlan` or
-/// `PhysicalExpr` that references a Python `ScalarUDF` round-trips
-/// regardless of which codec layer encoded it.
-///
-/// Encoding for Python-defined `ExecutionPlan` impls and `PhysicalExpr`
-/// impls (the `DFPYPE*` namespace) is reserved for a future change — no
-/// concrete Python-side physical extension type exists today, so all
-/// non-UDF calls delegate to `inner`.
+/// `PhysicalExtensionCodec` mirror of [`PythonLogicalCodec`]. Same
+/// motivation: a stable session field that follow-up patches can layer
+/// Python in-band encoding onto.
 #[derive(Debug)]
 pub struct PythonPhysicalCodec {
     inner: Arc<dyn PhysicalExtensionCodec>,
@@ -174,18 +149,14 @@ impl PythonPhysicalCodec {
         Self { inner }
     }
 
-    pub fn with_default_inner() -> Self {
-        Self::new(Arc::new(DefaultPhysicalExtensionCodec {}))
-    }
-
     pub fn inner(&self) -> &Arc<dyn PhysicalExtensionCodec> {
         &self.inner
     }
 }
 
 impl Default for PythonPhysicalCodec {
     fn default() -> Self {
-        Self::with_default_inner()
+        Self::new(Arc::new(DefaultPhysicalExtensionCodec {}))
     }
 }
 
@@ -204,128 +175,10 @@ impl PhysicalExtensionCodec for PythonPhysicalCodec {
     }
 
     fn try_encode_udf(&self, node: &ScalarUDF, buf: &mut Vec<u8>) -> Result<()> {
-        if try_encode_python_scalar_udf(node, buf)? {
-            return Ok(());
-        }
         self.inner.try_encode_udf(node, buf)
     }
 
     fn try_decode_udf(&self, name: &str, buf: &[u8]) -> Result<Arc<ScalarUDF>> {
-        match try_decode_python_scalar_udf(buf)? {
-            Some(udf) => Ok(udf),
-            None => self.inner.try_decode_udf(name, buf),
-        }
-    }
-}
-
-/// Encode a Python scalar UDF inline if `node` is one. Returns `Ok(true)`
-/// when the payload (magic prefix + cloudpickle tuple) was written, or
-/// `Ok(false)` to signal the caller should delegate to its `inner` codec
-/// (non-Python UDF, e.g. built-in or FFI capsule).
-pub(crate) fn try_encode_python_scalar_udf(node: &ScalarUDF, buf: &mut Vec<u8>) -> Result<bool> {
-    let Some(py_udf) = node
-        .inner()
-        .as_any()
-        .downcast_ref::<PythonFunctionScalarUDF>()
-    else {
-        return Ok(false);
-    };
-
-    Python::attach(|py| -> Result<bool> {
-        let bytes = encode_python_scalar_udf(py, py_udf)
-            .map_err(|e| datafusion::error::DataFusionError::External(Box::new(e)))?;
-        buf.extend_from_slice(PY_SCALAR_UDF_MAGIC);
-        buf.extend_from_slice(&bytes);
-        Ok(true)
-    })
-}
-
-/// Decode an inline Python scalar UDF payload. Returns `Ok(None)` when
-/// `buf` does not carry the magic prefix, signalling the caller should
-/// delegate to its `inner` codec (which will typically defer to the
-/// `FunctionRegistry`).
-pub(crate) fn try_decode_python_scalar_udf(buf: &[u8]) -> Result<Option<Arc<ScalarUDF>>> {
-    if buf.is_empty() || !buf.starts_with(PY_SCALAR_UDF_MAGIC) {
-        return Ok(None);
+        self.inner.try_decode_udf(name, buf)
     }
-    let payload = &buf[PY_SCALAR_UDF_MAGIC.len()..];
-
-    Python::attach(|py| -> Result<Option<Arc<ScalarUDF>>> {
-        let udf = decode_python_scalar_udf(py, payload)
-            .map_err(|e| datafusion::error::DataFusionError::External(Box::new(e)))?;
-        Ok(Some(Arc::new(ScalarUDF::new_from_impl(udf))))
-    })
-}
-
-/// Build the cloudpickle payload for a `PythonFunctionScalarUDF`.
-///
-/// Layout: `cloudpickle.dumps((name, func, input_schema_bytes,
-/// return_field, volatility_str))`. Input fields ride along as an
-/// IPC-encoded pyarrow Schema so they round-trip without extra plumbing.
-fn encode_python_scalar_udf(py: Python<'_>, udf: &PythonFunctionScalarUDF) -> PyResult<Vec<u8>> {
-    let cloudpickle = py.import("cloudpickle")?;
-
-    let input_schema = Schema::new(udf.input_fields().to_vec());
-    let pa_schema_obj = input_schema.to_pyarrow(py)?;
-    let pa_schema = pa_schema_obj.into_bound();
-    let schema_bytes: Vec<u8> = pa_schema
-        .call_method0("serialize")?
-        .call_method0("to_pybytes")?
-        .extract()?;
-
-    let return_field_obj = udf.return_field().as_ref().to_pyarrow(py)?;
-    let volatility = format!("{:?}", udf.volatility()).to_lowercase();
-
-    let payload = PyTuple::new(
-        py,
-        [
-            udf.name().into_pyobject(py)?.into_any(),
-            udf.func().bind(py).clone().into_any(),
-            PyBytes::new(py, &schema_bytes).into_any(),
-            return_field_obj.into_bound(),
-            volatility.into_pyobject(py)?.into_any(),
-        ],
-    )?;
-
-    let blob = cloudpickle.call_method1("dumps", (payload,))?;
-    blob.extract::<Vec<u8>>()
-}
-
-/// Inverse of [`encode_python_scalar_udf`].
-fn decode_python_scalar_udf(py: Python<'_>, payload: &[u8]) -> PyResult<PythonFunctionScalarUDF> {
-    let cloudpickle = py.import("cloudpickle")?;
-    let pyarrow = py.import("pyarrow")?;
-
-    let tuple = cloudpickle
-        .call_method1("loads", (PyBytes::new(py, payload),))?
-        .cast_into::<PyTuple>()?;
-
-    let name: String = tuple.get_item(0)?.extract()?;
-    let func: Py<PyAny> = tuple.get_item(1)?.unbind();
-    let schema_bytes: Vec<u8> = tuple.get_item(2)?.extract()?;
-    let return_field_py = tuple.get_item(3)?;
-    let volatility_str: String = tuple.get_item(4)?.extract()?;
-
-    let buffer = pyarrow.call_method1("py_buffer", (PyBytes::new(py, &schema_bytes),))?;
-    let pa_schema = pyarrow
-        .getattr("ipc")?
-        .call_method1("read_schema", (buffer,))?;
-
-    let schema = Schema::from_pyarrow_bound(&pa_schema)
-        .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
-    let input_fields: Vec<Field> = schema.fields().iter().map(|f| f.as_ref().clone()).collect();
-
-    let return_field = Field::from_pyarrow_bound(&return_field_py)
-        .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
-
-    let volatility = datafusion_python_util::parse_volatility(&volatility_str)
-        .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
-
-    Ok(PythonFunctionScalarUDF::from_parts(
-        name,
-        func,
-        input_fields,
-        return_field,
-        volatility,
-    ))
 }

crates/core/src/context.rs

@@ -398,12 +398,10 @@ impl PySessionContext {
             .with_default_features()
             .build();
         let ctx = Arc::new(SessionContext::new_with_state(session_state));
-        let logical_codec = Self::default_logical_codec();
-        let physical_codec = Self::default_physical_codec();
         Ok(PySessionContext {
             ctx,
-            logical_codec,
-            physical_codec,
+            logical_codec: Arc::new(PythonLogicalCodec::default()),
+            physical_codec: Arc::new(PythonPhysicalCodec::default()),
         })
     }
 
@@ -419,12 +417,10 @@ impl PySessionContext {
     #[pyo3(signature = ())]
     pub fn global_ctx() -> PyResult<Self> {
         let ctx = get_global_ctx().clone();
-        let logical_codec = Self::default_logical_codec();
-        let physical_codec = Self::default_physical_codec();
         Ok(Self {
             ctx,
-            logical_codec,
-            physical_codec,
+            logical_codec: Arc::new(PythonLogicalCodec::default()),
+            physical_codec: Arc::new(PythonPhysicalCodec::default()),
         })
     }
 
@@ -1458,14 +1454,6 @@ impl PySessionContext {
         Ok(())
     }
 
-    fn default_logical_codec() -> Arc<PythonLogicalCodec> {
-        Arc::new(PythonLogicalCodec::default())
-    }
-
-    fn default_physical_codec() -> Arc<PythonPhysicalCodec> {
-        Arc::new(PythonPhysicalCodec::default())
-    }
-
     /// Session-scoped logical codec. Sibling modules read this when they
     /// need to serialize/deserialize logical-layer types (LogicalPlan,
     /// Expr) against the user-installed (or default) codec stack.
@@ -1525,14 +1513,10 @@ impl From<PySessionContext> for SessionContext {
 
 impl From<SessionContext> for PySessionContext {
     fn from(ctx: SessionContext) -> PySessionContext {
-        let ctx = Arc::new(ctx);
-        let logical_codec = Self::default_logical_codec();
-        let physical_codec = Self::default_physical_codec();
-
         PySessionContext {
-            ctx,
-            logical_codec,
-            physical_codec,
+            ctx: Arc::new(ctx),
+            logical_codec: Arc::new(PythonLogicalCodec::default()),
+            physical_codec: Arc::new(PythonPhysicalCodec::default()),
         }
     }
 }

crates/core/src/lib.rs

@@ -97,7 +97,6 @@ fn _internal(py: Python, m: Bound<'_, PyModule>) -> PyResult<()> {
     m.add_class::<metrics::PyMetricsSet>()?;
     m.add_class::<metrics::PyMetric>()?;
     m.add_class::<physical_plan::PyExecutionPlan>()?;
-    m.add_class::<physical_plan::PyPhysicalExpr>()?;
     m.add_class::<record_batch::PyRecordBatch>()?;
     m.add_class::<record_batch::PyRecordBatchStream>()?;