Add CAST expression support to parser

pyiceberg/expressions/parser.py

@@ -16,6 +16,7 @@
 #  under the License.
 import re
 from decimal import Decimal
+from typing import Any
 
 from pyparsing import (
     CaselessKeyword,
@@ -58,6 +59,7 @@
     Or,
     Reference,
     StartsWith,
+    UnboundTerm,
 )
 from pyiceberg.expressions.literals import (
     BooleanLiteral,
@@ -66,6 +68,14 @@
     LongLiteral,
     StringLiteral,
 )
+from pyiceberg.transforms import (
+    DayTransform,
+    HourTransform,
+    MonthTransform,
+    Transform,
+    UnboundTransform,
+    YearTransform,
+)
 from pyiceberg.typedef import L
 from pyiceberg.types import strtobool
 
@@ -80,6 +90,8 @@
 NAN = CaselessKeyword("nan")
 LIKE = CaselessKeyword("like")
 BETWEEN = CaselessKeyword("between")
+CAST = CaselessKeyword("cast")
+AS = CaselessKeyword("as")
 
 unquoted_identifier = Word(alphas + "_", alphanums + "_$")
 quoted_identifier = QuotedString('"', esc_quote="\\", unquote_results=True)
@@ -153,26 +165,30 @@ def _(result: ParseResults) -> Literal[L]:
 between = column + BETWEEN + literal + AND + literal
 
 
+def _comparison_to_predicate(term: UnboundTerm, op: str, literal: Literal[Any]) -> BooleanExpression:
+    if op == "<":
+        return LessThan(term, literal)
+    elif op == "<=":
+        return LessThanOrEqual(term, literal)
+    elif op == ">":
+        return GreaterThan(term, literal)
+    elif op == ">=":
+        return GreaterThanOrEqual(term, literal)
+    elif op in ("=", "=="):
+        return EqualTo(term, literal)
+    elif op in ("!=", "<>"):
+        return NotEqualTo(term, literal)
+    raise ValueError(f"Unsupported operation type: {op}")
+
+
 @between.set_parse_action
 def _(result: ParseResults) -> BooleanExpression:
     return And(GreaterThanOrEqual(result.column, result[2]), LessThanOrEqual(result.column, result[4]))
 
 
 @left_ref.set_parse_action
 def _(result: ParseResults) -> BooleanExpression:
-    if result.op == "<":
-        return LessThan(result.column, result.literal)
-    elif result.op == "<=":
-        return LessThanOrEqual(result.column, result.literal)
-    elif result.op == ">":
-        return GreaterThan(result.column, result.literal)
-    elif result.op == ">=":
-        return GreaterThanOrEqual(result.column, result.literal)
-    if result.op in ("=", "=="):
-        return EqualTo(result.column, result.literal)
-    if result.op in ("!=", "<>"):
-        return NotEqualTo(result.column, result.literal)
-    raise ValueError(f"Unsupported operation type: {result.op}")
+    return _comparison_to_predicate(result.column, result.op, result.literal)
 
 
 @right_ref.set_parse_action
@@ -265,7 +281,38 @@ def _evaluate_like_statement(result: ParseResults) -> BooleanExpression:
         return EqualTo(result.column, StringLiteral(literal_like.value.replace("\\%", "%")))
 
 
-predicate = (between | comparison | in_check | null_check | nan_check | starts_check | boolean).set_results_name("predicate")
+# CAST expression support: CAST(column AS type) maps to Iceberg transforms
+_CAST_TYPE_TO_TRANSFORM: dict[str, Transform[Any, Any]] = {
+    "date": DayTransform(),
+    "year": YearTransform(),
+    "month": MonthTransform(),
+    "hour": HourTransform(),
+}
+
+cast_type = Word(alphas, alphanums + "_")
+cast_term = Suppress(CAST) + Suppress("(") + column + Suppress(AS) + cast_type + Suppress(")")
+
+
+@cast_term.set_parse_action
+def _(result: ParseResults) -> UnboundTransform:
+    ref = result[0]
+    target = str(result[1]).lower()
+    if target not in _CAST_TYPE_TO_TRANSFORM:
+        raise ValueError(f"Unsupported CAST target type: {target}")
+    return UnboundTransform(ref, _CAST_TYPE_TO_TRANSFORM[target])
+
+
+cast_left_ref = cast_term + comparison_op + literal
+
+
+@cast_left_ref.set_parse_action
+def _(result: ParseResults) -> BooleanExpression:
+    return _comparison_to_predicate(result[0], result.op, result.literal)
+
+
+predicate = (between | cast_left_ref | comparison | in_check | null_check | nan_check | starts_check | boolean).set_results_name(
+    "predicate"
+)
 
 
 def handle_not(result: ParseResults) -> Not:

