rebase_main

Author: coderfender

Cargo.lock

@@ -73,6 +73,31 @@ with your library and then to connect it to DataFusion using this crate.
 Alternatively, you could use [bindgen] to interface directly to the [FFI] provided
 by this crate, but that is currently not supported.
 
+## Stabby Usage
+
+This crate uses [stabby] for ABI-stable types like `stabby::string::String` and
+`stabby::vec::Vec`. We chose stabby because [abi_stable] is no longer actively
+maintained.
+
+We intentionally use `#[repr(C)]` for our struct definitions instead of stabby's
+`#[stabby::stabby]` macro. The reason is that stabby's `IStable` trait (required
+by the macro) demands that all inner types also implement `IStable`. This creates
+challenges for our use case:
+
+1. **Arrow types**: Arrow's FFI types like `FFI_ArrowSchema` do not implement
+   `IStable`, and adding such implementations would be laborious and error-prone.
+
+2. **Self-referential function pointers**: Many of our FFI structs contain
+   function pointers that reference `&Self`, which complicates `IStable`
+   implementations.
+
+3. **FFI_Option and FFI_Result**: For similar reasons, we provide our own
+   `FFI_Option<T>` and `FFI_Result<T, E>` types using `#[repr(C, u8)]` instead
+   of stabby's `Option` and `Result`, which require inner types to be `IStable`.
+
+This hybrid approach gives us stabby's maintained, ABI-stable collection types
+while retaining flexibility for our complex FFI struct layouts.
+
 ## FFI Boundary
 
 We expect this crate to be used by both sides of the FFI Boundary. This should
@@ -198,6 +223,7 @@ and it is easy to implement on any struct that implements `Session`.
 [rust abi]: https://doc.rust-lang.org/reference/abi.html
 [ffi]: https://doc.rust-lang.org/nomicon/ffi.html
 [stabby]: https://crates.io/crates/stabby
+[abi_stable]: https://crates.io/crates/abi_stable
 [async-ffi]: https://crates.io/crates/async-ffi
 [bindgen]: https://crates.io/crates/bindgen
 [`datafusion-python`]: https://datafusion.apache.org/python/

datafusion/ffi/README.md

@@ -20,19 +20,6 @@ use std::collections::HashMap;
 use std::ffi::c_void;
 use std::sync::Arc;
 
-use crate::arrow_wrappers::WrappedSchema;
-use crate::execution::FFI_TaskContext;
-use crate::execution_plan::FFI_ExecutionPlan;
-use crate::physical_expr::FFI_PhysicalExpr;
-use crate::proto::logical_extension_codec::FFI_LogicalExtensionCodec;
-use crate::session::config::FFI_SessionConfig;
-use crate::udaf::FFI_AggregateUDF;
-use crate::udf::FFI_ScalarUDF;
-use crate::udwf::FFI_WindowUDF;
-use crate::util::FFIResult;
-use crate::{df_result, rresult, rresult_return};
-use abi_stable::StableAbi;
-use abi_stable::std_types::{RHashMap, RResult, RStr, RString, RVec};
 use arrow_schema::SchemaRef;
 use arrow_schema::ffi::FFI_ArrowSchema;
 use async_ffi::{FfiFuture, FutureExt};
@@ -63,6 +50,18 @@ use stabby::string::String as SString;
 use stabby::vec::Vec as SVec;
 use tokio::runtime::Handle;
 
+use crate::arrow_wrappers::WrappedSchema;
+use crate::execution::FFI_TaskContext;
+use crate::execution_plan::FFI_ExecutionPlan;
+use crate::physical_expr::FFI_PhysicalExpr;
+use crate::proto::logical_extension_codec::FFI_LogicalExtensionCodec;
+use crate::session::config::FFI_SessionConfig;
+use crate::udaf::FFI_AggregateUDF;
+use crate::udf::FFI_ScalarUDF;
+use crate::udwf::FFI_WindowUDF;
+use crate::util::{FFI_Result, FFIResult};
+use crate::{df_result, sresult, sresult_return};
+
 pub mod config;
 
 /// A stable struct for sharing [`Session`] across FFI boundaries.
