Add an aditional test

Author: Fokko

pyiceberg/expressions/visitors.py

@@ -893,15 +893,28 @@ def visit_unbound_predicate(self, predicate: UnboundPredicate[L]) -> BooleanExpr
         raise TypeError(f"Expected Bound Predicate, got: {predicate.term}")
 
     def visit_bound_predicate(self, predicate: BoundPredicate[L]) -> BooleanExpression:
-        file_column_name = self.file_schema.find_column_name(predicate.term.ref().field.field_id)
+        field = predicate.term.ref().field
+        file_column_name = self.file_schema.find_column_name(field.field_id)
 
         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()
+            # we can use the default value as a constant and evaluate it against
+            # the predicate
+            pred: BooleanExpression
+            if isinstance(predicate, BoundUnaryPredicate):
+                pred = predicate.as_unbound(field.name)
+            elif isinstance(predicate, BoundLiteralPredicate):
+                pred = predicate.as_unbound(field.name, predicate.literal)
+            elif isinstance(predicate, BoundSetPredicate):
+                pred = predicate.as_unbound(field.name, predicate.literals)
             else:
-                return AlwaysFalse()
+                raise ValueError(f"Unsupported predicate: {predicate}")
+
+            return (
+                AlwaysTrue()
+                if expression_evaluator(Schema(field), pred, case_sensitive=self.case_sensitive)(Record(field.initial_default))
+                else AlwaysFalse()
+            )
 
         if isinstance(predicate, BoundUnaryPredicate):
             return predicate.as_unbound(file_column_name)

pyiceberg/io/pyarrow.py

@@ -1820,6 +1820,7 @@ def struct(
                     field_arrays.append(pa.nulls(len(struct_array), type=arrow_type))
                 else:
                     field_arrays.append(pa.repeat(field.initial_default, len(struct_array)))
+                fields.append(self._construct_field(field, arrow_type))
             else:
                 raise ResolveError(f"Field is required, and could not be found in the file: {field}")
 
@@ -2251,7 +2252,7 @@ def parquet_path_to_id_mapping(
     Compute the mapping of parquet column path to Iceberg ID.
 
     For each column, the parquet file metadata has a path_in_schema attribute that follows
-    a specific naming scheme for nested columnds. This function computes a mapping of
+    a specific naming scheme for nested columns. This function computes a mapping of
     the full paths to the corresponding Iceberg IDs.
 
     Args:

tests/integration/test_reads.py

@@ -29,6 +29,7 @@
 from hive_metastore.ttypes import LockRequest, LockResponse, LockState, UnlockRequest
 from pyarrow.fs import S3FileSystem
 from pydantic_core import ValidationError
+from pyspark.sql import SparkSession
 
 from pyiceberg.catalog import Catalog
 from pyiceberg.catalog.hive import HiveCatalog, _HiveClient
@@ -1024,3 +1025,30 @@ def test_scan_with_datetime(catalog: Catalog) -> None:
 
     df = table.scan(row_filter=LessThan("datetime", yesterday)).to_pandas()
     assert len(df) == 0
+
+
+@pytest.mark.integration
+# @pytest.mark.parametrize("catalog", [pytest.lazy_fixture("session_catalog_hive"), pytest.lazy_fixture("session_catalog")])
+@pytest.mark.parametrize("catalog", [pytest.lazy_fixture("session_catalog")])
+def test_initial_default(catalog: Catalog, spark: SparkSession) -> None:
+    identifier = "default.test_initial_default"
+    try:
+        catalog.drop_table(identifier)
+    except NoSuchTableError:
+        pass
+
+    one_column = pa.table([pa.nulls(10, pa.int32())], ["some_field"])
+
+    tbl = catalog.create_table(identifier, schema=one_column.schema, properties={"format-version": "2"})
+
+    tbl.append(one_column)
+
+    # Do the bump version through Spark, since PyIceberg does not support this (yet)
+    spark.sql(f"ALTER TABLE {identifier} SET TBLPROPERTIES('format-version'='3')")
+
+    with tbl.update_schema() as upd:
+        upd.add_column("so_true", BooleanType(), required=False, default_value=True)
+
+    result_table = tbl.scan().filter("so_true == True").to_arrow()
+
+    assert len(result_table) == 10