pyiceberg/transforms.py

@@ -44,6 +44,7 @@
     BoundNotIn,
     BoundNotStartsWith,
     BoundPredicate,
+    BoundReference,
     BoundSetPredicate,
     BoundStartsWith,
     BoundTerm,
@@ -58,6 +59,7 @@
     Reference,
     StartsWith,
     UnboundPredicate,
+    UnboundTerm,
 )
 from pyiceberg.expressions.literals import (
     DateLiteral,
@@ -67,7 +69,7 @@
     TimestampLiteral,
     literal,
 )
-from pyiceberg.typedef import IcebergRootModel, L
+from pyiceberg.typedef import IcebergRootModel, L, StructProtocol
 from pyiceberg.types import (
     BinaryType,
     DateType,
@@ -92,6 +94,8 @@
 if TYPE_CHECKING:
     import pyarrow as pa
 
+    from pyiceberg.schema import Schema
+
     ArrayLike = TypeVar("ArrayLike", pa.Array, pa.ChunkedArray)
 
 S = TypeVar("S")
@@ -1152,3 +1156,49 @@ class BoundTransform(BoundTerm):
     def __init__(self, term: BoundTerm, transform: Transform[Any, Any]):
         self.term: BoundTerm = term
         self.transform = transform
+
+    def ref(self) -> BoundReference:
+        """Return the bound reference of the underlying term."""
+        return self.term.ref()
+
+    def eval(self, struct: StructProtocol) -> Any:
+        """Evaluate the transform on the struct value."""
+        return self.transform.transform(self.term.ref().field.field_type)(self.term.eval(struct))
+
+
+class UnboundTransform(UnboundTerm):
+    """An unbound transform expression that pairs a column reference with a transform.
+
+    When bound to a schema, validates that the transform is compatible with the
+    source column type and produces a BoundTransform.
+    """
+
+    transform: Transform[Any, Any]
+
+    def __init__(self, term: UnboundTerm, transform: Transform[Any, Any]):
+        self.term: UnboundTerm = term
+        self.transform = transform
+
+    def bind(self, schema: "Schema", case_sensitive: bool = True) -> BoundTransform:
+        """Bind this unbound transform to a schema, returning a BoundTransform."""
+        bound_term = self.term.bind(schema, case_sensitive)
+        source_type = bound_term.ref().field.field_type
+        if not self.transform.can_transform(source_type):
+            raise TypeError(f"Cannot apply {self.transform} to {source_type}")
+        return BoundTransform(bound_term, self.transform)
+
+    @property
+    def as_bound(self) -> type[BoundTransform]:
+        return BoundTransform
+
+    def __eq__(self, other: object) -> bool:
+        """Return the equality of two instances of the UnboundTransform class."""
+        return isinstance(other, UnboundTransform) and self.term == other.term and self.transform == other.transform
+
+    def __repr__(self) -> str:
+        """Return the string representation of the UnboundTransform class."""
+        return f"UnboundTransform(term={self.term}, transform={self.transform})"
+
+    def __hash__(self) -> int:
+        """Return the hash of the UnboundTransform."""
+        return hash((self.term, self.transform))

tests/expressions/test_parser.py

@@ -42,7 +42,17 @@
     Reference,
     StartsWith,
 )
