fix: allow user defined config keys (#1365)

# Description

When switching to the create operation to back `write_deltalake` we
allowed only known configuration keys on metadata. THis fixed that
regression.

# Related Issue(s)

closes #1353

python/src/lib.rs

@@ -20,9 +20,7 @@ use deltalake::operations::optimize::OptimizeBuilder;
 use deltalake::operations::transaction::commit;
 use deltalake::operations::vacuum::VacuumBuilder;
 use deltalake::partitions::PartitionFilter;
-use deltalake::{
-    DeltaConfigKey, DeltaDataTypeLong, DeltaDataTypeTimestamp, DeltaOps, Invariant, Schema,
-};
+use deltalake::{DeltaDataTypeLong, DeltaDataTypeTimestamp, DeltaOps, Invariant, Schema};
 use pyo3::create_exception;
 use pyo3::exceptions::PyException;
 use pyo3::exceptions::PyValueError;
@@ -33,7 +31,6 @@ use std::collections::HashMap;
 use std::collections::HashSet;
 use std::convert::TryFrom;
 use std::future::IntoFuture;
-use std::str::FromStr;
 use std::sync::Arc;
 use std::time::SystemTime;
 use std::time::UNIX_EPOCH;
@@ -776,23 +773,10 @@ fn write_new_deltalake(
         .try_into()
         .map_err(PyDeltaTableError::from_arrow)?;
 
-    let configuration = configuration
-        .unwrap_or_default()
-        .into_iter()
-        .filter_map(|(key, value)| {
-            if let Ok(key) = DeltaConfigKey::from_str(&key) {
-                Some((key, value))
-            } else {
-                None
-            }
-        })
-        .collect();
-
     let mut builder = DeltaOps(table)
         .create()
         .with_columns(schema.get_fields().clone())
         .with_partition_columns(partition_by)
-        .with_configuration(configuration)
         .with_actions(add_actions.iter().map(|add| Action::add(add.into())));
 
     if let Some(name) = &name {
@@ -803,6 +787,10 @@ fn write_new_deltalake(
         builder = builder.with_comment(description);
     };
 
+    if let Some(config) = configuration {
+        builder = builder.with_configuration(config);
+    };
+
     rt()?
         .block_on(builder.into_future())
         .map_err(PyDeltaTableError::from_raw)?;

python/tests/test_writer.py

@@ -142,7 +142,7 @@ def test_roundtrip_metadata(tmp_path: pathlib.Path, sample_data: pa.Table):
         sample_data,
         name="test_name",
         description="test_desc",
-        configuration={"delta.appendOnly": "false"},
+        configuration={"delta.appendOnly": "false", "foo": "bar"},
     )
 
     delta_table = DeltaTable(tmp_path)
@@ -151,7 +151,7 @@ def test_roundtrip_metadata(tmp_path: pathlib.Path, sample_data: pa.Table):
 
     assert metadata.name == "test_name"
     assert metadata.description == "test_desc"
-    assert metadata.configuration == {"delta.appendOnly": "false"}
+    assert metadata.configuration == {"delta.appendOnly": "false", "foo": "bar"}
 
 
 @pytest.mark.parametrize(

rust/src/operations/create.rs

@@ -158,11 +158,11 @@ impl CreateBuilder {
     /// Set configuration on created table
     pub fn with_configuration(
         mut self,
-        configuration: HashMap<DeltaConfigKey, Option<impl Into<String>>>,
+        configuration: impl IntoIterator<Item = (impl Into<String>, Option<impl Into<String>>)>,
     ) -> Self {
         self.configuration = configuration
             .into_iter()
-            .map(|(k, v)| (k.as_ref().into(), v.map(|s| s.into())))
+            .map(|(k, v)| (k.into(), v.map(|s| s.into())))
             .collect();
         self
     }