wip

Author: timsaucer

datafusion/ffi/src/config_options.rs

@@ -16,17 +16,19 @@
 // under the License.
 
 use std::{any::Any, ffi::c_void, ops::Deref};
-
-use abi_stable::{std_types::{RHashMap, RResult, RStr, RString, RVec}, RTuple, StableAbi};
-use abi_stable::std_types::{ROption, Tuple3};
+use std::collections::HashMap;
+use abi_stable::{
+    std_types::{RHashMap, ROption, RResult, RStr, RString, RVec, Tuple3},
+    RTuple, StableAbi,
+};
 use arrow::{array::ArrayRef, error::ArrowError};
 use datafusion::{
     error::{DataFusionError, Result},
     scalar::ScalarValue,
 };
-use datafusion_common::config::{ConfigEntry, ExtensionOptions};
+use datafusion_common::config::{ConfigEntry, ConfigExtension, ExtensionOptions};
 use prost::Message;
-
+use datafusion_common::exec_err;
 use crate::{arrow_wrappers::WrappedArray, df_result, rresult, rresult_return};
 
 /// A stable struct for sharing [`ExtensionOptions`] across FFI boundaries.
@@ -38,10 +40,14 @@ use crate::{arrow_wrappers::WrappedArray, df_result, rresult, rresult_return};
 pub struct FFI_ExtensionOptions {
     pub cloned: unsafe extern "C" fn(&Self) -> FFI_ExtensionOptions,
 
-    pub set: unsafe extern "C" fn(&mut Self, key: RStr, value: RStr) -> RResult<(), RString>,
+    pub set:
+        unsafe extern "C" fn(&mut Self, key: RStr, value: RStr) -> RResult<(), RString>,
 
-    pub entries:
-        unsafe extern "C" fn(&Self) -> RVec<Tuple3<RString, ROption<RString>, RStr>>,
+    pub entries: unsafe extern "C" fn(
+        &Self,
+    ) -> RVec<
+        Tuple3<RString, ROption<RString>, RStr<'static>>,
+    >,
 
     /// Release the memory of the private data when it is no longer being used.
     pub release: unsafe extern "C" fn(&mut Self),
@@ -55,24 +61,26 @@ unsafe impl Send for FFI_ExtensionOptions {}
 unsafe impl Sync for FFI_ExtensionOptions {}
 
 pub struct ExtensionOptionsPrivateData {
-    pub options: Box<dyn ExtensionOptions>,
+    pub options: HashMap<String, String>,
 }
 
 impl FFI_ExtensionOptions {
     #[inline]
-    unsafe fn inner_mut(&mut self) -> &mut Box<dyn ExtensionOptions> {
+    unsafe fn inner_mut(&mut self) -> &mut HashMap<String, String> {
         let private_data = self.private_data as *mut ExtensionOptionsPrivateData;
         &mut (*private_data).options
     }
 
     #[inline]
-    unsafe fn inner(&self) -> &dyn ExtensionOptions {
+    unsafe fn inner(&self) -> &HashMap<String, String> {
         let private_data = self.private_data as *const ExtensionOptionsPrivateData;
-        (*private_data).options.deref()
+        &(*private_data).options
     }
 }
 
-unsafe extern "C" fn cloned_fn_wrapper(options: &FFI_ExtensionOptions) -> FFI_ExtensionOptions {
+unsafe extern "C" fn cloned_fn_wrapper(
+    options: &FFI_ExtensionOptions,
+) -> FFI_ExtensionOptions {
     options.inner().cloned().into()
 }
 
@@ -81,19 +89,24 @@ unsafe extern "C" fn set_fn_wrapper(
     key: RStr,
     value: RStr,
 ) -> RResult<(), RString> {
-    let key = key.as_str();
-    let value = value.as_str();
-
-    rresult!(options.inner_mut().set(key, value))
+    rresult!(options.inner_mut().set(key.into(), value.into()))
 }
 
 unsafe extern "C" fn entries_fn_wrapper(
     options: &FFI_ExtensionOptions,
-) -> RVec<Tuple3<RString, ROption<RString>, RStr>> {
-    options.inner()
+) -> RVec<Tuple3<RString, ROption<RString>, RStr<'static>>> {
+    options
+        .inner()
         .entries()
         .into_iter()
-        .map(|entry| (entry.key.into(), entry.value.map(Into::into).into(), entry.description.into()).into())
+        .map(|entry| {
+            (
+                entry.key.into(),
+                entry.value.map(Into::into).into(),
+                entry.description.into(),
+            )
+                .into()
+        })
         .collect()
 }
 
@@ -103,9 +116,9 @@ unsafe extern "C" fn release_fn_wrapper(options: &mut FFI_ExtensionOptions) {
     drop(private_data);
 }
 
-impl From<Box<dyn ExtensionOptions>> for FFI_ExtensionOptions {
-    fn from(options: Box<dyn ExtensionOptions>) -> Self {
-        let private_data = ExtensionOptionsPrivateData { options };
+impl Default for FFI_ExtensionOptions {
+     fn default() -> Self {
+        let private_data = ExtensionOptionsPrivateData { options: HashMap::new() };
 
         Self {
             cloned: cloned_fn_wrapper,
@@ -135,12 +148,34 @@ pub struct ForeignExtensionOptions(FFI_ExtensionOptions);
 unsafe impl Send for ForeignExtensionOptions {}
 unsafe impl Sync for ForeignExtensionOptions {}
 
-impl From<FFI_ExtensionOptions> for ForeignExtensionOptions {
-    fn from(options: FFI_ExtensionOptions) -> Self {
-        Self(options)
+impl<T: ConfigExtension + Default> TryFrom<FFI_ExtensionOptions> for T {
+    type Error = DataFusionError;
+
+    fn try_from(options: &FFI_ExtensionOptions) -> Result<Self, Self::Error> {
+        let mut config = T::default();
+
+        let mut found = false;
+        unsafe {
+            for entry_tuple in (options.entries)(&options)
+                .into_iter() {
+                if let ROption::RSome(value) = entry_tuple.1
+                && let Some((namespace, key)) = entry_tuple.0.as_str().split_once('.') {
+                    if namespace == T::PREFIX {
+                        found = true;
+                        config.set(key, value.as_str())?;
+                    }
+                }
+            }
+        }
+
+        Ok(config)
     }
 }
 
+impl ConfigExtension for ForeignExtensionOptions {
+    const PREFIX: &'static str = "datafusion_ffi";
+}
+
 impl ExtensionOptions for ForeignExtensionOptions {
     fn as_any(&self) -> &dyn Any {
         self
@@ -151,31 +186,44 @@ impl ExtensionOptions for ForeignExtensionOptions {
     }
 
     fn cloned(&self) -> Box<dyn ExtensionOptions> {
-        (self.0.cloned)(&self.0).into()
+        unsafe { (self.0.cloned)(&self.0).into() }
     }
 
     fn set(&mut self, key: &str, value: &str) -> Result<()> {
-        df_result!({
-            (self.0.set)(&mut self.0, key.into(), value.into())
-        })
+        let Some((namespace, key)) = key.split_once('.') else {
+            return exec_err!("Unable to set FFI config value without namespace set");
+        };
+
+        if namespace != ForeignExtensionOptions::PREFIX {
+            return exec_err!("Unexpected namespace {namespace} set for FFI config");
+        }
+
+        df_result!(unsafe { (self.0.set)(&mut self.0, key.into(), value.into()) })
     }
 
     fn entries(&self) -> Vec<ConfigEntry> {
-        (self.0.entries)(&self.0).into_iter()
-            .map(|entry_tuple| ConfigEntry {
-                key: entry_tuple.0.into(),
-                value: entry_tuple.1.into(),
-                description: entry_tuple.2.into(),
-            })
-            .collect()
+        unsafe {
+            (self.0.entries)(&self.0)
+                .into_iter()
+                .map(|entry_tuple| ConfigEntry {
+                    key: entry_tuple.0.into(),
+                    value: entry_tuple.1.map(Into::into).into(),
+                    description: entry_tuple.2.into(),
+                })
+                .collect()
+        }
     }
 }
 
 #[cfg(test)]
 mod tests {
     use datafusion_common::{
-        config::ConfigExtension, config::ConfigOptions, extensions_options,
+        config::{ConfigExtension, ConfigOptions, ExtensionOptions},
+        extensions_options,
     };
+
+    use crate::config_options::FFI_ExtensionOptions;
+
     // Define a new configuration struct using the `extensions_options` macro
     extensions_options! {
        /// My own config options.
@@ -194,10 +242,10 @@ mod tests {
 
     #[test]
     fn round_trip_ffi_extension_options() {
-
         // set up config struct and register extension
         let mut config = ConfigOptions::default();
-        config.extensions.insert(MyConfig::default());
+        config.extensions.insert(FFI_ExtensionOptions::default());
+        // config.extensions.insert(MyConfig::default());
 
         // overwrite config default
         config.set("my_config.baz_count", "42").unwrap();
@@ -207,5 +255,11 @@ mod tests {
         assert!(my_config.foo_to_bar,);
         assert_eq!(my_config.baz_count, 42,);
 
+        // let boxed_config = Box::new(MyConfig::default()) as Box<dyn ExtensionOptions>;
+        // let mut ffi_config = FFI_ExtensionOptions::from(boxed_config);
+        // ffi_config.library_marker_id = crate::mock_foreign_marker_id;
+        // let foreign_config: Box<dyn ExtensionOptions> = ffi_config.into();
+        //
+        // config.extensions.insert(foreign_config);
     }
 }