fix: allow writing to DeltaTable objects across Python threads

FREEZE!

Looks like we were _really_ close to full multi-threaded support before,
and just needed to sprinkle a little macro magic on the pyclass
definition.

See [pyo3 docs](https://pyo3.rs/v0.23.0/class/thread-safety)

Fixes #3594

Signed-off-by: R. Tyler Croy <[email protected]>

Author: rtyler

python/deltalake/writer/_utils.py

@@ -33,7 +33,7 @@ def try_get_table_and_table_uri(
         table_uri = str(table_or_uri)
     else:
         table = table_or_uri
-        table_uri = table._table.table_uri()
+        table_uri = table.table_uri
 
     return (table, table_uri)
 

python/src/lib.rs

@@ -100,7 +100,7 @@ enum PartitionFilterValue {
     Multiple(Vec<PyBackedStr>),
 }
 
-#[pyclass(module = "deltalake._internal")]
+#[pyclass(module = "deltalake._internal", frozen)]
 struct RawDeltaTable {
     /// The internal reference to the table is guarded by a Mutex to allow for re-using the same
     /// [DeltaTable] instance across multiple Python threads
@@ -109,7 +109,7 @@ struct RawDeltaTable {
     _config: FsConfig,
 }
 
-#[pyclass]
+#[pyclass(frozen)]
 struct RawDeltaTableMetaData {
     #[pyo3(get)]
     id: String,
@@ -1645,7 +1645,7 @@ impl RawDeltaTable {
     #[allow(clippy::too_many_arguments)]
     #[pyo3(signature = (data, batch_schema, mode, schema_mode=None, partition_by=None, predicate=None, target_file_size=None, name=None, description=None, configuration=None, writer_properties=None, commit_properties=None, post_commithook_properties=None))]
     fn write(
-        &mut self,
+        &self,
         py: Python,
         data: PyRecordBatchReader,
         batch_schema: PyArrowSchema,

python/tests/test_threaded.py

@@ -59,13 +59,16 @@ def test_multithreaded_write_using_table(tmp_path: pathlib.Path):
 
     dt = DeltaTable(tmp_path)
 
-    with pytest.raises(RuntimeError, match="borrowed"):
-        with ThreadPoolExecutor() as exe:
-            list(exe.map(lambda _: write_deltalake(dt, table, mode="append"), range(5)))
+    with ThreadPoolExecutor() as exe:
+        list(
+            exe.map(
+                lambda i: write_deltalake(dt, pl.DataFrame({"a": [i]}), mode="append"),
+                range(5),
+            )
+        )
 
 
 @pytest.mark.polars
-@pytest.mark.xfail(reason="Can fail because of already borrowed")
 def test_multithreaded_write_using_path(tmp_path: pathlib.Path):
     import polars as pl