Use canonical manifest records in Avro tests

Project records to versioned writer schemas by Iceberg field ID so DataFile keeps one stable in-memory layout while v2 field ordering remains schema-correct.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Author: kevinjqliu

pyiceberg/manifest.py

@@ -541,21 +541,19 @@ def sort_order_id(self) -> int | None:
 
     @property
     def first_row_id(self) -> int | None:
-        return self._data[16] if len(self._data) >= len(DATA_FILE_TYPE[3].fields) else None
+        return self._data[16]
 
     @property
     def referenced_data_file(self) -> str | None:
-        if len(self._data) == len(DATA_FILE_TYPE[2].fields):
-            return self._data[16]
-        return self._data[17] if len(self._data) >= len(DATA_FILE_TYPE[3].fields) else None
+        return self._data[17]
 
     @property
     def content_offset(self) -> int | None:
-        return self._data[18] if len(self._data) >= len(DATA_FILE_TYPE[3].fields) else None
+        return self._data[18]
 
     @property
     def content_size_in_bytes(self) -> int | None:
-        return self._data[19] if len(self._data) >= len(DATA_FILE_TYPE[3].fields) else None
+        return self._data[19]
 
     # Spec ID should not be stored in the file
     _spec_id: int

tests/avro/test_file.py

@@ -155,7 +155,6 @@ def test_write_manifest_entry_with_iceberg_read_with_fastavro_v1() -> None:
 
 def test_write_v2_manifest_entry_with_fastavro() -> None:
     data_file = DataFile.from_args(
-        _table_format_version=2,
         content=DataFileContent.POSITION_DELETES,
         file_path="s3://some-path/delete-file.parquet",
         file_format=FileFormat.PARQUET,
@@ -174,13 +173,8 @@ def test_write_v2_manifest_entry_with_fastavro() -> None:
         sort_order_id=4,
         referenced_data_file="s3://some-path/data-file.parquet",
     )
-    assert data_file.first_row_id is None
-    assert data_file.referenced_data_file == "s3://some-path/data-file.parquet"
-    assert data_file.content_offset is None
-    assert data_file.content_size_in_bytes is None
 
     entry = ManifestEntry.from_args(
-        _table_format_version=2,
         status=ManifestEntryStatus.ADDED,
         snapshot_id=8638475580105682862,
         sequence_number=0,
@@ -197,6 +191,7 @@ def test_write_v2_manifest_entry_with_fastavro() -> None:
             output_file=PyArrowFileIO().new_output(tmp_avro_file),
             file_schema=MANIFEST_ENTRY_SCHEMAS[2],
             schema_name="manifest_entry",
+            record_schema=MANIFEST_ENTRY_SCHEMAS[3],
             metadata=additional_metadata,
         ) as out:
             out.write_block([entry])
@@ -212,7 +207,15 @@ def test_write_v2_manifest_entry_with_fastavro() -> None:
 
             fa_entry = next(it)
 
-        assert record_to_fastavro(entry, MANIFEST_ENTRY_SCHEMAS[2].as_struct()) == fa_entry
+        assert fa_entry["data_file"]["referenced_data_file"] == data_file.referenced_data_file
+        assert (
+            record_to_fastavro(
+                entry,
+                record_struct=MANIFEST_ENTRY_SCHEMAS[3].as_struct(),
+                file_struct=MANIFEST_ENTRY_SCHEMAS[2].as_struct(),
+            )
+            == fa_entry
+        )
 
 
 @pytest.mark.parametrize("format_version", [1, 2])

tests/conftest.py

@@ -106,17 +106,28 @@
     from pyiceberg.io.pyarrow import PyArrowFileIO
 
 
-def record_to_fastavro(obj: Any, struct: StructType | None = None) -> Any:
+def record_to_fastavro(
+    obj: Any,
+    record_struct: StructType | None = None,
+    file_struct: StructType | None = None,
+) -> Any:
     """Convert a manifest record to the representation returned by FastAvro."""
     if isinstance(obj, Record):
