rebase

Author: gabeiglio

pyiceberg/partitioning.py

@@ -32,7 +32,7 @@
     model_validator,
 )
 
-from pyiceberg.schema import Schema
+from pyiceberg.schema import Schema, index_by_id
 from pyiceberg.transforms import (
     BucketTransform,
     DayTransform,
@@ -56,6 +56,7 @@
     TimestampType,
     TimestamptzType,
     TimeType,
+    UnknownType,
     UUIDType,
 )
 from pyiceberg.utils.datetime import date_to_days, datetime_to_micros, time_to_micros
@@ -222,11 +223,13 @@ def partition_type(self, schema: Schema) -> StructType:
         :return: A StructType that represents the PartitionSpec, with a NestedField for each PartitionField.
         """
         nested_fields = []
+        schema_ids = index_by_id(schema)
         for field in self.fields:
-            source_type = schema.find_type(field.source_id)
-            result_type = field.transform.result_type(source_type)
-            required = schema.find_field(field.source_id).required
-            nested_fields.append(NestedField(field.field_id, field.name, result_type, required=required))
+            if source_field := schema_ids.get(field.source_id):
+                result_type = field.transform.result_type(source_field.field_type)
+                nested_fields.append(NestedField(field.field_id, field.name, result_type, required=source_field.required))
+            else:
+                nested_fields.append(NestedField(field.field_id, field.name, UnknownType()))
         return StructType(*nested_fields)
 
     def partition_to_path(self, data: Record, schema: Schema) -> str:

tests/integration/test_reads.py

@@ -1083,3 +1083,19 @@ def test_filter_after_arrow_scan(catalog: Catalog) -> None:
 
     scan = scan.filter("ts >= '2023-03-05T00:00:00+00:00'")
     assert len(scan.to_arrow()) > 0
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("catalog", [pytest.lazy_fixture("session_catalog")])
+def test_scan_source_field_missing_in_spec(catalog: Catalog, spark: SparkSession) -> None:
+    identifier = "default.test_dropped_field"
+    spark.sql(f"DROP TABLE IF EXISTS {identifier}")
+    spark.sql(f"CREATE TABLE {identifier} (foo int, bar int, jaz string) USING ICEBERG PARTITIONED BY (foo, bar)")
+    spark.sql(
+        f"INSERT INTO {identifier} (foo, bar, jaz) VALUES (1, 1, 'dummy data'), (1, 2, 'dummy data again'), (2, 1, 'another partition')"
+    )
+    spark.sql(f"ALTER TABLE {identifier} DROP PARTITION FIELD foo")
+    spark.sql(f"ALTER TABLE {identifier} DROP COLUMN  foo")
+
+    table = catalog.load_table(identifier)
+    assert len(list(table.scan().plan_files())) == 3

tests/table/test_partitioning.py

@@ -47,6 +47,7 @@
     TimestampType,
     TimestamptzType,
     TimeType,
+    UnknownType,
     UUIDType,
 )
 
@@ -178,6 +179,19 @@ def test_partition_type(table_schema_simple: Schema) -> None:
     )
 
 
+def test_partition_type_missing_source_field(table_schema_simple: Schema) -> None:
+    spec = PartitionSpec(
+        PartitionField(source_id=1, field_id=1000, transform=TruncateTransform(width=19), name="str_truncate"),
+        PartitionField(source_id=10, field_id=1001, transform=BucketTransform(num_buckets=25), name="int_bucket"),
+        spec_id=3,
+    )
+
+    assert spec.partition_type(table_schema_simple) == StructType(
+        NestedField(field_id=1000, name="str_truncate", field_type=StringType(), required=False),
+        NestedField(field_id=1001, name="int_bucket", field_type=UnknownType(), required=False),
+    )
+
+
 @pytest.mark.parametrize(
     "source_type, value",
     [