Add support for FFI config extensions (#19469)

## Which issue does this PR close?

This addresses part of #17035

This is also a blocker for
#20450

## Rationale for this change

Currently we cannot support user defined configuration extensions via
FFI. This is because much of the infrastructure on how to add and
extract custom extensions relies on knowing concrete types of the
extensions. This is not supported in FFI. This PR adds an implementation
of configuration extensions that can be used across a FFI boundary.

## What changes are included in this PR?

- Implement `FFI_ExtensionOptions`.
- Update `ConfigOptions` to check if a `datafusion_ffi` namespace exists
when setting values
- Add unit test

## Are these changes tested?

Unit test added.

Also tested against `datafusion-python` locally. With this code I have
the following test that passes. I have created a simple python exposed
`MyConfig`:

```python
from datafusion import SessionConfig
from datafusion_ffi_example import MyConfig

def test_catalog_provider():
    config = MyConfig()
    config = SessionConfig().with_extension(config)
    config.set("my_config.baz_count", "42")
```

## Are there any user-facing changes?

New addition only.

Author: timsaucer

datafusion/common/src/config.rs

@@ -1256,7 +1256,7 @@ impl<'a> TryInto<arrow::util::display::FormatOptions<'a>> for &'a FormatOptions
 }
 
 /// A key value pair, with a corresponding description
-#[derive(Debug, Hash, PartialEq, Eq)]
+#[derive(Debug, Clone, Hash, PartialEq, Eq)]
 pub struct ConfigEntry {
     /// A unique string to identify this config value
     pub key: String,
@@ -1352,6 +1352,10 @@ impl ConfigField for ConfigOptions {
     }
 }
 