@@ -639,9 +638,11 @@ impl Session for ForeignSession {
 
 #[cfg(test)]
 mod tests {
+    use std::sync::Arc;
 
     use arrow_schema::{DataType, Field, Schema};
     use datafusion::execution::SessionStateBuilder;
+    use datafusion_common::DataFusionError;
     use datafusion_expr::col;
     use datafusion_expr::registry::FunctionRegistry;
     use datafusion_proto::logical_plan::DefaultLogicalExtensionCodec;

datafusion/ffi/src/session/mod.rs

@@ -19,27 +19,23 @@ use std::any::Any;
 use std::ffi::c_void;
 use std::sync::Arc;
 
-use abi_stable::StableAbi;
-use abi_stable::std_types::{RResult, RVec};
-use datafusion_catalog::{TableFunctionArgs, TableFunctionImpl, TableProvider};
-use datafusion_common::DataFusionError;
+use datafusion_catalog::{TableFunctionImpl, TableProvider};
 use datafusion_common::error::Result;
 use datafusion_execution::TaskContext;
+use datafusion_expr::Expr;
 use datafusion_proto::logical_plan::from_proto::parse_exprs;
 use datafusion_proto::logical_plan::to_proto::serialize_exprs;
 use datafusion_proto::logical_plan::{
     DefaultLogicalExtensionCodec, LogicalExtensionCodec,
 };
 use datafusion_proto::protobuf::LogicalExprList;
-use datafusion_session::Session;
 use prost::Message;
 
 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::session::{FFI_SessionRef, ForeignSession};
 use crate::table_provider::FFI_TableProvider;
 use crate::util::{FFI_Result, FFIResult};
 use crate::{df_result, sresult_return};
@@ -48,22 +44,11 @@ use crate::{df_result, sresult_return};
 #[repr(C)]
 #[derive(Debug)]
 pub struct FFI_TableFunction {
-    /// Equivalent to the [`TableFunctionImpl::call`].
+    /// Equivalent to the `call` function of the TableFunctionImpl.
     /// The arguments are Expr passed as protobuf encoded bytes.
-    #[deprecated(
-        since = "53.0.0",
-        note = "See TableFunctionImpl::call deprecation note"
-    )]
     pub call:
         unsafe extern "C" fn(udtf: &Self, args: SVec<u8>) -> FFIResult<FFI_TableProvider>,
 
-    /// Equivalent to the [`TableFunctionImpl::call_with_args`].
-    call_with_args: unsafe extern "C" fn(
-        udtf: &Self,
-        args: RVec<u8>,
-        session: FFI_SessionRef,
-    ) -> FFIResult<FFI_TableProvider>,
-
     pub logical_codec: FFI_LogicalExtensionCodec,
 
     /// Used to create a clone on the provider of the udtf. This should
@@ -131,47 +116,6 @@ unsafe extern "C" fn call_fn_wrapper(
     ))
 }
 
