Fix deepcopy for And, Or, and Not expressions

pyiceberg/expressions/__init__.py

@@ -17,6 +17,7 @@
 
 from __future__ import annotations
 
+import copy
 from abc import ABC, abstractmethod
 from collections.abc import Callable, Iterable, Sequence
 from functools import cached_property
@@ -51,6 +52,12 @@ def _to_literal(value: L | Literal[L]) -> Literal[L]:
 class BooleanExpression(IcebergBaseModel, ABC):
     """An expression that evaluates to a boolean."""
 
+    def __deepcopy__(self, memo: dict[int, Any]) -> BooleanExpression:
+        if isinstance(self, Singleton):
+            return self
+        fields = {name: copy.deepcopy(getattr(self, name), memo) for name in type(self).model_fields}
+        return type(self)(**fields)
+
     @abstractmethod
     def __invert__(self) -> BooleanExpression:
         """Transform the Expression into its negated version."""

tests/expressions/test_expressions.py

@@ -16,6 +16,7 @@
 # under the License.
 # pylint:disable=redefined-outer-name,eval-used
 
+import copy
 import pickle
 import uuid
 from decimal import Decimal
@@ -1292,6 +1293,74 @@ def test_bind_ambiguous_name() -> None:
     assert "Invalid schema, multiple fields for name foo.bar: 2 and 3" in str(exc_info)
 
 
+# --- deepcopy tests ---
+
+
+def test_deepcopy_and() -> None:
+    expr = And(EqualTo("x", 1), EqualTo("y", 2))
+    copied = copy.deepcopy(expr)
+    assert copied == expr
+
+
+def test_deepcopy_or() -> None:
+    expr = Or(EqualTo("x", 1), EqualTo("y", 2))
+    copied = copy.deepcopy(expr)
+    assert copied == expr
+
+
+def test_deepcopy_not() -> None:
+    expr = Not(EqualTo("x", 1))
+    copied = copy.deepcopy(expr)
+    assert copied == expr
+
+
+def test_deepcopy_equal_to() -> None:
+    expr = EqualTo("x", 1)
+    copied = copy.deepcopy(expr)
+    assert copied == expr
+
+
+def test_deepcopy_always_true() -> None:
+    copied = copy.deepcopy(AlwaysTrue())
+    assert copied is AlwaysTrue()
+
+
+def test_deepcopy_always_false() -> None:
+    copied = copy.deepcopy(AlwaysFalse())
+    assert copied is AlwaysFalse()
+
+
+def test_deepcopy_always_true_then_pickle() -> None:
+    copied = copy.deepcopy(AlwaysTrue())
+    restored = pickle.loads(pickle.dumps(copied))
+    assert restored is AlwaysTrue()
+
+
+def test_deepcopy_balanced_and() -> None:
+    expr = And(EqualTo("a", 1), EqualTo("b", 2), EqualTo("c", 3), EqualTo("d", 4))
+    copied = copy.deepcopy(expr)
+    assert copied == expr
+
+
+def test_deepcopy_balanced_or() -> None:
+    expr = Or(EqualTo("a", 1), EqualTo("b", 2), EqualTo("c", 3), EqualTo("d", 4))
+    copied = copy.deepcopy(expr)
+    assert copied == expr
+
+
+def test_deepcopy_nested_expression() -> None:
+    expr = And(Or(EqualTo("a", 1), EqualTo("b", 2)), Not(EqualTo("c", 3)))
+    copied = copy.deepcopy(expr)
+    assert copied == expr
+
+
+def test_deepcopy_then_pickle() -> None:
+    expr = And(EqualTo("x", 1), EqualTo("y", 2))
+    copied = copy.deepcopy(expr)
+    restored = pickle.loads(pickle.dumps(copied))
+    assert restored == expr
+
+
 #   __  __      ___
 #  |  \/  |_  _| _ \_  _
 #  | |\/| | || |  _/ || |