+/// This namespace is reserved for interacting with Foreign Function Interface
+/// (FFI) based configuration extensions.
+pub const DATAFUSION_FFI_CONFIG_NAMESPACE: &str = "datafusion_ffi";
+
 impl ConfigOptions {
     /// Creates a new [`ConfigOptions`] with default values
     pub fn new() -> Self {
@@ -1366,12 +1370,12 @@ impl ConfigOptions {
 
     /// Set a configuration option
     pub fn set(&mut self, key: &str, value: &str) -> Result<()> {
-        let Some((prefix, key)) = key.split_once('.') else {
+        let Some((mut prefix, mut inner_key)) = key.split_once('.') else {
             return _config_err!("could not find config namespace for key \"{key}\"");
         };
 
         if prefix == "datafusion" {
-            if key == "optimizer.enable_dynamic_filter_pushdown" {
+            if inner_key == "optimizer.enable_dynamic_filter_pushdown" {
                 let bool_value = value.parse::<bool>().map_err(|e| {
                     DataFusionError::Configuration(format!(
                         "Failed to parse '{value}' as bool: {e}",
@@ -1386,13 +1390,23 @@ impl ConfigOptions {
                 }
                 return Ok(());
             }
-            return ConfigField::set(self, key, value);
+            return ConfigField::set(self, inner_key, value);
+        }
+
+        if !self.extensions.0.contains_key(prefix)
+            && self
+                .extensions
+                .0
+                .contains_key(DATAFUSION_FFI_CONFIG_NAMESPACE)
+        {
+            inner_key = key;
+            prefix = DATAFUSION_FFI_CONFIG_NAMESPACE;
         }
 
         let Some(e) = self.extensions.0.get_mut(prefix) else {
             return _config_err!("Could not find config namespace \"{prefix}\"");
         };
-        e.0.set(key, value)
+        e.0.set(inner_key, value)
     }
 
     /// Create new [`ConfigOptions`], taking values from environment variables
@@ -2157,7 +2171,7 @@ impl TableOptions {
     ///
     /// A result indicating success or failure in setting the configuration option.
     pub fn set(&mut self, key: &str, value: &str) -> Result<()> {
-        let Some((prefix, _)) = key.split_once('.') else {
+        let Some((mut prefix, _)) = key.split_once('.') else {
             return _config_err!("could not find config namespace for key \"{key}\"");
         };
 
@@ -2169,6 +2183,15 @@ impl TableOptions {
             return Ok(());
         }
 
+        if !self.extensions.0.contains_key(prefix)
+            && self
+                .extensions
+                .0
+                .contains_key(DATAFUSION_FFI_CONFIG_NAMESPACE)
+        {
+            prefix = DATAFUSION_FFI_CONFIG_NAMESPACE;
+        }
+
         let Some(e) = self.extensions.0.get_mut(prefix) else {
             return _config_err!("Could not find config namespace \"{prefix}\"");
         };

datafusion/ffi/src/config/extension_options.rs

@@ -0,0 +1,288 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::any::Any;
+use std::collections::HashMap;
+use std::ffi::c_void;
+
+use abi_stable::StableAbi;
+use abi_stable::std_types::{RResult, RStr, RString, RVec, Tuple2};
+use datafusion_common::config::{ConfigEntry, ConfigExtension, ExtensionOptions};
+use datafusion_common::{Result, exec_err};
+
+use crate::df_result;
+
+/// A stable struct for sharing [`ExtensionOptions`] across FFI boundaries.
+///
+/// Unlike other FFI structs in this crate, we do not construct a foreign
+/// variant of this object. This is due to the typical method for interacting
+/// with extension options is by creating a local struct of your concrete type.
+/// To support this methodology use the `to_extension` method instead.
+///
+/// When using [`FFI_ExtensionOptions`] with multiple extensions, all extension
+/// values are stored on a single [`FFI_ExtensionOptions`] object. The keys
+/// are stored with the full path prefix to avoid overwriting values when using
+/// multiple extensions.
+#[repr(C)]
+#[derive(Debug, StableAbi)]
+pub struct FFI_ExtensionOptions {
+    /// Return a deep clone of this [`ExtensionOptions`]
+    pub cloned: unsafe extern "C" fn(&Self) -> FFI_ExtensionOptions,
+
+    /// Set the given `key`, `value` pair
+    pub set:
+        unsafe extern "C" fn(&mut Self, key: RStr, value: RStr) -> RResult<(), RString>,
+
+    /// Returns the [`ConfigEntry`] stored in this [`ExtensionOptions`]
+    pub entries: unsafe extern "C" fn(&Self) -> RVec<Tuple2<RString, RString>>,
+
+    /// Release the memory of the private data when it is no longer being used.
+    pub release: unsafe extern "C" fn(&mut Self),
+
+    /// Internal data. This is only to be accessed by the provider of the options.
+    pub private_data: *mut c_void,
+}
+
+unsafe impl Send for FFI_ExtensionOptions {}
+unsafe impl Sync for FFI_ExtensionOptions {}
+
+pub struct ExtensionOptionsPrivateData {
+    pub options: HashMap<String, String>,
+}
+
+impl FFI_ExtensionOptions {
+    #[inline]
+    fn inner_mut(&mut self) -> &mut HashMap<String, String> {
+        let private_data = self.private_data as *mut ExtensionOptionsPrivateData;
+        unsafe { &mut (*private_data).options }
+    }
+
+    #[inline]
+    fn inner(&self) -> &HashMap<String, String> {
+        let private_data = self.private_data as *const ExtensionOptionsPrivateData;
+        unsafe { &(*private_data).options }
+    }
+}
+
+unsafe extern "C" fn cloned_fn_wrapper(
+    options: &FFI_ExtensionOptions,
+) -> FFI_ExtensionOptions {
+    options
+        .inner()
+        .iter()
+        .map(|(k, v)| (k.to_owned(), v.to_owned()))
+        .collect::<HashMap<String, String>>()
+        .into()
+}
+
+unsafe extern "C" fn set_fn_wrapper(
+    options: &mut FFI_ExtensionOptions,
+    key: RStr,
+    value: RStr,
+) -> RResult<(), RString> {
+    let _ = options.inner_mut().insert(key.into(), value.into());
+    RResult::ROk(())
+}
+
+unsafe extern "C" fn entries_fn_wrapper(
+    options: &FFI_ExtensionOptions,
+) -> RVec<Tuple2<RString, RString>> {
+    options
+        .inner()
+        .iter()
+        .map(|(key, value)| (key.to_owned().into(), value.to_owned().into()).into())
+        .collect()
+}
+
+unsafe extern "C" fn release_fn_wrapper(options: &mut FFI_ExtensionOptions) {
+    unsafe {
+        debug_assert!(!options.private_data.is_null());
+        let private_data =
+            Box::from_raw(options.private_data as *mut ExtensionOptionsPrivateData);
+        drop(private_data);
+        options.private_data = std::ptr::null_mut();
+    }
+}
+
+impl Default for FFI_ExtensionOptions {
+    fn default() -> Self {
+        HashMap::new().into()
+    }
+}
+
+impl From<HashMap<String, String>> for FFI_ExtensionOptions {
+    fn from(options: HashMap<String, String>) -> Self {
+        let private_data = ExtensionOptionsPrivateData { options };
+
+        Self {
+            cloned: cloned_fn_wrapper,
+            set: set_fn_wrapper,
+            entries: entries_fn_wrapper,
+            release: release_fn_wrapper,
+            private_data: Box::into_raw(Box::new(private_data)) as *mut c_void,
+        }
+    }
+}
+
+impl Drop for FFI_ExtensionOptions {
+    fn drop(&mut self) {
+        unsafe { (self.release)(self) }
+    }
+}
+
+impl Clone for FFI_ExtensionOptions {
+    fn clone(&self) -> Self {
+        unsafe { (self.cloned)(self) }
+    }
+}
+
+impl ConfigExtension for FFI_ExtensionOptions {
+    const PREFIX: &'static str =
+        datafusion_common::config::DATAFUSION_FFI_CONFIG_NAMESPACE;
+}
+
+impl ExtensionOptions for FFI_ExtensionOptions {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn as_any_mut(&mut self) -> &mut dyn Any {
+        self
+    }
+
+    fn cloned(&self) -> Box<dyn ExtensionOptions> {
+        let ffi_options = unsafe { (self.cloned)(self) };
+        Box::new(ffi_options)
+    }
+
+    fn set(&mut self, key: &str, value: &str) -> Result<()> {
+        if key.split_once('.').is_none() {
+            return exec_err!("Unable to set FFI config value without namespace set");
+        };
+
+        df_result!(unsafe { (self.set)(self, key.into(), value.into()) })
+    }
+
+    fn entries(&self) -> Vec<ConfigEntry> {
+        unsafe {
+            (self.entries)(self)
+                .into_iter()
+                .map(|entry_tuple| ConfigEntry {
+                    key: entry_tuple.0.into(),
+                    value: Some(entry_tuple.1.into()),
+                    description: "ffi_config_options",
+                })
+                .collect()
+        }
+    }
+}
+
+impl FFI_ExtensionOptions {
+    /// Add all of the values in a concrete configuration extension to the
+    /// FFI variant. This is safe to call on either side of the FFI
+    /// boundary.
+    pub fn add_config<C: ConfigExtension>(&mut self, config: &C) -> Result<()> {
+        for entry in config.entries() {
+            if let Some(value) = entry.value {
+                let key = format!("{}.{}", C::PREFIX, entry.key);
+                self.set(key.as_str(), value.as_str())?;
+            }
+        }
+
+        Ok(())
+    }
+
+    /// Merge another `FFI_ExtensionOptions` configurations into this one.
+    /// This is safe to call on either side of the FFI boundary.
+    pub fn merge(&mut self, other: &FFI_ExtensionOptions) -> Result<()> {
+        for entry in other.entries() {
+            if let Some(value) = entry.value {
+                self.set(entry.key.as_str(), value.as_str())?;
+            }
+        }
+        Ok(())
+    }
+
+    /// Create a concrete extension type from the FFI variant.
+    /// This is safe to call on either side of the FFI boundary.
+    pub fn to_extension<C: ConfigExtension + Default>(&self) -> Result<C> {
+        let mut result = C::default();
+
+        unsafe {
+            for entry in (self.entries)(self) {
+                let key = entry.0.as_str();
+                let value = entry.1.as_str();
+
+                if let Some((prefix, inner_key)) = key.split_once('.')
+                    && prefix == C::PREFIX
+                {
+                    result.set(inner_key, value)?;
+                }
+            }
+        }
+
+        Ok(result)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use datafusion_common::config::{ConfigExtension, ConfigOptions};
+    use datafusion_common::extensions_options;
+
+    use crate::config::extension_options::FFI_ExtensionOptions;
+
+    // Define a new configuration struct using the `extensions_options` macro
+    extensions_options! {
+       /// My own config options.
+       pub struct MyConfig {
+           /// Should "foo" be replaced by "bar"?
+           pub foo_to_bar: bool, default = true
+
+           /// How many "baz" should be created?
+           pub baz_count: usize, default = 1337
+       }
+    }
+
+    impl ConfigExtension for MyConfig {
+        const PREFIX: &'static str = "my_config";
+    }
+
+    #[test]
+    fn round_trip_ffi_extension_options() {
+        // set up config struct and register extension
+        let mut config = ConfigOptions::default();
+        let mut ffi_options = FFI_ExtensionOptions::default();
+        ffi_options.add_config(&MyConfig::default()).unwrap();
+
+        config.extensions.insert(ffi_options);
+
+        // overwrite config default
+        config.set("my_config.baz_count", "42").unwrap();
+
+        // check config state
+        let returned_ffi_config =
+            config.extensions.get::<FFI_ExtensionOptions>().unwrap();
+        let my_config: MyConfig = returned_ffi_config.to_extension().unwrap();
+
+        // check default value
+        assert!(my_config.foo_to_bar);
+
+        // check overwritten value
+        assert_eq!(my_config.baz_count, 42);
+    }
+}