implement_custom_stabby_friendly_ffi

Author: coderfender

Cargo.lock

@@ -22,7 +22,6 @@ edition = { workspace = true }
 publish = false
 
 [dependencies]
-abi_stable = "0.11.3"
 arrow = { workspace = true }
 datafusion = { workspace = true }
 datafusion-ffi = { workspace = true }

datafusion-examples/examples/ffi/ffi_example_table_provider/Cargo.toml

@@ -17,13 +17,12 @@
 
 use std::sync::Arc;
 
-use abi_stable::{export_root_module, prefix_type::PrefixTypeTrait};
 use arrow::array::RecordBatch;
 use arrow::datatypes::{DataType, Field, Schema};
 use datafusion::{common::record_batch, datasource::MemTable};
 use datafusion_ffi::proto::logical_extension_codec::FFI_LogicalExtensionCodec;
 use datafusion_ffi::table_provider::FFI_TableProvider;
-use ffi_module_interface::{TableProviderModule, TableProviderModuleRef};
+use ffi_module_interface::TableProviderModule;
 
 fn create_record_batch(start_value: i32, num_values: usize) -> RecordBatch {
     let end_value = start_value + num_values as i32;
@@ -56,11 +55,10 @@ extern "C" fn construct_simple_table_provider(
     FFI_TableProvider::new_with_ffi_codec(Arc::new(table_provider), true, None, codec)
 }
 
-#[export_root_module]
+#[unsafe(no_mangle)]
 /// This defines the entry point for using the module.
-pub fn get_simple_memory_table() -> TableProviderModuleRef {
+pub unsafe extern "C" fn create_table_provider_module() -> TableProviderModule {
     TableProviderModule {
         create_table: construct_simple_table_provider,
     }
-    .leak_into_prefix()
 }

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

@@ -25,5 +25,4 @@ publish = false
 workspace = true
 
 [dependencies]
-abi_stable = "0.11.3"
 datafusion-ffi = { workspace = true }

datafusion-examples/examples/ffi/ffi_module_interface/Cargo.toml

@@ -15,18 +15,10 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use abi_stable::{
-    StableAbi, declare_root_module_statics,
-    library::{LibraryError, RootModule},
-    package_version_strings,
-    sabi_types::VersionStrings,
-};
 use datafusion_ffi::proto::logical_extension_codec::FFI_LogicalExtensionCodec;
 use datafusion_ffi::table_provider::FFI_TableProvider;
 
 #[repr(C)]
-#[derive(StableAbi)]
-#[sabi(kind(Prefix(prefix_ref = TableProviderModuleRef)))]
 /// This struct defines the module interfaces. It is to be shared by
 /// both the module loading program and library that implements the
 /// module. It is possible to move this definition into the loading
@@ -38,13 +30,10 @@ pub struct TableProviderModule {
         extern "C" fn(codec: FFI_LogicalExtensionCodec) -> FFI_TableProvider,
 }
 
-impl RootModule for TableProviderModuleRef {
-    declare_root_module_statics! {TableProviderModuleRef}
-    const BASE_NAME: &'static str = "ffi_example_table_provider";
-    const NAME: &'static str = "ffi_example_table_provider";
-    const VERSION_STRINGS: VersionStrings = package_version_strings!();
+/// # Safety
+/// This function is intended to be called from a dynamically loaded library.
+/// The caller must ensure proper initialization.
+pub type CreateTableProviderModuleFn = unsafe extern "C" fn() -> TableProviderModule;
 
-    fn initialization(self) -> Result<Self, LibraryError> {
-        Ok(self)
-    }
-}
+/// The symbol name that the module loader will look for when loading the library.
+pub const CREATE_TABLE_PROVIDER_MODULE_SYMBOL: &str = "create_table_provider_module";

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

@@ -24,16 +24,14 @@ use datafusion_common::error::Result;
 use datafusion_proto::logical_plan::{
     DefaultLogicalExtensionCodec, LogicalExtensionCodec,
 };
-use stabby::option::Option as StabbyOption;
-use stabby::result::Result as StabbyResult;
 use stabby::string::String as StabbyString;
 use stabby::vec::Vec as StabbyVec;
 use tokio::runtime::Handle;
 
 use crate::execution::FFI_TaskContextProvider;
 use crate::proto::logical_extension_codec::FFI_LogicalExtensionCodec;
 use crate::schema_provider::{FFI_SchemaProvider, ForeignSchemaProvider};
-use crate::util::FFIResult;
+use crate::util::{FFIResult, FfiOption, FfiResult};
 use crate::{df_result, rresult_return};
 
 /// A stable struct for sharing [`CatalogProvider`] across FFI boundaries.
@@ -45,21 +43,21 @@ pub struct FFI_CatalogProvider {
     pub schema: unsafe extern "C" fn(
         provider: &Self,
         name: StabbyString,
-    ) -> StabbyOption<FFI_SchemaProvider>,
+    ) -> FfiOption<FFI_SchemaProvider>,
 
     pub register_schema:
         unsafe extern "C" fn(
             provider: &Self,
             name: StabbyString,
             schema: &FFI_SchemaProvider,
-        ) -> FFIResult<StabbyOption<FFI_SchemaProvider>>,
+        ) -> FFIResult<FfiOption<FFI_SchemaProvider>>,
 
     pub deregister_schema:
         unsafe extern "C" fn(
             provider: &Self,
             name: StabbyString,
             cascade: bool,
-        ) -> FFIResult<StabbyOption<FFI_SchemaProvider>>,
+        ) -> FFIResult<FfiOption<FFI_SchemaProvider>>,
 
     pub logical_codec: FFI_LogicalExtensionCodec,
 
