make PyObjectWrapper picklable (#7664)

GitOrigin-RevId: c9f5fac558a160da6224bf71b7fe9a5279ea8876

CHANGELOG.md

@@ -13,6 +13,9 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - `pw.Table.diff` now supports setting `instance` parameter that allows computing differences for multiple groups.
 - `pw.io.postgres.write_snapshot` now keeps the Postgres table fully in sync with the current state of the table in Pathway. This means that if an entry is deleted in Pathway, the same entry will also be deleted from the Postgres table managed by the output connector.
 
+### Fixed
+- `pw.PyObjectWrapper` is now picklable.
+
 ## [0.15.3] - 2024-11-07
 
 ### Added

python/pathway/tests/test_py_object_wrapper.py

@@ -1,5 +1,6 @@
 # Copyright © 2024 Pathway
 
+import copy
 import pickle
 from dataclasses import dataclass
 from pathlib import Path
@@ -309,3 +310,48 @@ def test_serialization_simple(serialization: str) -> None:
     des = pickle.loads(ser_bytes)
     ob = des.loads(b)
     assert ob == SimpleStr("def")
+
+
+def test_copy():
+    s = [1, 2, 3]
+    p = pw.PyObjectWrapper(s)
+    q = copy.copy(p)
+    q.value.append(4)
+    assert s == [1, 2, 3, 4]
+    assert q.value == [1, 2, 3, 4]
+
+
+def test_deepcopy():
+    s = [1, 2, 3]
+    p = pw.PyObjectWrapper(s)
+    q = copy.deepcopy(p)
+    q.value.append(4)
+    assert s == [1, 2, 3]
+    assert q.value == [1, 2, 3, 4]
+
+
+def test_cache(tmp_path):
+    t = pw.debug.table_from_markdown(
+        """
+        a
+        1
+        2
+        3
+        2
+    """
+    )
+
+    @pw.udf(cache_strategy=pw.udfs.DiskCache())
+    def f(a: int) -> pw.PyObjectWrapper:
+        return pw.PyObjectWrapper(a)
+
+    @pw.udf
+    def g(a: pw.PyObjectWrapper) -> int:
+        return a.value
+
+    res = t.select(a=g(f(pw.this.a)))
+    persistence_config = pw.persistence.Config(
+        pw.persistence.Backend.filesystem(tmp_path),
+        persistence_mode=pw.PersistenceMode.UDF_CACHING,
+    )
+    assert_table_equality(t, res, persistence_config=persistence_config)

src/python_api.rs

@@ -814,6 +814,10 @@ impl PyObjectWrapper {
     ) -> PyResult<Bound<'py, PyAny>> {
         generic_alias_class_getitem(cls, item)
     }
+
+    fn __getnewargs__(&self) -> (PyObject,) {
+        (self.value.clone(),)
+    }
 }
 
 impl PyObjectWrapper {