-        if struct is not None:
-            return {
-                field.name: record_to_fastavro(
+        if record_struct is not None:
+            record_positions = {field.field_id: pos for pos, field in enumerate(record_struct.fields)}
+            result = {}
+            for file_field in (file_struct or record_struct).fields:
+                if (pos := record_positions.get(file_field.field_id)) is None:
+                    result[file_field.name] = file_field.write_default
+                    continue
+
+                record_field = record_struct.fields[pos]
+                result[file_field.name] = record_to_fastavro(
                     obj[pos],
-                    field.field_type if isinstance(field.field_type, StructType) else None,
+                    record_field.field_type if isinstance(record_field.field_type, StructType) else None,
+                    file_field.field_type if isinstance(file_field.field_type, StructType) else None,
                 )
-                for pos, field in enumerate(struct.fields)
-            }
+            return result
         return {
             key: record_to_fastavro(value)
             for key, value in inspect.getmembers(obj)

tests/integration/test_rest_manifest.py

@@ -26,8 +26,6 @@
 from pyiceberg.catalog import Catalog, load_catalog
 from pyiceberg.io.pyarrow import PyArrowFileIO
 from pyiceberg.manifest import (
-    DataFile,
-    ManifestEntry,
     data_file_with_partition,
     manifest_entry_schema_with_data_file,
     write_manifest,
@@ -66,38 +64,15 @@ def test_write_sample_manifest(table_test_all_types: Table, compression: AvroCom
     entry = test_manifest_entries[0]
     test_schema = table_test_all_types.schema()
     test_spec = table_test_all_types.spec()
-    data_file_v2 = DataFile.from_args(
-        _table_format_version=2,
-        content=entry.data_file.content,
-        file_path=entry.data_file.file_path,
-        file_format=entry.data_file.file_format,
-        partition=entry.data_file.partition,
-        record_count=entry.data_file.record_count,
-        file_size_in_bytes=entry.data_file.file_size_in_bytes,
-        column_sizes=entry.data_file.column_sizes,
-        value_counts=entry.data_file.value_counts,
-        null_value_counts=entry.data_file.null_value_counts,
-        nan_value_counts=entry.data_file.nan_value_counts,
-        lower_bounds=entry.data_file.lower_bounds,
-        upper_bounds=entry.data_file.upper_bounds,
-        key_metadata=entry.data_file.key_metadata,
-        split_offsets=entry.data_file.split_offsets,
-        equality_ids=entry.data_file.equality_ids,
-        sort_order_id=entry.data_file.sort_order_id,
-        referenced_data_file=entry.data_file.referenced_data_file,
-    )
-    entry_v2 = ManifestEntry.from_args(
-        _table_format_version=2,
-        status=entry.status,
-        snapshot_id=entry.snapshot_id,
-        sequence_number=entry.sequence_number,
-        file_sequence_number=entry.file_sequence_number,
-        data_file=data_file_v2,
+    data_file_v3_type = data_file_with_partition(
+        partition_type=test_spec.partition_type(test_schema),
+        format_version=3,
     )
     data_file_v2_type = data_file_with_partition(
         partition_type=test_spec.partition_type(test_schema),
         format_version=2,
     )
+    entry_v3_type = manifest_entry_schema_with_data_file(format_version=3, data_file=data_file_v3_type).as_struct()
     entry_v2_type = manifest_entry_schema_with_data_file(format_version=2, data_file=data_file_v2_type).as_struct()
 
     with TemporaryDirectory() as tmpdir:
@@ -119,4 +94,4 @@ def test_write_sample_manifest(table_test_all_types: Table, compression: AvroCom
             it = iter(r)
             fa_entry = next(it)
 
-            assert fa_entry == record_to_fastavro(entry_v2, entry_v2_type)
+            assert fa_entry == record_to_fastavro(entry, record_struct=entry_v3_type, file_struct=entry_v2_type)