feat(codec): forward every LogicalExtensionCodec /

PhysicalExtensionCodec method to inner

PythonLogicalCodec previously only overrode the four required methods
on the trait plus the scalar UDF pair, so the default trait impls
(returning "LogicalExtensionCodec is not provided") shadowed any
downstream FFI codec for file formats, aggregate UDFs, and window
UDFs. A user installing their own codec via
`SessionContext.with_logical_extension_codec(...)` would silently
lose access to its `try_*_file_format`, `try_*_udaf`,
`try_*_udwf` implementations.

Forward every trait method to `inner` so the user-installed codec is
fully reachable. Same change on the physical side, including
`try_*_expr`, `try_*_udaf`, `try_*_udwf` — the corresponding
Python-aware paths can layer on later by intercepting before
delegation.

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

Author: timsaucer

crates/core/src/codec.rs

@@ -80,8 +80,10 @@ use std::sync::Arc;
 use arrow::datatypes::SchemaRef;
 use datafusion::common::{Result, TableReference};
 use datafusion::datasource::TableProvider;
+use datafusion::datasource::file_format::FileFormatFactory;
 use datafusion::execution::TaskContext;
-use datafusion::logical_expr::{Extension, LogicalPlan, ScalarUDF};
+use datafusion::logical_expr::{AggregateUDF, Extension, LogicalPlan, ScalarUDF, WindowUDF};
+use datafusion::physical_expr::PhysicalExpr;
 use datafusion::physical_plan::ExecutionPlan;
 use datafusion_proto::logical_plan::{DefaultLogicalExtensionCodec, LogicalExtensionCodec};
 use datafusion_proto::physical_plan::{DefaultPhysicalExtensionCodec, PhysicalExtensionCodec};
@@ -158,13 +160,45 @@ impl LogicalExtensionCodec for PythonLogicalCodec {
         self.inner.try_encode_table_provider(table_ref, node, buf)
     }
 
+    fn try_decode_file_format(
+        &self,
+        buf: &[u8],
+        ctx: &TaskContext,
+    ) -> Result<Arc<dyn FileFormatFactory>> {
+        self.inner.try_decode_file_format(buf, ctx)
+    }
+
+    fn try_encode_file_format(
+        &self,
+        buf: &mut Vec<u8>,
+        node: Arc<dyn FileFormatFactory>,
+    ) -> Result<()> {
+        self.inner.try_encode_file_format(buf, node)
+    }
+
     fn try_encode_udf(&self, node: &ScalarUDF, buf: &mut Vec<u8>) -> Result<()> {
         self.inner.try_encode_udf(node, buf)
     }
 
     fn try_decode_udf(&self, name: &str, buf: &[u8]) -> Result<Arc<ScalarUDF>> {
         self.inner.try_decode_udf(name, buf)
     }
+
+    fn try_encode_udaf(&self, node: &AggregateUDF, buf: &mut Vec<u8>) -> Result<()> {
+        self.inner.try_encode_udaf(node, buf)
+    }
+
+    fn try_decode_udaf(&self, name: &str, buf: &[u8]) -> Result<Arc<AggregateUDF>> {
+        self.inner.try_decode_udaf(name, buf)
+    }
+
+    fn try_encode_udwf(&self, node: &WindowUDF, buf: &mut Vec<u8>) -> Result<()> {
+        self.inner.try_encode_udwf(node, buf)
+    }
+
+    fn try_decode_udwf(&self, name: &str, buf: &[u8]) -> Result<Arc<WindowUDF>> {
+        self.inner.try_decode_udwf(name, buf)
+    }
 }
 
 /// `PhysicalExtensionCodec` mirror of [`PythonLogicalCodec`] parked
@@ -221,4 +255,32 @@ impl PhysicalExtensionCodec for PythonPhysicalCodec {
     fn try_decode_udf(&self, name: &str, buf: &[u8]) -> Result<Arc<ScalarUDF>> {
         self.inner.try_decode_udf(name, buf)
     }
+
+    fn try_encode_expr(&self, node: &Arc<dyn PhysicalExpr>, buf: &mut Vec<u8>) -> Result<()> {
+        self.inner.try_encode_expr(node, buf)
+    }
+
+    fn try_decode_expr(
+        &self,
+        buf: &[u8],
+        inputs: &[Arc<dyn PhysicalExpr>],
+    ) -> Result<Arc<dyn PhysicalExpr>> {
+        self.inner.try_decode_expr(buf, inputs)
+    }
+
+    fn try_encode_udaf(&self, node: &AggregateUDF, buf: &mut Vec<u8>) -> Result<()> {
+        self.inner.try_encode_udaf(node, buf)
+    }
+
+    fn try_decode_udaf(&self, name: &str, buf: &[u8]) -> Result<Arc<AggregateUDF>> {
+        self.inner.try_decode_udaf(name, buf)
+    }
+
+    fn try_encode_udwf(&self, node: &WindowUDF, buf: &mut Vec<u8>) -> Result<()> {
+        self.inner.try_encode_udwf(node, buf)
+    }
+
+    fn try_decode_udwf(&self, name: &str, buf: &[u8]) -> Result<Arc<WindowUDF>> {
+        self.inner.try_decode_udwf(name, buf)
+    }
 }