Fix residual ordered comparisons for null identity partitions

pyiceberg/expressions/visitors.py

@@ -1880,25 +1880,29 @@ def visit_not_nan(self, term: BoundTerm) -> BooleanExpression:
             return self.visit_true()
 
     def visit_less_than(self, term: BoundTerm, literal: LiteralValue) -> BooleanExpression:
-        if term.eval(self.struct) < literal.value:
+        value = term.eval(self.struct)
+        if value is not None and value < literal.value:
             return self.visit_true()
         else:
             return self.visit_false()
 
     def visit_less_than_or_equal(self, term: BoundTerm, literal: LiteralValue) -> BooleanExpression:
-        if term.eval(self.struct) <= literal.value:
+        value = term.eval(self.struct)
+        if value is not None and value <= literal.value:
             return self.visit_true()
         else:
             return self.visit_false()
 
     def visit_greater_than(self, term: BoundTerm, literal: LiteralValue) -> BooleanExpression:
-        if term.eval(self.struct) > literal.value:
+        value = term.eval(self.struct)
+        if value is not None and value > literal.value:
             return self.visit_true()
         else:
             return self.visit_false()
 
     def visit_greater_than_or_equal(self, term: BoundTerm, literal: LiteralValue) -> BooleanExpression:
-        if term.eval(self.struct) >= literal.value:
+        value = term.eval(self.struct)
+        if value is not None and value >= literal.value:
             return self.visit_true()
         else:
             return self.visit_false()

tests/expressions/test_residual_evaluator.py

@@ -21,13 +21,15 @@
     AlwaysFalse,
     AlwaysTrue,
     And,
+    BooleanExpression,
     EqualTo,
     GreaterThan,
     GreaterThanOrEqual,
     In,
     IsNaN,
     IsNull,
     LessThan,
+    LessThanOrEqual,
     NotIn,
     NotNaN,
     NotNull,
@@ -235,6 +237,24 @@ def test_is_not_nan() -> None:
     assert residual == AlwaysTrue()
 
 
+@pytest.mark.parametrize(
+    "predicate",
+    [
+        pytest.param(LessThan("x", 1), id="less-than"),
+        pytest.param(LessThanOrEqual("x", 1), id="less-than-or-equal"),
+        pytest.param(GreaterThan("x", 1), id="greater-than"),
+        pytest.param(GreaterThanOrEqual("x", 1), id="greater-than-or-equal"),
+    ],
+)
+def test_ordered_comparison_residual_for_null_identity_partition(predicate: BooleanExpression) -> None:
+    schema = Schema(NestedField(50, "x", IntegerType(), required=False))
+    spec = PartitionSpec(PartitionField(50, 1050, IdentityTransform(), "x_part"))
+
+    res_eval = residual_evaluator_of(spec=spec, expr=predicate, case_sensitive=True, schema=schema)
+
+    assert res_eval.residual_for(Record(None)) == AlwaysFalse()
+
+
 def test_not_in_timestamp() -> None:
     schema = Schema(NestedField(50, "ts", TimestampType()), NestedField(51, "dateint", IntegerType()))
 

tests/table/test_init.py

@@ -32,6 +32,8 @@
     And,
     EqualTo,
     In,
+    LessThan,
+    Or,
 )
 from pyiceberg.expressions.visitors import bind
 from pyiceberg.io import PY_IO_IMPL, FileIO, load_file_io
@@ -356,6 +358,44 @@ def test_data_scan_plan_files_no_current_snapshot(example_table_metadata_no_snap
     assert len(scan.to_arrow()) == 0
 
 
+def test_data_scan_count_with_less_than_on_null_identity_partition(catalog: Catalog) -> None:
+    import pyarrow as pa
+
+    catalog.create_namespace("default")
+    schema = Schema(
+        NestedField(1, "x", IntegerType(), required=False),
+        NestedField(2, "y", IntegerType(), required=False),
+    )
+    spec = PartitionSpec(PartitionField(1, 1000, IdentityTransform(), "x"))
+    table = catalog.create_table("default.null_identity_partition", schema=schema, partition_spec=spec)
+    table.append(
+        pa.table(
+            {
+                "x": pa.array([None, None], type=pa.int32()),
+                "y": pa.array([0, 2], type=pa.int32()),
+            }
+        )
+    )
+
+    # To exercise the residual evaluator code path, include y == 2 so partition pruning keeps the file.
+    #
+    # Partition pruning:
+    #   x < 1 -> false for the null x partition
+    #   y == 2 -> unknown because y is not partitioned
+    #   false OR unknown -> keep the file
+    #
+    # Residual evaluation:
+    #   x < 1 -> false for the null x partition
+    #   y == 2 -> retained because no partition value is available for y
+    #   false OR y == 2 -> residual is y == 2
+    scan = table.scan(row_filter=Or(LessThan("x", 1), EqualTo("y", 2)))
+    tasks = list(scan.plan_files())
+
+    assert len(tasks) == 1
+    assert tasks[0].residual == EqualTo("y", 2)
+    assert scan.count() == 1  # Only the y == 2 row matches.
+
+
 def test_incremental_append_scan_default(table_v2: Table) -> None:
     scan = table_v2.incremental_append_scan()
     assert scan.row_filter == AlwaysTrue()