-from pyiceberg.expressions.literals import DecimalLiteral, LongLiteral, literal
+from pyiceberg.expressions.literals import DateLiteral, DecimalLiteral, LongLiteral, StringLiteral, literal
+from pyiceberg.schema import Schema
+from pyiceberg.transforms import (
+    BoundTransform,
+    DayTransform,
+    HourTransform,
+    MonthTransform,
+    UnboundTransform,
+    YearTransform,
+)
+from pyiceberg.types import DateType, NestedField
 
 
 def test_always_true() -> None:
@@ -272,3 +282,94 @@ def test_valid_between_with_numerics() -> None:
     ) == parser.parse("foo between '2025-01-01T00:00:00.000000' and '2025-01-10T12:00:00.000000'")
 
     assert parser.parse("foo between 1 and 3") == parser.parse("1 <= foo and foo <= 3")
+
+
+def test_cast_date_comparison() -> None:
+    expected = EqualTo(
+        UnboundTransform(Reference("created_at"), DayTransform()),
+        StringLiteral("2024-01-01"),
+    )
+    assert expected == parser.parse("CAST(created_at AS date) = '2024-01-01'")
+
+
+def test_cast_date_binds_string_literal_to_date() -> None:
+    """Binding a CAST(... AS date) predicate converts the StringLiteral to a DateLiteral."""
+    schema = Schema(NestedField(1, "created_at", DateType(), required=False))
+    bound = parser.parse("CAST(created_at AS date) = '2024-01-01'").bind(schema)
+    assert isinstance(bound.term, BoundTransform)
+    assert isinstance(bound.literal, DateLiteral)
+
+
+def test_cast_year_comparison() -> None:
+    expected = GreaterThan(
+        UnboundTransform(Reference("ts"), YearTransform()),
+        LongLiteral(2020),
+    )
+    assert expected == parser.parse("CAST(ts AS year) > 2020")
+
+
+def test_cast_month_comparison() -> None:
+    expected = EqualTo(
+        UnboundTransform(Reference("ts"), MonthTransform()),
+        LongLiteral(6),
+    )
+    assert expected == parser.parse("CAST(ts AS month) = 6")
+
+
+def test_cast_hour_comparison() -> None:
+    expected = GreaterThanOrEqual(
+        UnboundTransform(Reference("ts"), HourTransform()),
+        LongLiteral(12),
+    )
+    assert expected == parser.parse("CAST(ts AS hour) >= 12")
+
+
+def test_cast_case_insensitive() -> None:
+    """CAST keyword and type name should be case-insensitive."""
+    expected = EqualTo(
+        UnboundTransform(Reference("created_at"), DayTransform()),
+        StringLiteral("2024-01-01"),
+    )
+    assert expected == parser.parse("cast(created_at as DATE) = '2024-01-01'")
+
+
+def test_cast_nested_field() -> None:
+    """CAST should work with dotted column references."""
+    expected = EqualTo(
+        UnboundTransform(Reference("event.timestamp"), DayTransform()),
+        StringLiteral("2024-01-01"),
+    )
+    assert expected == parser.parse("CAST(event.timestamp AS date) = '2024-01-01'")
+
+
+def test_cast_unsupported_type() -> None:
+    with pytest.raises(ValueError, match="Unsupported CAST target type"):
+        parser.parse("CAST(col AS foobar) = 5")
+
+
+def test_cast_not_equal() -> None:
+    expected = NotEqualTo(
+        UnboundTransform(Reference("ts"), YearTransform()),
+        LongLiteral(2020),
+    )
+    assert expected == parser.parse("CAST(ts AS year) != 2020")
+
+
+def test_cast_less_than_or_equal() -> None:
+    expected = LessThanOrEqual(
+        UnboundTransform(Reference("ts"), MonthTransform()),
+        LongLiteral(6),
+    )
+    assert expected == parser.parse("CAST(ts AS month) <= 6")
+
+
+def test_cast_with_and() -> None:
+    """CAST predicates compose with AND/OR via infix_notation."""
+    result = parser.parse("CAST(ts AS date) = '2024-01-01' and status = 'active'")
+    assert isinstance(result, And)
+
+
+def test_cast_with_not() -> None:
+    """NOT should negate a CAST predicate."""
+    result = parser.parse("not CAST(ts AS date) = '2024-01-01'")
+    assert isinstance(result, Not)