fix: add raise_if_key_not_exists to CreateBuilder (delta-io#2565)

This PR adds a `raise_if_key_not_exists` flag to the `CreateBuilder`
which we propagate (currently always `true`), such that users can choose
not to fail if the configuration contains a key that is not a valid
DeltaConfigKey. Also expose this in the Python binding.

Closes delta-io#2564.

@ion-elgreco

crates/core/src/operations/create.rs

@@ -62,6 +62,7 @@ pub struct CreateBuilder {
     log_store: Option<LogStoreRef>,
     configuration: HashMap<String, Option<String>>,
     metadata: Option<HashMap<String, Value>>,
+    raise_if_key_not_exists: bool,
 }
 
 impl super::Operation<()> for CreateBuilder {}
@@ -87,6 +88,7 @@ impl CreateBuilder {
             log_store: None,
             configuration: Default::default(),
             metadata: Default::default(),
+            raise_if_key_not_exists: true,
         }
     }
 
@@ -196,6 +198,12 @@ impl CreateBuilder {
         self
     }
 
+    /// Specify whether to raise an error if the table properties in the configuration are not DeltaConfigKeys
+    pub fn with_raise_if_not_exists(mut self, raise_if_key_not_exists: bool) -> Self {
+        self.raise_if_key_not_exists = raise_if_key_not_exists;
+        self
+    }
+
     /// Specify additional actions to be added to the commit.
     ///
     /// This method is mainly meant for internal use. Manually adding inconsistent
@@ -279,7 +287,7 @@ impl CreateBuilder {
                 .iter()
                 .map(|(k, v)| (k.clone(), v.clone().unwrap()))
                 .collect::<HashMap<String, String>>(),
-            true,
+            self.raise_if_key_not_exists,
         )?;
 
         let protocol = convert_properties_to_features(protocol, &configuration);
@@ -569,4 +577,43 @@ mod tests {
         // Checks if files got removed after overwrite
         assert_eq!(table.get_files_count(), 0);
     }
+
+    #[tokio::test]
+    async fn test_create_table_metadata_raise_if_key_not_exists() {
+        let schema = get_delta_schema();
+        let config: HashMap<String, Option<String>> =
+            vec![("key".to_string(), Some("value".to_string()))]
+                .into_iter()
+                .collect();
+
+        // Fail to create table with unknown Delta key
+        let table = CreateBuilder::new()
+            .with_location("memory://")
+            .with_columns(schema.fields().clone())
+            .with_configuration(config.clone())
+            .await;
+        assert!(table.is_err());
+
+        // Succeed in creating table with unknown Delta key since we set raise_if_key_not_exists to false
+        let table = CreateBuilder::new()
+            .with_location("memory://")
+            .with_columns(schema.fields().clone())
+            .with_raise_if_not_exists(false)
+            .with_configuration(config)
+            .await;
+        assert!(table.is_ok());
+
+        // Ensure the non-Delta key was set correctly
+        let value = table
+            .unwrap()
+            .metadata()
+            .unwrap()
+            .configuration
+            .get("key")
+            .unwrap()
+            .as_ref()
+            .unwrap()
+            .clone();
+        assert_eq!(String::from("value"), value);
+    }
 }

python/deltalake/_internal.pyi

@@ -214,6 +214,7 @@ def create_deltalake(
     schema: pyarrow.Schema,
     partition_by: List[str],
     mode: str,
+    raise_if_key_not_exists: bool,
     name: Optional[str],
     description: Optional[str],
     configuration: Optional[Mapping[str, Optional[str]]],

python/deltalake/table.py

@@ -457,6 +457,7 @@ def create(
         configuration: Optional[Mapping[str, Optional[str]]] = None,
         storage_options: Optional[Dict[str, str]] = None,
         custom_metadata: Optional[Dict[str, str]] = None,
+        raise_if_key_not_exists: bool = True,
     ) -> "DeltaTable":
         """`CREATE` or `CREATE_OR_REPLACE` a delta table given a table_uri.
 
@@ -471,8 +472,9 @@ def create(
             name: User-provided identifier for this table.
             description: User-provided description for this table.
             configuration:  A map containing configuration options for the metadata action.
-            storage_options: options passed to the object store crate.
-            custom_metadata: custom metadata that will be added to the transaction commit.
+            storage_options: Options passed to the object store crate.
+            custom_metadata: Custom metadata that will be added to the transaction commit.
+            raise_if_key_not_exists: Whether to raise an error if the configuration uses keys that are not Delta keys
 
         Returns:
             DeltaTable: created delta table
@@ -506,6 +508,7 @@ def create(
             schema,
             partition_by or [],
             mode,
+            raise_if_key_not_exists,
             name,
             description,
             configuration,

python/src/lib.rs

@@ -1650,6 +1650,7 @@ fn create_deltalake(
     schema: PyArrowType<ArrowSchema>,
     partition_by: Vec<String>,
     mode: String,
+    raise_if_key_not_exists: bool,
     name: Option<String>,
     description: Option<String>,
     configuration: Option<HashMap<String, Option<String>>>,
@@ -1669,6 +1670,7 @@ fn create_deltalake(
             .create()
             .with_columns(schema.fields().clone())
             .with_save_mode(mode)
+            .with_raise_if_not_exists(raise_if_key_not_exists)
             .with_partition_columns(partition_by);
 
         if let Some(name) = &name {