-unsafe extern "C" fn call_with_args_wrapper(
-    udtf: &FFI_TableFunction,
-    args: RVec<u8>,
-    session: FFI_SessionRef,
-) -> FFIResult<FFI_TableProvider> {
-    let runtime = udtf.runtime();
-    let udtf_inner = udtf.inner();
-
-    let ctx: Arc<TaskContext> =
-        rresult_return!((&udtf.logical_codec.task_ctx_provider).try_into());
-    let codec: Arc<dyn LogicalExtensionCodec> = (&udtf.logical_codec).into();
-
-    let proto_filters = rresult_return!(LogicalExprList::decode(args.as_ref()));
-
-    let args = rresult_return!(parse_exprs(
-        proto_filters.expr.iter(),
-        ctx.as_ref(),
-        codec.as_ref()
-    ));
-
-    let mut foreign_session = None;
-    let session = rresult_return!(
-        session
-            .as_local()
-            .map(Ok::<&(dyn Session + Send + Sync), DataFusionError>)
-            .unwrap_or_else(|| {
-                foreign_session = Some(ForeignSession::try_from(&session)?);
-                Ok(foreign_session.as_ref().unwrap())
-            })
-    );
-    let table_provider = rresult_return!(
-        udtf_inner.call_with_args(TableFunctionArgs::new(&args, session))
-    );
-    RResult::ROk(FFI_TableProvider::new_with_ffi_codec(
-        table_provider,
-        false,
-        runtime,
-        udtf.logical_codec.clone(),
-    ))
-}
-
 unsafe extern "C" fn release_fn_wrapper(udtf: &mut FFI_TableFunction) {
     unsafe {
         debug_assert!(!udtf.private_data.is_null());
@@ -232,9 +176,7 @@ impl FFI_TableFunction {
         let private_data = Box::new(TableFunctionPrivateData { udtf, runtime });
 
         Self {
-            #[expect(deprecated)]
             call: call_fn_wrapper,
-            call_with_args: call_with_args_wrapper,
             logical_codec,
             clone: clone_fn_wrapper,
             release: release_fn_wrapper,
@@ -273,35 +215,13 @@ impl From<FFI_TableFunction> for Arc<dyn TableFunctionImpl> {
 }
 
 impl TableFunctionImpl for ForeignTableFunction {
-    fn call_with_args(&self, args: TableFunctionArgs) -> Result<Arc<dyn TableProvider>> {
-        let session = FFI_SessionRef::new(
-            args.session(),
-            self.0.runtime(),
-            self.0.logical_codec.clone(),
-        );
-        let codec: Arc<dyn LogicalExtensionCodec> = (&self.0.logical_codec).into();
-        let expr_list = LogicalExprList {
-            expr: serialize_exprs(args.exprs(), codec.as_ref())?,
-        };
-        let filters_serialized = expr_list.encode_to_vec().into();
-
-        let table_provider =
-            unsafe { (self.0.call_with_args)(&self.0, filters_serialized, session) };
-
-        let table_provider = df_result!(table_provider)?;
-        let table_provider: Arc<dyn TableProvider> = (&table_provider).into();
-
-        Ok(table_provider)
-    }
-
-    fn call(&self, args: &[datafusion_expr::Expr]) -> Result<Arc<dyn TableProvider>> {
+    fn call(&self, args: &[Expr]) -> Result<Arc<dyn TableProvider>> {
         let codec: Arc<dyn LogicalExtensionCodec> = (&self.0.logical_codec).into();
         let expr_list = LogicalExprList {
             expr: serialize_exprs(args, codec.as_ref())?,
         };
         let filters_serialized = expr_list.encode_to_vec().into_iter().collect();
 
-        #[expect(deprecated)]
         let table_provider = unsafe { (self.0.call)(&self.0, filters_serialized) };
 
         let table_provider = df_result!(table_provider)?;
@@ -322,22 +242,16 @@ mod tests {
     use datafusion::logical_expr::ptr_eq::arc_ptr_eq;
     use datafusion::prelude::{SessionContext, lit};
     use datafusion::scalar::ScalarValue;
-    use datafusion_catalog::TableFunctionArgs;
     use datafusion_execution::TaskContextProvider;
-    use datafusion_expr::Expr;
 
     use super::*;
 
     #[derive(Debug)]
     struct TestUDTF {}
 
     impl TableFunctionImpl for TestUDTF {
-        fn call_with_args(
-            &self,
-            args: TableFunctionArgs,
-        ) -> Result<Arc<dyn TableProvider>> {
+        fn call(&self, args: &[Expr]) -> Result<Arc<dyn TableProvider>> {
             let args = args
-                .exprs()
                 .iter()
                 .map(|arg| {
                     if let Expr::Literal(scalar, _) = arg {
@@ -427,10 +341,7 @@ mod tests {
 
         let foreign_udf: Arc<dyn TableFunctionImpl> = local_udtf.into();
 
-        let table = foreign_udf.call_with_args(TableFunctionArgs::new(
-            &[lit(6_u64), lit("one"), lit(2.0), lit(3_u64)],
-            &ctx.state(),
-        ))?;
+        let table = foreign_udf.call(&[lit(6_u64), lit("one"), lit(2.0), lit(3_u64)])?;
 
         let _ = ctx.register_table("test-table", table)?;
 

datafusion/ffi/src/udtf.rs

@@ -24,6 +24,7 @@ mod tests {
     use std::sync::Arc;
 
     use datafusion::catalog::{CatalogProvider, CatalogProviderList};
+    use datafusion_common::DataFusionError;
     use datafusion_ffi::tests::utils::get_module;
 
     #[tokio::test]

datafusion/ffi/tests/ffi_catalog.rs

@@ -19,7 +19,7 @@
 /// when the feature integration-tests is built
 #[cfg(feature = "integration-tests")]
 mod tests {
-    use datafusion::error::Result;
+    use datafusion::error::{DataFusionError, Result};
     use datafusion_common::ScalarValue;
     use datafusion_common::config::{ConfigOptions, TableOptions};
     use datafusion_execution::config::SessionConfig;
@@ -31,7 +31,13 @@ mod tests {
     fn test_ffi_config_options_extension() -> Result<()> {
         let module = get_module()?;
 
-        let extension_options = (module.create_extension_options)();
+        let extension_options =
+            module
+                .create_extension_options()
+                .ok_or(DataFusionError::NotImplemented(
+                    "External test library failed to implement create_extension_options"
+                        .to_string(),
+                ))?();
 
         let mut config = ConfigOptions::new();
         config.extensions.insert(extension_options);
@@ -55,7 +61,13 @@ mod tests {
     fn test_ffi_table_options_extension() -> Result<()> {
         let module = get_module()?;
 
-        let extension_options = (module.create_extension_options)();
+        let extension_options =
+            module
+                .create_extension_options()
+                .ok_or(DataFusionError::NotImplemented(
+                    "External test library failed to implement create_extension_options"
+                        .to_string(),
+                ))?();
 
         let mut table_options = TableOptions::new();
         table_options.extensions.insert(extension_options);
@@ -80,7 +92,13 @@ mod tests {
     fn test_ffi_session_config_options_extension() -> Result<()> {
         let module = get_module()?;
 
-        let extension_options = (module.create_extension_options)();
+        let extension_options =
+            module
+                .create_extension_options()
+                .ok_or(DataFusionError::NotImplemented(
+                    "External test library failed to implement create_extension_options"
+                        .to_string(),
+                ))?();
 
         let mut config = SessionConfig::new().with_option_extension(extension_options);
 

datafusion/ffi/tests/ffi_config.rs

@@ -49,7 +49,12 @@ mod tests {
             Arc::new(EmptyExec::new(schema))
         }
 
-        let child_plan = (module.create_empty_exec)();
+        let child_plan =
+            module
+                .create_empty_exec()
+                .ok_or(DataFusionError::NotImplemented(
+                    "External module failed to implement create_empty_exec".to_string(),
+                ))?();
         let child_plan: Arc<dyn ExecutionPlan> = (&child_plan)
             .try_into()
             .expect("should be able create plan");

datafusion/ffi/tests/ffi_execution_plan.rs

@@ -26,7 +26,7 @@ mod tests {
 
     use arrow::datatypes::Schema;
     use datafusion::catalog::{TableProvider, TableProviderFactory};
-    use datafusion::error::Result;
+    use datafusion::error::{DataFusionError, Result};
     use datafusion_common::TableReference;
     use datafusion_common::ToDFSchema;
     use datafusion_expr::CreateExternalTable;
@@ -42,7 +42,11 @@ mod tests {
 
         // 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)(synchronous, codec);
+        let ffi_table_provider = table_provider_module.create_table().ok_or(
+            DataFusionError::NotImplemented(
+                "External table provider failed to implement create_table".to_string(),
+            ),
+        )?(synchronous, codec);
 
         // In order to access the table provider within this executable, we need to
         // turn it into a `TableProvider`.
@@ -76,8 +80,11 @@ mod tests {
         let table_provider_module = get_module()?;
         let (ctx, codec) = super::utils::ctx_and_codec();
 
-        let ffi_table_provider_factory =
-            (table_provider_module.create_table_factory)(codec);
+        let ffi_table_provider_factory = table_provider_module
+            .create_table_factory()
+            .ok_or(DataFusionError::NotImplemented(
+                "External table provider factory failed to implement create".to_string(),
+            ))?(codec);
 
         let foreign_table_provider_factory: Arc<dyn TableProviderFactory> =
             (&ffi_table_provider_factory).into();