refactor(udf): scalar from_parts now takes Vec<DataType>, matching udaf/udwf

`PythonFunctionScalarUDF::from_parts` previously took `Vec<Field>`
for inputs only to call `.data_type()` on every entry inside `new` —
nullability and metadata went straight in the bin because
`Signature::exact` cannot represent them. The codec carried full
Field info through IPC but threw it away one frame later.

Take `Vec<DataType>` instead. The codec extracts data types from the
decoded IPC schema and feeds them in directly. `return_field` is
still a `Field` so per-output nullability/metadata round-trips
intact, the same way `state_fields` is preserved on the aggregate
side. The Python constructor path (`PyScalarUDF::new`) still goes
through `new` with `Vec<Field>` since PyArrow already hands it the
full Field list.

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

Author: timsaucer

crates/core/src/codec.rs

@@ -527,10 +527,10 @@ fn decode_python_scalar_udf(py: Python<'_>, payload: &[u8]) -> PyResult<PythonFu
 
     let input_schema = schema_from_ipc_bytes(&input_schema_bytes)
         .map_err(|e| pyo3::exceptions::PyValueError::new_err(format!("{e}")))?;
-    let input_fields: Vec<Field> = input_schema
+    let input_types: Vec<arrow::datatypes::DataType> = input_schema
         .fields()
         .iter()
-        .map(|f| f.as_ref().clone())
+        .map(|f| f.data_type().clone())
         .collect();
 
     let return_schema = schema_from_ipc_bytes(&return_schema_bytes)
@@ -552,7 +552,7 @@ fn decode_python_scalar_udf(py: Python<'_>, payload: &[u8]) -> PyResult<PythonFu
     Ok(PythonFunctionScalarUDF::from_parts(
         name,
         func,
-        input_fields,
+        input_types,
         return_field,
         volatility,
     ))

crates/core/src/udf.rs

@@ -79,18 +79,24 @@ impl PythonFunctionScalarUDF {
     }
 
     /// Reconstruct a `PythonFunctionScalarUDF` from the parts emitted
-    /// by the codec. `input_fields` here only contributes `data_type`
-    /// info (collapsed into `Signature::exact`); their names,
-    /// nullability, and metadata are not retained, so the decoder is
-    /// free to fabricate them from `Vec<DataType>`.
+    /// by the codec. Inputs collapse to `Vec<DataType>` because
+    /// `Signature::exact` cannot carry per-input nullability or
+    /// metadata — the encoder is free to discard that side of the
+    /// schema. `return_field` is kept as a `Field` so the post-decode
+    /// nullability and metadata match the sender's instance.
     pub(crate) fn from_parts(
         name: String,
         func: Py<PyAny>,
-        input_fields: Vec<Field>,
+        input_types: Vec<DataType>,
         return_field: Field,
         volatility: Volatility,
     ) -> Self {
-        Self::new(name, func, input_fields, return_field, volatility)
+        Self {
+            name,
+            func,
+            signature: Signature::exact(input_types, volatility),
+            return_field: Arc::new(return_field),
+        }
     }
 }