Skip to content

Read partitioned tables with source field missing from schema #2367

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Read partitioned tables with source field missing from schema #2367

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8fcf6d6
Ignore inactive partition fields
gabeiglio Aug 21, 2025
a7b118b
lint, simplify
gabeiglio Aug 21, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 8 additions & 5 deletions pyiceberg/partitioning.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -56,6 +56,7 @@
TimestampType,
TimestamptzType,
TimeType,
UnknownType,
UUIDType,
)
from pyiceberg.utils.datetime import date_to_days, datetime_to_micros, time_to_micros
Expand Down Expand Up @@ -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:
Copy link
Contributor Author

@gabeiglio gabeiglio Aug 21, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Wanted to get some opinions for just allowing this for VoidTransforms fields, as as of now we can drop columns without dropping the partition first

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Unfortunally, dropping is not an option. The V1 tables do not have field-IDs encoded in the struct, and is purely positional based. See the spec for details. Dropping a field, would change the position, potentially resulting in data integrity issues.

Copy link
Contributor Author

@gabeiglio gabeiglio Aug 21, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I was referring more to that if the field's transform is not VoidTransform then we should potentially fail here. Java does not allow dropping a schema column if there is an non-void partition referencing that field, but here it is not guaranteed that if there is no source-field for a partition then the partition will be void.

nested_fields.append(NestedField(field.field_id, field.name, UnknownType()))
return StructType(*nested_fields)

def partition_to_path(self, data: Record, schema: Schema) -> str:
Expand Down
16 changes: 16 additions & 0 deletions tests/integration/test_reads.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1074,3 +1074,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
14 changes: 14 additions & 0 deletions tests/table/test_partitioning.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@
TimestampType,
TimestamptzType,
TimeType,
UnknownType,
UUIDType,
)

Expand Down Expand Up @@ -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",
[
Expand Down