From a2c270c9f20fa54716336e1013e4f84db1a6d08e Mon Sep 17 00:00:00 2001 From: "Sung Yun (CODE SIGNING KEY)" Date: Tue, 6 Oct 2026 17:30:24 -0400 Subject: [PATCH 1/2] fix column projection edge case with null identity partition --- pyiceberg/io/pyarrow.py | 8 +++--- tests/io/test_pyarrow.py | 57 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 60 insertions(+), 5 deletions(-) diff --git a/pyiceberg/io/pyarrow.py b/pyiceberg/io/pyarrow.py index 70dd37146b..0cab26309b 100644 --- a/pyiceberg/io/pyarrow.py +++ b/pyiceberg/io/pyarrow.py @@ -1686,9 +1686,7 @@ def _get_column_projection_values( for field_id in missing_field_ids: for partition_field in partition_spec.fields_by_source_id(field_id): if isinstance(partition_field.transform, IdentityTransform): - partition_value = accessors[partition_field.field_id].get(file.partition) - if partition_value is not None: - projected_missing_fields[field_id] = partition_value + projected_missing_fields[field_id] = accessors[partition_field.field_id].get(file.partition) return projected_missing_fields @@ -2130,8 +2128,8 @@ def struct( elif field.optional or field.initial_default is not None: # When an optional field is added, or when a required field with a non-null initial default is added arrow_type = schema_to_pyarrow(field.field_type, include_field_ids=self._include_field_ids) - projected_value = self._projected_missing_fields.get(field.field_id) - if projected_value is not None: + if field.field_id in self._projected_missing_fields: + projected_value = self._projected_missing_fields[field.field_id] field_arrays.append(pa.repeat(pa.scalar(projected_value, type=arrow_type), len(struct_array))) elif field.initial_default is None: field_arrays.append(pa.nulls(len(struct_array), type=arrow_type)) diff --git a/tests/io/test_pyarrow.py b/tests/io/test_pyarrow.py index d665df9a76..d223ad3a69 100644 --- a/tests/io/test_pyarrow.py +++ b/tests/io/test_pyarrow.py @@ -63,7 +63,9 @@ BoundNotStartsWith, BoundReference, BoundStartsWith, + EqualTo, GreaterThan, + IsNull, Not, Or, ) @@ -1495,6 +1497,61 @@ def test_identity_transform_column_projection_with_falsy_value( ) +@pytest.mark.parametrize( + "row_filter, expected_other_field", + [ + (None, ["foo", "bar"]), + (IsNull("partition_col"), ["foo", "bar"]), + (EqualTo("partition_col", 7), []), + ], +) +def test_identity_transform_null_partition_value_takes_precedence_over_initial_default( + tmpdir: str, row_filter: BooleanExpression | None, expected_other_field: list[str] +) -> None: + """A null identity partition value must project as null, not as the field's initial-default.""" + from pyiceberg.expressions.visitors import bind + + table_schema = Schema( + NestedField(1, "other_field", StringType(), required=False), + NestedField(2, "partition_col", IntegerType(), required=False, initial_default=7), + ) + partition_spec = PartitionSpec(PartitionField(2, 1000, IdentityTransform(), "partition_col")) + + arrow_schema = pa.schema([pa.field("other_field", pa.string(), metadata={PYARROW_PARQUET_FIELD_ID_KEY: "1"})]) + file_path = _write_table_to_file( + f"{tmpdir}/null_partition.parquet", arrow_schema, pa.table({"other_field": ["foo", "bar"]}, schema=arrow_schema) + ) + data_file = DataFile.from_args( + content=DataFileContent.DATA, + file_path=file_path, + file_format=FileFormat.PARQUET, + partition=Record(None), + record_count=2, + file_size_in_bytes=os.path.getsize(file_path), + spec_id=partition_spec.spec_id, + ) + + batches = list( + _task_to_record_batches( + PyArrowFileIO(), + FileScanTask(data_file), + bound_row_filter=AlwaysTrue() if row_filter is None else bind(table_schema, row_filter, case_sensitive=True), + projected_schema=table_schema, + table_schema=table_schema, + projected_field_ids={1, 2}, + positional_deletes=None, + case_sensitive=True, + partition_spec=partition_spec, + format_version=3, + ) + ) + + assert [value for batch in batches for value in batch.column("other_field").to_pylist()] == expected_other_field + assert [value for batch in batches for value in batch.column("partition_col").to_pylist()] == [None] * len( + expected_other_field + ) + + @pytest.mark.parametrize("with_field_ids", [False, True], ids=["name-mapping", "field-ids"]) def test_identity_transform_columns_projection(tmp_path: str, catalog: InMemoryCatalog, with_field_ids: bool) -> None: # Test by adding a non-partitioned data file to a multi-partitioned table, verifying partition value From 0776e5f19492e2fdaea566c84b78f3df2cde2e21 Mon Sep 17 00:00:00 2001 From: Kevin Liu Date: Sat, 24 Oct 2026 15:21:21 -0700 Subject: [PATCH 2/2] Cover filter-only null identity partitions and simplify test Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- tests/io/test_pyarrow.py | 43 +++++++++++----------------------------- 1 file changed, 12 insertions(+), 31 deletions(-) diff --git a/tests/io/test_pyarrow.py b/tests/io/test_pyarrow.py index d223ad3a69..0ba8117ef3 100644 --- a/tests/io/test_pyarrow.py +++ b/tests/io/test_pyarrow.py @@ -1497,25 +1497,16 @@ def test_identity_transform_column_projection_with_falsy_value( ) -@pytest.mark.parametrize( - "row_filter, expected_other_field", - [ - (None, ["foo", "bar"]), - (IsNull("partition_col"), ["foo", "bar"]), - (EqualTo("partition_col", 7), []), - ], -) -def test_identity_transform_null_partition_value_takes_precedence_over_initial_default( - tmpdir: str, row_filter: BooleanExpression | None, expected_other_field: list[str] -) -> None: +def test_identity_transform_null_partition_value_takes_precedence_over_initial_default(tmpdir: str) -> None: """A null identity partition value must project as null, not as the field's initial-default.""" - from pyiceberg.expressions.visitors import bind - table_schema = Schema( NestedField(1, "other_field", StringType(), required=False), NestedField(2, "partition_col", IntegerType(), required=False, initial_default=7), ) partition_spec = PartitionSpec(PartitionField(2, 1000, IdentityTransform(), "partition_col")) + table_metadata = TableMetadataV3( + location=str(tmpdir), last_column_id=2, schemas=[table_schema], partition_specs=[partition_spec], next_row_id=0 + ) arrow_schema = pa.schema([pa.field("other_field", pa.string(), metadata={PYARROW_PARQUET_FIELD_ID_KEY: "1"})]) file_path = _write_table_to_file( @@ -1531,25 +1522,15 @@ def test_identity_transform_null_partition_value_takes_precedence_over_initial_d spec_id=partition_spec.spec_id, ) - batches = list( - _task_to_record_batches( - PyArrowFileIO(), - FileScanTask(data_file), - bound_row_filter=AlwaysTrue() if row_filter is None else bind(table_schema, row_filter, case_sensitive=True), - projected_schema=table_schema, - table_schema=table_schema, - projected_field_ids={1, 2}, - positional_deletes=None, - case_sensitive=True, - partition_spec=partition_spec, - format_version=3, - ) - ) + def scan(projected_schema: Schema, row_filter: BooleanExpression) -> pa.Table: + return ArrowScan(table_metadata, PyArrowFileIO(), projected_schema, row_filter).to_table([FileScanTask(data_file)]) - assert [value for batch in batches for value in batch.column("other_field").to_pylist()] == expected_other_field - assert [value for batch in batches for value in batch.column("partition_col").to_pylist()] == [None] * len( - expected_other_field - ) + assert scan(table_schema, AlwaysTrue()).to_pydict() == {"other_field": ["foo", "bar"], "partition_col": [None, None]} + + # Filters on the partition column see null whether or not the column is projected + for projected_schema in (table_schema, Schema(table_schema.fields[0])): + assert scan(projected_schema, IsNull("partition_col"))["other_field"].to_pylist() == ["foo", "bar"] + assert scan(projected_schema, EqualTo("partition_col", 7)).num_rows == 0 @pytest.mark.parametrize("with_field_ids", [False, True], ids=["name-mapping", "field-ids"])