add test to check projected field predicate evaluator

Author: Roman Shanin

pyiceberg/expressions/visitors.py

@@ -860,7 +860,9 @@ class _ColumnNameTranslator(BooleanExpressionVisitor[BooleanExpression]):
 
     Args:
       file_schema (Schema): The schema of the file.
+      projected_schema (Schema): The schema to project onto the data files.
       case_sensitive (bool): Whether to consider case when binding a reference to a field in a schema, defaults to True.
+      projected_missing_fields(dict[str, Any]): Map of fields missing in file_schema, but present as partition values.
 
     Raises:
         TypeError: In the case of an UnboundPredicate.

tests/expressions/test_visitors.py

@@ -68,10 +68,12 @@
     BooleanExpressionVisitor,
     BoundBooleanExpressionVisitor,
     _ManifestEvalVisitor,
+    bind,
     expression_evaluator,
     expression_to_plain_format,
     rewrite_not,
     rewrite_to_dnf,
+    translate_column_names,
     visit,
     visit_bound_predicate,
 )
@@ -1623,3 +1625,22 @@ def test_expression_evaluator_null() -> None:
     assert expression_evaluator(schema, LessThan("a", 1), case_sensitive=True)(struct) is False
     assert expression_evaluator(schema, StartsWith("a", 1), case_sensitive=True)(struct) is False
     assert expression_evaluator(schema, NotStartsWith("a", 1), case_sensitive=True)(struct) is True
+
+@pytest.mark.parametrize(
+    "before_expression,after_expression",
+    [
+        (In("id", {1, 2, 3}), AlwaysTrue()),
+        (EqualTo("id", 3), AlwaysFalse()),
+        (And(EqualTo("id", 1), EqualTo("all_same_value_or_null", "string")), And(AlwaysTrue(), EqualTo("all_same_value_or_null", "string"))),
+        (And(EqualTo("all_same_value_or_null", "string"), GreaterThan("id", 2)), And(EqualTo("all_same_value_or_null", "string"), AlwaysFalse())),
+        (
+            Or(And(EqualTo("id", 1), EqualTo("all_same_value_or_null", "string")), And(EqualTo("all_same_value_or_null", "string"), GreaterThan("id", 2))),
+            Or(And(AlwaysTrue(), EqualTo("all_same_value_or_null", "string")), And(EqualTo("all_same_value_or_null", "string"), AlwaysFalse())),
+        )
+    ]
+)
+def test_translate_column_names_eval_projected_fields(schema: Schema, before_expression: BooleanExpression, after_expression: BooleanExpression) -> None: 
+    # exclude id from file_schema pretending that it's part of partition values
+    file_schema = Schema(*[field for field in schema.columns if field.name != "id"])
+    projected_missing_fields = { "id": 1 }
+    assert translate_column_names(bind(schema, before_expression, True), file_schema, schema, True, projected_missing_fields) == after_expression