add case for column projection

Author: kevinjqliu

pyiceberg/expressions/visitors.py

@@ -897,6 +897,12 @@ def visit_bound_predicate(self, predicate: BoundPredicate[L]) -> BooleanExpressi
         file_column_name = self.file_schema.find_column_name(field.field_id)
 
         if file_column_name is None:
+            # In the case of column projection, the field might not be present in the file schema
+            # If the field has no initial_default, return AlwaysTrue to include all rows
+            # for further evaluation
+            if field.initial_default is None:
+                return AlwaysTrue()
+
             # In the case of schema evolution, the column might not be present
             # we can use the default value as a constant and evaluate it against
             # the predicate

tests/expressions/test_visitors.py

@@ -1799,3 +1799,29 @@ def test_translate_column_names_always_true_false() -> None:
     # Test AlwaysFalse
     translated_false = translate_column_names(AlwaysFalse(), file_schema, case_sensitive=True)
     assert translated_false == AlwaysFalse()
+
+
+def test_translate_column_names_column_projection_missing_field_no_initial_default() -> None:
+    """Test translate_column_names for column projection when field is missing from file schema and has no initial_default."""
+    # Original schema with a field that has no initial_default (defaults to None)
+    original_schema = Schema(
+        NestedField(field_id=1, name="existing_col", field_type=StringType(), required=False),
+        NestedField(field_id=2, name="missing_col", field_type=IntegerType(), required=False),  # No initial_default specified
+        schema_id=1,
+    )
+
+    # Create bound expression for the missing column
+    unbound_expr = EqualTo("missing_col", 42)
+    bound_expr = visit(unbound_expr, visitor=BindVisitor(schema=original_schema, case_sensitive=True))
+
+    # File schema only has the existing column (field_id=1), missing field_id=2
+    file_schema = Schema(
+        NestedField(field_id=1, name="existing_col", field_type=StringType(), required=False),
+        schema_id=1,
+    )
+
+    # Translate column names
+    translated_expr = translate_column_names(bound_expr, file_schema, case_sensitive=True)
+
+    # Should evaluate to AlwaysTrue when field has no initial_default, allowing for further evaluation
+    assert translated_expr == AlwaysTrue()