fix: use nulls-first ordering for residual comparisons

Author: kevinjqliu

pyiceberg/expressions/visitors.py

@@ -1874,23 +1874,21 @@ def visit_is_nan(self, term: BoundTerm) -> BooleanExpression:
 
     def visit_not_nan(self, term: BoundTerm) -> BooleanExpression:
         val = term.eval(self.struct)
-        # Mirror _ExpressionEvaluator.visit_not_nan (val == val): only NaN fails not-NaN.
-        # A null (and any non-float value) is not NaN, so the predicate holds.
         if isinstance(val, SupportsFloat) and math.isnan(val):
             return self.visit_false()
         else:
             return self.visit_true()
 
     def visit_less_than(self, term: BoundTerm, literal: LiteralValue) -> BooleanExpression:
         value = term.eval(self.struct)
-        if value is not None and value < literal.value:
+        if value is None or value < literal.value:
             return self.visit_true()
         else:
             return self.visit_false()
 
     def visit_less_than_or_equal(self, term: BoundTerm, literal: LiteralValue) -> BooleanExpression:
         value = term.eval(self.struct)
-        if value is not None and value <= literal.value:
+        if value is None or value <= literal.value:
             return self.visit_true()
         else:
             return self.visit_false()

tests/expressions/test_residual_evaluator.py

@@ -236,18 +236,19 @@ def test_is_not_nan() -> None:
     assert residual == AlwaysTrue()
 
 
-def test_comparison_residual_with_null_partition_value() -> None:
-    # Regression test for https://github.com/apache/iceberg-python/issues/3498
-    # A nullable identity-partitioned column whose partition value is None must not raise a
-    # TypeError when compared against a literal; it should behave like row evaluation, where a
-    # null value never satisfies an ordering predicate.
-    schema = Schema(NestedField(50, "x", IntegerType(), required=False), NestedField(51, "hour", IntegerType()))
+def test_comparison_residuals_for_null_identity_partition() -> None:
+    schema = Schema(NestedField(50, "x", IntegerType(), required=False))
     spec = PartitionSpec(PartitionField(50, 1050, IdentityTransform(), "x_part"))
 
-    for predicate in (LessThan("x", 1), LessThanOrEqual("x", 1), GreaterThan("x", 1), GreaterThanOrEqual("x", 1)):
+    for predicate, expected in (
+        (LessThan("x", 1), AlwaysTrue()),
+        (LessThanOrEqual("x", 1), AlwaysTrue()),
+        (GreaterThan("x", 1), AlwaysFalse()),
+        (GreaterThanOrEqual("x", 1), AlwaysFalse()),
+    ):
         res_eval = residual_evaluator_of(spec=spec, expr=predicate, case_sensitive=True, schema=schema)
         residual = res_eval.residual_for(Record(None))
-        assert residual == AlwaysFalse(), f"null partition value should not match {predicate}"
+        assert residual == expected
 
 
 def test_not_in_timestamp() -> None: