Make And expression JSON serializable using Pydantic

Author: Aniketsy

pyiceberg/expressions/__init__.py

@@ -237,12 +237,19 @@ def as_bound(self) -> type[BoundReference]:
         return BoundReference
 
 
-class And(BooleanExpression):
+class And(IcebergBaseModel, BooleanExpression):
     """AND operation expression - logical conjunction."""
 
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    type: TypingLiteral["and"] = Field(default="and", alias="type")
     left: BooleanExpression
     right: BooleanExpression
 
+    def __init__(self, left: BooleanExpression, right: BooleanExpression, *rest: BooleanExpression) -> None:
+        if isinstance(self, And) and not hasattr(self, "left") and not hasattr(self, "right"):
+            super().__init__(left=left, right=right)
+
     def __new__(cls, left: BooleanExpression, right: BooleanExpression, *rest: BooleanExpression) -> BooleanExpression:  # type: ignore
         if rest:
             return _build_balanced_tree(And, (left, right, *rest))
@@ -254,6 +261,7 @@ def __new__(cls, left: BooleanExpression, right: BooleanExpression, *rest: Boole
             return left
         else:
             obj = super().__new__(cls)
+            obj.__pydantic_fields_set__ = set()
             obj.left = left
             obj.right = right
             return obj
@@ -264,7 +272,7 @@ def __eq__(self, other: Any) -> bool:
 
     def __str__(self) -> str:
         """Return the string representation of the And class."""
-        return f"And(left={str(self.left)}, right={str(self.right)})"
+        return f"{str(self.__class__.__name__)}(left={repr(self.left)}, right={repr(self.right)})"
 
     def __repr__(self) -> str:
         """Return the string representation of the And class."""

tests/table/test_partitioning.py

@@ -21,6 +21,7 @@
 
 import pytest
 
+from pyiceberg.expressions import And, EqualTo
 from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC, PartitionField, PartitionSpec
 from pyiceberg.schema import Schema
 from pyiceberg.transforms import (
@@ -125,6 +126,13 @@ def test_serialize_partition_spec() -> None:
     )
 
 
+def test_serialize_and_expression() -> None:
+    expr = And(EqualTo("foo", 1), EqualTo("bar", 2))
+    assert expr.model_dump_json(by_alias=True) == (
+        '{"type":"and","left":{"type":"equal_to","term":"foo","literal":1},"right":{"type":"equal_to","term":"bar","literal":2}}'
+    )
+
+
 def test_deserialize_unpartition_spec() -> None:
     json_partition_spec = """{"spec-id":0,"fields":[]}"""
     spec = PartitionSpec.model_validate_json(json_partition_spec)