Make And expression JSON serializable using Pydantic

pyiceberg/expressions/__init__.py

@@ -259,12 +259,18 @@ def as_bound(self) -> Type[BoundReference[L]]:
         return BoundReference[L]
 
 
-class And(BooleanExpression):
+class And(IcebergBaseModel, BooleanExpression):
     """AND operation expression - logical conjunction."""
 
+    model_config = ConfigDict(arbitrary_types_allowed=True, frozen=False)
+
+    type: TypingLiteral["and"] = Field(default="and", alias="type")
     left: BooleanExpression
     right: BooleanExpression
 
+    def __init__(self, left: BooleanExpression, right: BooleanExpression, *rest: BooleanExpression) -> None:
+        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))
@@ -276,6 +282,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

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)