expose custom metadata

python/deltalake/_internal.pyi

@@ -163,6 +163,7 @@ def write_to_deltalake(
     configuration: Optional[Mapping[str, Optional[str]]],
     storage_options: Optional[Dict[str, str]],
     writer_properties: Optional[Dict[str, Optional[str]]],
+    custom_metadata: Optional[Dict[str, str]],
 ) -> None: ...
 def convert_to_deltalake(
     uri: str,

python/deltalake/writer.py

@@ -128,6 +128,7 @@ def write_deltalake(
     large_dtypes: bool = ...,
     engine: Literal["rust"],
     writer_properties: WriterProperties = ...,
+    custom_metadata: Optional[Dict[str, str]] = ...,
 ) -> None:
     ...
 
@@ -163,6 +164,7 @@ def write_deltalake(
     large_dtypes: bool = False,
     engine: Literal["pyarrow", "rust"] = "pyarrow",
     writer_properties: Optional[WriterProperties] = None,
+    custom_metadata: Optional[Dict[str, str]] = None,
 ) -> None:
     """Write to a Delta Lake table
 
@@ -236,6 +238,7 @@ def write_deltalake(
         engine: writer engine to write the delta table. `Rust` engine is still experimental but you may
             see up to 4x performance improvements over pyarrow.
         writer_properties: Pass writer properties to the Rust parquet writer.
+        custom_metadata: Custom metadata to add to the commitInfo, rust writer only.
     """
     table, table_uri = try_get_table_and_table_uri(table_or_uri, storage_options)
     if table is not None:
@@ -300,6 +303,7 @@ def write_deltalake(
             writer_properties=writer_properties._to_dict()
             if writer_properties
             else None,
+            custom_metadata=custom_metadata,
         )
         if table:
             table.update_incremental()

python/src/lib.rs

@@ -1173,6 +1173,7 @@ fn write_to_deltalake(
     configuration: Option<HashMap<String, Option<String>>>,
     storage_options: Option<HashMap<String, String>>,
     writer_properties: Option<HashMap<String, Option<String>>>,
+    custom_metadata: Option<HashMap<String, String>>,
 ) -> PyResult<()> {
     let batches = data.0.map(|batch| batch.unwrap()).collect::<Vec<_>>();
     let save_mode = mode.parse().map_err(PythonError::from)?;
@@ -1216,6 +1217,12 @@ fn write_to_deltalake(
         builder = builder.with_configuration(config);
     };
 
+    if let Some(metadata) = custom_metadata {
+        let json_metadata: Map<String, Value> =
+            metadata.into_iter().map(|(k, v)| (k, v.into())).collect();
+        builder = builder.with_metadata(json_metadata);
+    };
+
     rt()?
         .block_on(builder.into_future())
         .map_err(PythonError::from)?;