fix(python): reuse table state in write engine

python/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "deltalake-python"
-version = "0.17.1"
+version = "0.17.2"
 authors = ["Qingping Hou <[email protected]>", "Will Jones <[email protected]>"]
 homepage = "https://github.com/delta-io/delta-rs"
 license = "Apache-2.0"

python/deltalake/_internal.pyi

@@ -174,6 +174,7 @@ def write_to_deltalake(
     data: pyarrow.RecordBatchReader,
     partition_by: Optional[List[str]],
     mode: str,
+    table: Optional[RawDeltaTable],
     schema_mode: Optional[str],
     predicate: Optional[str],
     name: Optional[str],

python/deltalake/writer.py

@@ -318,6 +318,7 @@ def write_deltalake(
             data=data,
             partition_by=partition_by,
             mode=mode,
+            table=table._table if table is not None else None,
             schema_mode=schema_mode,
             predicate=predicate,
             name=name,

python/src/lib.rs

@@ -1410,6 +1410,7 @@ fn write_to_deltalake(
     table_uri: String,
     data: PyArrowType<ArrowArrayStreamReader>,
     mode: String,
+    table: Option<&RawDeltaTable>,
     schema_mode: Option<String>,
     partition_by: Option<Vec<String>>,
     predicate: Option<String>,
@@ -1425,11 +1426,14 @@ fn write_to_deltalake(
         let save_mode = mode.parse().map_err(PythonError::from)?;
 
         let options = storage_options.clone().unwrap_or_default();
-        let table = rt()
-            .block_on(DeltaOps::try_from_uri_with_storage_options(
+        let table = if let Some(table) = table {
+            DeltaOps(table._table.clone())
+        } else {
+            rt().block_on(DeltaOps::try_from_uri_with_storage_options(
                 &table_uri, options,
             ))
-            .map_err(PythonError::from)?;
+            .map_err(PythonError::from)?
+        };
 
         let mut builder = table.write(batches).with_save_mode(save_mode);
         if let Some(schema_mode) = schema_mode {