Allow filtering on newly added columns (#246)

* Allow filtering on newly added columns

Resolves #217

* Thanks Amogh!

Author: Fokko

dev/provision.py

@@ -152,7 +152,6 @@
     """
 )
 
-
 spark.sql(
     """
 DELETE FROM default.test_positional_mor_deletes WHERE number = 9
@@ -315,9 +314,28 @@
 )
 
 spark.sql(
-    f"""
+    """
 INSERT INTO default.test_table_sanitized_character
 VALUES
     ('123')
 """
 )
+
+spark.sql(
+    """
+CREATE TABLE default.test_table_add_column (
+    a string
+)
+USING iceberg
+"""
+)
+
+spark.sql("INSERT INTO default.test_table_add_column VALUES ('1')")
+
+spark.sql(
+    """
+ALTER TABLE default.test_table_add_column ADD COLUMN b string
+"""
+)
+
+spark.sql("INSERT INTO default.test_table_add_column VALUES ('2', '2')")

pyiceberg/expressions/visitors.py

@@ -892,8 +892,13 @@ def visit_unbound_predicate(self, predicate: UnboundPredicate[L]) -> BooleanExpr
     def visit_bound_predicate(self, predicate: BoundPredicate[L]) -> BooleanExpression:
         file_column_name = self.file_schema.find_column_name(predicate.term.ref().field.field_id)
 
-        if not file_column_name:
-            raise ValueError(f"Not found in file schema: {file_column_name}")
+        if file_column_name is None:
+            # In the case of schema evolution, the column might not be present
+            # in the file schema when reading older data
+            if isinstance(predicate, BoundIsNull):
+                return AlwaysTrue()
+            else:
+                return AlwaysFalse()
 
         if isinstance(predicate, BoundUnaryPredicate):
             return predicate.as_unbound(file_column_name)

tests/test_integration.py

@@ -143,6 +143,12 @@ def test_positional_mor_deletes(catalog: Catalog) -> Table:
     return catalog.load_table("default.test_positional_mor_deletes")
 
 
+@pytest.fixture()
+def test_table_add_column(catalog: Catalog) -> Table:
+    """Table that has a new column"""
+    return catalog.load_table("default.test_table_add_column")
+
+
 @pytest.fixture()
 def test_positional_mor_double_deletes(catalog: Catalog) -> Table:
     """Table that has multiple positional deletes"""
@@ -373,6 +379,18 @@ def test_scan_branch(test_positional_mor_deletes: Table) -> None:
     assert arrow_table["number"].to_pylist() == [1, 2, 3, 4, 6, 7, 8, 9, 10, 11, 12]
 
 
+@pytest.mark.integration
+def test_filter_on_new_column(test_table_add_column: Table) -> None:
+    arrow_table = test_table_add_column.scan(row_filter="b == '2'").to_arrow()
+    assert arrow_table["b"].to_pylist() == ['2']
+
+    arrow_table = test_table_add_column.scan(row_filter="b is not null").to_arrow()
+    assert arrow_table["b"].to_pylist() == ['2']
+
+    arrow_table = test_table_add_column.scan(row_filter="b is null").to_arrow()
+    assert arrow_table["b"].to_pylist() == [None]
+
+
 @pytest.mark.integration
 def test_upgrade_table_version(table_test_table_version: Table) -> None:
     assert table_test_table_version.format_version == 1