fix adding lt literal and allow boundreference in _to_unbound_term

Author: jaimeferj

pyiceberg/expressions/__init__.py

@@ -33,7 +33,7 @@
 )
 from typing import Literal as TypingLiteral
 
-from pydantic import ConfigDict, Field, field_serializer
+from pydantic import ConfigDict, Field, field_serializer, field_validator
 
 from pyiceberg.expressions.literals import (
     AboveMax,
@@ -52,8 +52,14 @@
     ConfigDict = dict
 
 
-def _to_unbound_term(term: Union[str, UnboundTerm[Any]]) -> UnboundTerm[Any]:
-    return Reference(term) if isinstance(term, str) else term
+def _to_unbound_term(term: Union[str, UnboundTerm[Any], BoundReference[Any]]) -> UnboundTerm[Any]:
+    if isinstance(term, str):
+        return Reference(term)
+    if isinstance(term, UnboundTerm):
+        return term
+    if isinstance(term, BoundReference):
+        return Reference(term.field.name)
+    raise ValueError(f"Expected UnboundTerm | BoundReference | str, got {type(term).__name__}")
 
 
 def _to_literal_set(values: Union[Iterable[L], Iterable[Literal[L]]]) -> Set[Literal[L]]:
@@ -744,18 +750,28 @@ def as_bound(self) -> Type[BoundNotIn[L]]:
 
 
 class LiteralPredicate(IcebergBaseModel, UnboundPredicate[L], ABC):
-    type: TypingLiteral["lt-eq", "gt", "gt-eq", "eq", "not-eq", "starts-with", "not-starts-with"] = Field(alias="type")
-    term: Term[L]
+    type: TypingLiteral["lt", "lt-eq", "gt", "gt-eq", "eq", "not-eq", "starts-with", "not-starts-with"] = Field(alias="type")
+    term: UnboundTerm[L]
     literal: Literal[L] = Field(serialization_alias="value")
 
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
-    def __init__(self, term: Union[str, UnboundTerm[Any]], literal: Union[L, Literal[L]]):  # pylint: disable=W0621
-        super().__init__(term=_to_unbound_term(term), literal=_to_literal(literal))
-
-    @field_serializer("term")
-    def ser_term(self, v: Term[L]) -> str:
-        return v.name
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        if args:
+            if len(args) != 2:
+                raise TypeError("Expected (term, literal)")
+            kwargs = {"term": args[0], "literal": args[1], **kwargs}
+        super().__init__(**kwargs)
+
+    @field_validator("term", mode="before")
+    @classmethod
+    def _coerce_term(cls, v: Any) -> UnboundTerm[Any]:
+        return _to_unbound_term(v)
+
+    @field_validator("literal", mode="before")
+    @classmethod
+    def _coerce_literal(cls, v: Union[L, Literal[L]]) -> Literal[L]:
+        return _to_literal(v)
 
     @field_serializer("literal")
     def ser_literal(self, literal: Literal[L]) -> str: