feat(ffi): add an FFI extension codec bundle

.ai/skills/datafusion-ffi/SKILL.md

@@ -42,7 +42,7 @@ A new `FFI_X` for trait `X` must follow this template. Use `FFI_CatalogProvider`
 pub struct FFI_X {
     some_method: unsafe extern "C" fn(this: &Self, ...) -> FFI_Result<...>,
     optional_method: Option<unsafe extern "C" fn(&Self, ...) -> FFI_Result<...>>,
-    pub logical_codec: FFI_LogicalExtensionCodec,
+    pub codecs: FFI_ExtensionCodecBundle,
 
     clone: unsafe extern "C" fn(&Self) -> Self,
     release: unsafe extern "C" fn(&mut Self),
@@ -60,7 +60,7 @@ Field rules:
 
 - **One `unsafe extern "C" fn` per trait method.** Always populate — `Arc<dyn Trait>` dispatch picks override-or-default at call time, so the producer side gets the right answer without the consumer needing to know. See § "Method coverage".
 - **`Option<fn>` is the capability-flag exception**, not a template. Crate uses it exactly once: `FFI_TableProvider::supports_filters_pushdown`. See § "Method coverage".
-- **Codec field** (`FFI_LogicalExtensionCodec` / `FFI_PhysicalExtensionCodec`) only if the trait moves `Expr`s / `LogicalPlan`s / `ExecutionPlan`s across the boundary.
+- **Codec field** — carry an `FFI_ExtensionCodecBundle` (`src/proto/extension_codec_bundle.rs`), not a bare `FFI_LogicalExtensionCodec` / `FFI_PhysicalExtensionCodec`, whenever the trait moves `Expr`s / `LogicalPlan`s / `ExecutionPlan`s across the boundary **or** exports an `FFI_SessionRef`. The bundle pairs one `FFI_TaskContextProvider` with both codecs so the three cannot drift apart, and a wrapper that only serializes logical data still needs the physical codec for any planner a consumer reaches through the session it exports. Forward the bundle unchanged to every nested `FFI_X` the wrapper creates, including in `clone_fn_wrapper`. Never store a bundle inside a codec: cloning a bundle clones its logical codec, so that would recurse forever — the dependency direction is bundle → codecs → task context provider.
 - **Method function pointers are private by default.** Mark `pub` only if a downstream library needs to invoke them directly (rare — typically only `version`, `library_marker_id`, embedded codecs are `pub`).
 - **`version: super::version` is mandatory.** Consumers gate compatibility on it.
 - **`library_marker_id: crate::get_library_marker_id` is mandatory *when the wrapper uses the standard `ForeignX` adapter pattern*.** Two flavors exist:
@@ -126,19 +126,14 @@ impl Clone for FFI_X { fn clone(&self) -> Self { unsafe { (self.clone)(self) } }
 
 `release` must null `private_data` so a double-free debug-asserts loudly.
 
-### 5. Constructor split
+### 5. Constructor
+
+One constructor per wrapper, taking the bundle:
 
 ```rust
 impl FFI_X {
     pub fn new(inner: Arc<dyn X>, runtime: Option<Handle>,
-               task_ctx_provider: impl Into<FFI_TaskContextProvider>,
-               logical_codec: Option<Arc<dyn LogicalExtensionCodec>>) -> Self {
-        // build FFI_LogicalExtensionCodec from defaults, then forward
-        Self::new_with_ffi_codec(inner, runtime, ffi_codec)
-    }
-
-    pub fn new_with_ffi_codec(inner: Arc<dyn X>, runtime: Option<Handle>,
-                              logical_codec: FFI_LogicalExtensionCodec) -> Self {
+               codecs: FFI_ExtensionCodecBundle) -> Self {
         // Round-trip downcast: if inner is already a ForeignX, return its FFI directly.
         if let Some(foreign) = inner.downcast_ref::<ForeignX>() {
             return foreign.0.clone();
@@ -150,6 +145,8 @@ impl FFI_X {
 
 The round-trip downcast is **mandatory** — without it, repeated FFI hops nest `ForeignX(FFI_X(ForeignX(...)))` and you pay the boundary cost every layer.
 
+Do **not** add a second constructor that builds the bundle from native codecs plus an `Option<Arc<dyn LogicalExtensionCodec>>`. That shape makes "use the default physical codec" implicit, which is the failure `FFI_ExtensionCodecBundle` exists to prevent. Callers with native codecs build the bundle themselves via `FFI_ExtensionCodecBundle::new`, or state the defaults explicitly via `FFI_ExtensionCodecBundle::new_default`.
+
 ### 6. The `Foreign<X>` consumer
 
 ```rust
@@ -356,5 +353,5 @@ When reviewing a PR that touches `datafusion/ffi/`:
 - Canonical wrapper to model after: `src/catalog_provider.rs`. Async + capability-flag variants: `src/table_provider.rs`.
 - Mutable-trait variant: `src/udaf/accumulator.rs` (`Box<dyn Accumulator>`).
 - Optional-method pattern: `FFI_TableProvider::supports_filters_pushdown`.
-- Codec wiring: `src/proto/logical_extension_codec.rs`, `src/proto/physical_extension_codec.rs`.
+- Codec wiring: `src/proto/extension_codec_bundle.rs` (the pairing every wrapper carries), `src/proto/logical_extension_codec.rs`, `src/proto/physical_extension_codec.rs`.
 - Examples crate: `datafusion-examples/examples/ffi` (end-to-end producer + consumer).

datafusion-examples/examples/ffi/ffi_example_table_provider/src/lib.rs

@@ -21,7 +21,7 @@ use arrow::array::{RecordBatch, record_batch};
 use arrow::datatypes as arrow_schema;
 use arrow::datatypes::{DataType, Field, Schema};
 use datafusion::datasource::MemTable;
-use datafusion_ffi::proto::logical_extension_codec::FFI_LogicalExtensionCodec;
+use datafusion_ffi::proto::extension_codec_bundle::FFI_ExtensionCodecBundle;
 use datafusion_ffi::table_provider::FFI_TableProvider;
 use ffi_module_interface::TableProviderModule;
 
@@ -36,7 +36,7 @@ fn create_record_batch(start_value: i32, num_values: usize) -> RecordBatch {
 /// Here we only wish to create a simple table provider as an example.
 /// We create an in-memory table and convert it to it's FFI counterpart.
 extern "C" fn construct_simple_table_provider(
-    codec: FFI_LogicalExtensionCodec,
+    codecs: FFI_ExtensionCodecBundle,
 ) -> FFI_TableProvider {
     let schema = Arc::new(Schema::new(vec![
         Field::new("a", DataType::Int32, true),
@@ -53,7 +53,7 @@ extern "C" fn construct_simple_table_provider(
 
     let table_provider = MemTable::try_new(schema, vec![batches]).unwrap();
 
-    FFI_TableProvider::new_with_ffi_codec(Arc::new(table_provider), true, None, codec)
+    FFI_TableProvider::new(Arc::new(table_provider), true, None, codecs)
 }
 
 #[unsafe(no_mangle)]

datafusion-examples/examples/ffi/ffi_module_interface/src/lib.rs

@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use datafusion_ffi::proto::logical_extension_codec::FFI_LogicalExtensionCodec;
+use datafusion_ffi::proto::extension_codec_bundle::FFI_ExtensionCodecBundle;
 use datafusion_ffi::table_provider::FFI_TableProvider;
 
 /// This struct defines the module interfaces. It is to be shared by
@@ -27,5 +27,5 @@ use datafusion_ffi::table_provider::FFI_TableProvider;
 pub struct TableProviderModule {
     /// Constructs the table provider
     pub create_table:
-        extern "C" fn(codec: FFI_LogicalExtensionCodec) -> FFI_TableProvider,
+        extern "C" fn(codecs: FFI_ExtensionCodecBundle) -> FFI_TableProvider,
 }

datafusion-examples/examples/ffi/ffi_module_loader/src/main.rs

@@ -23,7 +23,7 @@ use datafusion::{
     execution::TaskContextProvider,
     prelude::SessionContext,
 };
-use datafusion_ffi::proto::logical_extension_codec::FFI_LogicalExtensionCodec;
+use datafusion_ffi::proto::extension_codec_bundle::FFI_ExtensionCodecBundle;
 use ffi_module_interface::TableProviderModule;
 
 #[tokio::main]
@@ -67,13 +67,17 @@ async fn main() -> Result<()> {
     let table_provider_module = get_module();
 
     let ctx = Arc::new(SessionContext::new());
-    let codec = FFI_LogicalExtensionCodec::new_default(
+    // The bundle pairs the task context provider with the logical and physical
+    // extension codecs the module should serialize with. This example moves no
+    // custom extension nodes, so both codecs are the defaults.
+    let codecs = FFI_ExtensionCodecBundle::new_default(
         &(Arc::clone(&ctx) as Arc<dyn TaskContextProvider>),
+        None,
     );
 
     // By calling the code below, the table provided will be created within
     // the module's code.
-    let ffi_table_provider = (table_provider_module.create_table)(codec);
+    let ffi_table_provider = (table_provider_module.create_table)(codecs);
 
     // In order to access the table provider within this executable, we need to
     // turn it into a `TableProvider`.

datafusion/datasource-parquet/src/file_format.rs

@@ -297,6 +297,7 @@ async fn get_file_decryption_properties(
 }
 
 #[cfg(not(feature = "parquet_encryption"))]
+#[expect(clippy::unused_async)]
 async fn get_file_decryption_properties(
     _state: &dyn Session,
     _options: &TableParquetOptions,

datafusion/datasource-parquet/src/opener/encryption.rs

@@ -76,6 +76,7 @@ impl EncryptionContext {
 
 #[cfg(not(feature = "parquet_encryption"))]
 #[expect(dead_code)]
+#[expect(clippy::unused_async)]
 impl EncryptionContext {
     pub(super) async fn get_file_decryption_properties(
         &self,

datafusion/datasource-parquet/src/sink.rs

@@ -239,6 +239,7 @@ async fn set_writer_encryption_properties(
 }
 
 #[cfg(not(feature = "parquet_encryption"))]
+#[expect(clippy::unused_async)]
 async fn set_writer_encryption_properties(
     builder: WriterPropertiesBuilder,
     _runtime: &Arc<RuntimeEnv>,

datafusion/ffi/README.md

@@ -218,6 +218,48 @@ these methods that your provider remains valid for the lifetime of the
 calls. The `FFI_TaskContextProvider` is implemented on `SessionContext`
 and it is easy to implement on any struct that implements `Session`.
 
+## Extension Codec Bundle
+
+Serializing plans across the boundary needs three values that have to agree
+with one another: the `FFI_TaskContextProvider` above, a logical extension
+codec, and a physical extension codec. `FFI_ExtensionCodecBundle` carries
+them as a single unit, and every wrapper that serializes a plan or expression
+takes one:
+
+```rust,ignore
+let codecs = FFI_ExtensionCodecBundle::new(
+    &task_ctx_provider,
+    None, // Option<tokio::runtime::Handle>
+    Arc::new(MyLogicalCodec),
+    Arc::new(MyPhysicalCodec),
+);
+
+let ffi_provider = FFI_TableProvider::new(provider, true, None, codecs.clone());
+let ffi_catalog = FFI_CatalogProvider::new(catalog, None, codecs);
+```
+
+Use `FFI_ExtensionCodecBundle::new_default` when no custom extension nodes
+cross the boundary; it selects the default logical and physical codecs
+explicitly.
+
+Both codecs are needed even by a wrapper that only serializes logical data,
+because such a wrapper still exports a `Session` and a consumer can reach
+`Session::query_planner` through it. That planner serializes physical plans,
+so it needs the physical codec belonging to the same environment.
+
+The bundle is propagated unchanged through nested construction — a catalog
+provider list hands it to each catalog, which hands it to each schema, which
+hands it to each table provider — so a table found by walking the hierarchy
+serializes exactly like the list it came from.
+
+One case does not get a full bundle: a table provider decoded out of a
+serialized logical plan is reconstructed by `FFI_LogicalExtensionCodec`, which
+holds no physical codec, so it is paired with the default one. Such a provider
+cannot carry custom physical extension nodes if the decoding library scans it
+with a session local to that library. Scanning it with the exporting library's
+own session handle is unaffected, because exporting an already-foreign session
+returns the original handle and its complete bundle.
+
 [apache datafusion]: https://datafusion.apache.org/
 [api docs]: http://docs.rs/datafusion-ffi/latest
 [rust abi]: https://doc.rust-lang.org/reference/abi.html

datafusion/ffi/src/catalog_provider.rs

@@ -20,15 +20,11 @@ use std::sync::Arc;
 
 use datafusion_catalog::{CatalogProvider, SchemaProvider};
 use datafusion_common::error::Result;
-use datafusion_proto::logical_plan::{
-    DefaultLogicalExtensionCodec, LogicalExtensionCodec,
-};
 use stabby::string::String as SString;
 use stabby::vec::Vec as SVec;
 use tokio::runtime::Handle;
 
-use crate::execution::FFI_TaskContextProvider;
-use crate::proto::logical_extension_codec::FFI_LogicalExtensionCodec;
+use crate::proto::extension_codec_bundle::FFI_ExtensionCodecBundle;
 use crate::schema_provider::{FFI_SchemaProvider, ForeignSchemaProvider};
 use crate::util::{FFI_Option, FFI_Result};
 use crate::{df_result, sresult_return};
@@ -58,7 +54,9 @@ pub struct FFI_CatalogProvider {
             cascade: bool,
         ) -> FFI_Result<FFI_Option<FFI_SchemaProvider>>,
 
-    pub logical_codec: FFI_LogicalExtensionCodec,
+    /// The serialization environment propagated to every schema reached through
+    /// this catalog.
+    pub codecs: FFI_ExtensionCodecBundle,
 
     /// Used to create a clone on the provider of the execution plan. This should
     /// only need to be called by the receiver of the plan.
@@ -121,10 +119,10 @@ unsafe extern "C" fn schema_fn_wrapper(
         let maybe_schema = provider.inner().schema(name.as_str());
         maybe_schema
             .map(|schema| {
-                FFI_SchemaProvider::new_with_ffi_codec(
+                FFI_SchemaProvider::new(
                     schema,
                     provider.runtime(),
-                    provider.logical_codec.clone(),
+                    provider.codecs.clone(),
                 )
             })
             .into()
@@ -144,11 +142,7 @@ unsafe extern "C" fn register_schema_fn_wrapper(
         let returned_schema =
             sresult_return!(inner_provider.register_schema(name.as_str(), schema))
                 .map(|schema| {
-                    FFI_SchemaProvider::new_with_ffi_codec(
-                        schema,
-                        runtime,
-                        provider.logical_codec.clone(),
-                    )
+                    FFI_SchemaProvider::new(schema, runtime, provider.codecs.clone())
                 })
                 .into();
 
@@ -171,11 +165,7 @@ unsafe extern "C" fn deregister_schema_fn_wrapper(
         FFI_Result::Ok(
             maybe_schema
                 .map(|schema| {
-                    FFI_SchemaProvider::new_with_ffi_codec(
-                        schema,
-                        runtime,
-                        provider.logical_codec.clone(),
-                    )
+                    FFI_SchemaProvider::new(schema, runtime, provider.codecs.clone())
                 })
                 .into(),
         )
@@ -209,7 +199,7 @@ unsafe extern "C" fn clone_fn_wrapper(
             schema: schema_fn_wrapper,
             register_schema: register_schema_fn_wrapper,
             deregister_schema: deregister_schema_fn_wrapper,
-            logical_codec: provider.logical_codec.clone(),
+            codecs: provider.codecs.clone(),
             clone: clone_fn_wrapper,
             release: release_fn_wrapper,
             version: super::version,
@@ -227,27 +217,13 @@ impl Drop for FFI_CatalogProvider {
 
 impl FFI_CatalogProvider {
     /// Creates a new [`FFI_CatalogProvider`].
+    ///
+    /// `codecs` must describe the extension nodes used by every table below this
+    /// catalog, since schemas and tables reached through it inherit it.
     pub fn new(
         provider: Arc<dyn CatalogProvider>,
         runtime: Option<Handle>,
-        task_ctx_provider: impl Into<FFI_TaskContextProvider>,
-        logical_codec: Option<Arc<dyn LogicalExtensionCodec>>,
-    ) -> Self {
-        let task_ctx_provider = task_ctx_provider.into();
-        let logical_codec =
-            logical_codec.unwrap_or_else(|| Arc::new(DefaultLogicalExtensionCodec {}));
-        let logical_codec = FFI_LogicalExtensionCodec::new(
-            logical_codec,
-            runtime.clone(),
-            task_ctx_provider.clone(),
-        );
-        Self::new_with_ffi_codec(provider, runtime, logical_codec)
-    }
-
-    pub fn new_with_ffi_codec(
-        provider: Arc<dyn CatalogProvider>,
-        runtime: Option<Handle>,
-        logical_codec: FFI_LogicalExtensionCodec,
+        codecs: FFI_ExtensionCodecBundle,
     ) -> Self {
         if let Some(provider) = provider.downcast_ref::<ForeignCatalogProvider>() {
             return provider.0.clone();
@@ -260,7 +236,7 @@ impl FFI_CatalogProvider {
             schema: schema_fn_wrapper,
             register_schema: register_schema_fn_wrapper,
             deregister_schema: deregister_schema_fn_wrapper,
-            logical_codec,
+            codecs,
             clone: clone_fn_wrapper,
             release: release_fn_wrapper,
             version: super::version,
@@ -325,11 +301,7 @@ impl CatalogProvider for ForeignCatalogProvider {
         unsafe {
             let schema = match schema.downcast_ref::<ForeignSchemaProvider>() {
                 Some(s) => &s.0,
-                None => &FFI_SchemaProvider::new_with_ffi_codec(
-                    schema,
-                    None,
-                    self.0.logical_codec.clone(),
-                ),
+                None => &FFI_SchemaProvider::new(schema, None, self.0.codecs.clone()),
             };
             let returned_schema: Option<FFI_SchemaProvider> =
                 df_result!((self.0.register_schema)(&self.0, name.into(), schema))?
@@ -375,9 +347,9 @@ mod tests {
                 .is_none()
         );
         let (_ctx, task_ctx_provider) = crate::util::tests::test_session_and_ctx();
+        let codecs = FFI_ExtensionCodecBundle::new_default(task_ctx_provider, None);
 
-        let mut ffi_catalog =
-            FFI_CatalogProvider::new(catalog, None, task_ctx_provider, None);
+        let mut ffi_catalog = FFI_CatalogProvider::new(catalog, None, codecs);
         ffi_catalog.library_marker_id = crate::mock_foreign_marker_id;
 
         let foreign_catalog: Arc<dyn CatalogProvider> = (&ffi_catalog).into();
@@ -421,8 +393,8 @@ mod tests {
         let catalog = Arc::new(MemoryCatalogProvider::new());
 
         let (_ctx, task_ctx_provider) = crate::util::tests::test_session_and_ctx();
-        let mut ffi_catalog =
-            FFI_CatalogProvider::new(catalog, None, task_ctx_provider, None);
+        let codecs = FFI_ExtensionCodecBundle::new_default(task_ctx_provider, None);
+        let mut ffi_catalog = FFI_CatalogProvider::new(catalog, None, codecs);
 
         // Verify local libraries can be downcast to their original
         let foreign_catalog: Arc<dyn CatalogProvider> = (&ffi_catalog).into();