@@ -119,7 +117,7 @@ unsafe extern "C" fn schema_names_fn_wrapper(
 unsafe extern "C" fn schema_fn_wrapper(
     provider: &FFI_CatalogProvider,
     name: StabbyString,
-) -> StabbyOption<FFI_SchemaProvider> {
+) -> FfiOption<FFI_SchemaProvider> {
     unsafe {
         let maybe_schema = provider.inner().schema(name.as_str());
         maybe_schema
@@ -138,7 +136,7 @@ unsafe extern "C" fn register_schema_fn_wrapper(
     provider: &FFI_CatalogProvider,
     name: StabbyString,
     schema: &FFI_SchemaProvider,
-) -> FFIResult<StabbyOption<FFI_SchemaProvider>> {
+) -> FFIResult<FfiOption<FFI_SchemaProvider>> {
     unsafe {
         let runtime = provider.runtime();
         let inner_provider = provider.inner();
@@ -155,23 +153,23 @@ unsafe extern "C" fn register_schema_fn_wrapper(
                 })
                 .into();
 
-        StabbyResult::Ok(returned_schema)
+        FfiResult::Ok(returned_schema)
     }
 }
 
 unsafe extern "C" fn deregister_schema_fn_wrapper(
     provider: &FFI_CatalogProvider,
     name: StabbyString,
     cascade: bool,
-) -> FFIResult<StabbyOption<FFI_SchemaProvider>> {
+) -> FFIResult<FfiOption<FFI_SchemaProvider>> {
     unsafe {
         let runtime = provider.runtime();
         let inner_provider = provider.inner();
 
         let maybe_schema =
             rresult_return!(inner_provider.deregister_schema(name.as_str(), cascade));
 
-        StabbyResult::Ok(
+        FfiResult::Ok(
             maybe_schema
                 .map(|schema| {
                     FFI_SchemaProvider::new_with_ffi_codec(

datafusion/ffi/src/catalog_provider.rs

@@ -23,14 +23,14 @@ use datafusion_catalog::{CatalogProvider, CatalogProviderList};
 use datafusion_proto::logical_plan::{
     DefaultLogicalExtensionCodec, LogicalExtensionCodec,
 };
-use stabby::option::Option as StabbyOption;
 use stabby::string::String as StabbyString;
 use stabby::vec::Vec as StabbyVec;
 use tokio::runtime::Handle;
 
 use crate::catalog_provider::{FFI_CatalogProvider, ForeignCatalogProvider};
 use crate::execution::FFI_TaskContextProvider;
 use crate::proto::logical_extension_codec::FFI_LogicalExtensionCodec;
+use crate::util::FfiOption;
 
 /// A stable struct for sharing [`CatalogProviderList`] across FFI boundaries.
 #[repr(C)]
@@ -41,7 +41,7 @@ pub struct FFI_CatalogProviderList {
         &Self,
         name: StabbyString,
         catalog: &FFI_CatalogProvider,
-    ) -> StabbyOption<FFI_CatalogProvider>,
+    ) -> FfiOption<FFI_CatalogProvider>,
 
     /// List of existing catalogs
     pub catalog_names: unsafe extern "C" fn(&Self) -> StabbyVec<StabbyString>,
@@ -50,7 +50,7 @@ pub struct FFI_CatalogProviderList {
     pub catalog: unsafe extern "C" fn(
         &Self,
         name: StabbyString,
-    ) -> StabbyOption<FFI_CatalogProvider>,
+    ) -> FfiOption<FFI_CatalogProvider>,
 
     pub logical_codec: FFI_LogicalExtensionCodec,
 
@@ -111,7 +111,7 @@ unsafe extern "C" fn register_catalog_fn_wrapper(
     provider: &FFI_CatalogProviderList,
     name: StabbyString,
     catalog: &FFI_CatalogProvider,
-) -> StabbyOption<FFI_CatalogProvider> {
+) -> FfiOption<FFI_CatalogProvider> {
     unsafe {
         let runtime = provider.runtime();
         let inner_provider = provider.inner();
@@ -133,7 +133,7 @@ unsafe extern "C" fn register_catalog_fn_wrapper(
 unsafe extern "C" fn catalog_fn_wrapper(
     provider: &FFI_CatalogProviderList,
     name: StabbyString,
-) -> StabbyOption<FFI_CatalogProvider> {
+) -> FfiOption<FFI_CatalogProvider> {
     unsafe {
         let runtime = provider.runtime();
         let inner_provider = provider.inner();

datafusion/ffi/src/catalog_provider_list.rs

@@ -21,12 +21,13 @@ use std::ffi::c_void;
 
 use datafusion_common::config::{ConfigEntry, ConfigExtension, ExtensionOptions};
 use datafusion_common::{Result, exec_err};
-use stabby::result::Result as StabbyResult;
+
 use stabby::str::Str as StabbyStr;
 use stabby::string::String as StabbyString;
 use stabby::vec::Vec as StabbyVec;
 
 use crate::df_result;
+use crate::util::{FFIResult, FfiResult};
 
 /// A stable struct for sharing [`ExtensionOptions`] across FFI boundaries.
 ///
@@ -50,7 +51,7 @@ pub struct FFI_ExtensionOptions {
         &mut Self,
         key: StabbyStr,
         value: StabbyStr,
-    ) -> StabbyResult<(), StabbyString>,
+    ) -> FFIResult<()>,
 
     /// Returns the [`ConfigEntry`] stored in this [`ExtensionOptions`]
     pub entries: unsafe extern "C" fn(&Self) -> StabbyVec<(StabbyString, StabbyString)>,
@@ -98,11 +99,11 @@ unsafe extern "C" fn set_fn_wrapper(
     options: &mut FFI_ExtensionOptions,
     key: StabbyStr,
     value: StabbyStr,
-) -> StabbyResult<(), StabbyString> {
+) -> FFIResult<()> {
     let _ = options
         .inner_mut()
         .insert(key.as_str().into(), value.as_str().into());
-    StabbyResult::Ok(())
+    FfiResult::Ok(())
 }
 
 unsafe extern "C" fn entries_fn_wrapper(

datafusion/ffi/src/config/extension_options.rs

@@ -24,14 +24,15 @@ use datafusion_execution::runtime_env::RuntimeEnv;
 use datafusion_expr::{
     AggregateUDF, AggregateUDFImpl, ScalarUDF, ScalarUDFImpl, WindowUDF, WindowUDFImpl,
 };
-use stabby::option::Option as StabbyOption;
+
 use stabby::string::String as StabbyString;
 use stabby::vec::Vec as StabbyVec;
 
 use crate::session::config::FFI_SessionConfig;
 use crate::udaf::FFI_AggregateUDF;
 use crate::udf::FFI_ScalarUDF;
 use crate::udwf::FFI_WindowUDF;
+use crate::util::FfiOption;
 
 /// A stable struct for sharing [`TaskContext`] across FFI boundaries.
 #[repr(C)]
@@ -41,7 +42,7 @@ pub struct FFI_TaskContext {
     pub session_id: unsafe extern "C" fn(&Self) -> StabbyString,
 
     /// Return the task ID.
-    pub task_id: unsafe extern "C" fn(&Self) -> StabbyOption<StabbyString>,
+    pub task_id: unsafe extern "C" fn(&Self) -> FfiOption<StabbyString>,
 
     /// Return the session configuration.
     pub session_config: unsafe extern "C" fn(&Self) -> FFI_SessionConfig,
@@ -93,7 +94,7 @@ unsafe extern "C" fn session_id_fn_wrapper(ctx: &FFI_TaskContext) -> StabbyStrin
 
 unsafe extern "C" fn task_id_fn_wrapper(
     ctx: &FFI_TaskContext,
-) -> StabbyOption<StabbyString> {
+) -> FfiOption<StabbyString> {
     unsafe {
         let ctx = ctx.inner();
         ctx.task_id().map(|s| s.as_str().into()).into()
@@ -202,7 +203,7 @@ impl From<FFI_TaskContext> for Arc<TaskContext> {
                     let udf = <Arc<dyn ScalarUDFImpl>>::from(&kv_pair.1);
 
                     (
-                        kv_pair.0.into_string(),
+                        kv_pair.0.to_string(),
                         Arc::new(ScalarUDF::new_from_shared_impl(udf)),
                     )
                 })
@@ -213,7 +214,7 @@ impl From<FFI_TaskContext> for Arc<TaskContext> {
                     let udaf = <Arc<dyn AggregateUDFImpl>>::from(&kv_pair.1);
 
                     (
-                        kv_pair.0.into_string(),
+                        kv_pair.0.to_string(),
                         Arc::new(AggregateUDF::new_from_shared_impl(udaf)),
                     )
                 })
@@ -224,7 +225,7 @@ impl From<FFI_TaskContext> for Arc<TaskContext> {
                     let udwf = <Arc<dyn WindowUDFImpl>>::from(&kv_pair.1);
 
                     (
-                        kv_pair.0.into_string(),
+                        kv_pair.0.to_string(),
                         Arc::new(WindowUDF::new_from_shared_impl(udwf)),
                     )
                 })

datafusion/ffi/src/execution/task_ctx.rs

@@ -102,13 +102,10 @@ unsafe extern "C" fn children_fn_wrapper(
         let plan = &(*private_data).plan;
         let runtime = &(*private_data).runtime;
 
-        let children: Vec<_> = plan
-            .children()
+        plan.children()
             .into_iter()
             .map(|child| FFI_ExecutionPlan::new(Arc::clone(child), runtime.clone()))
-            .collect();
-
-        children.into()
+            .collect()
     }
 }