Fix v2 referenced data file schema

pyiceberg/manifest.py

@@ -306,6 +306,13 @@ def __repr__(self) -> str:
             required=False,
             doc="ID representing sort order for this file",
         ),
+        NestedField(
+            field_id=143,
+            name="referenced_data_file",
+            field_type=StringType(),
+            required=False,
+            doc="Fully qualified location (URI with FS scheme) of a data file that all deletes reference",
+        ),
     ),
     3: StructType(
         NestedField(

tests/avro/test_file.py

@@ -14,18 +14,16 @@
 #  KIND, either express or implied.  See the License for the
 #  specific language governing permissions and limitations
 #  under the License.
-import inspect
 from _decimal import Decimal
 from datetime import datetime
-from enum import Enum
 from tempfile import TemporaryDirectory
-from typing import Any
 from uuid import UUID
 
 import pytest
 from fastavro import reader, writer
 
 import pyiceberg.avro.file as avro
+from conftest import record_to_fastavro
 from pyiceberg.avro.codecs.deflate import DeflateCodec
 from pyiceberg.avro.file import AvroFileHeader
 from pyiceberg.io.pyarrow import PyArrowFileIO
@@ -87,24 +85,6 @@ def test_missing_schema() -> None:
     assert "No schema found in Avro file headers" in str(exc_info.value)
 
 
-# helper function to serialize our objects to dicts to enable
-# direct comparison with the dicts returned by fastavro
-def todict(obj: Any) -> Any:
-    if isinstance(obj, dict):
-        data = []
-        for k, v in obj.items():
-            data.append({"key": k, "value": v})
-        return data
-    elif isinstance(obj, Enum):
-        return obj.value
-    elif hasattr(obj, "__iter__") and not isinstance(obj, str) and not isinstance(obj, bytes):
-        return [todict(v) for v in obj]
-    elif isinstance(obj, Record):
-        return {key: todict(value) for key, value in inspect.getmembers(obj) if not callable(value) and not key.startswith("_")}
-    else:
-        return obj
-
-
 def test_write_manifest_entry_with_iceberg_read_with_fastavro_v1() -> None:
     data_file = DataFile.from_args(
         content=DataFileContent.DATA,
@@ -157,7 +137,7 @@ def test_write_manifest_entry_with_iceberg_read_with_fastavro_v1() -> None:
 
             fa_entry = next(it)
 
-        v2_entry = todict(entry)
+        v2_entry = record_to_fastavro(entry)
 
         # These are not written in V1
         del v2_entry["sequence_number"]
@@ -173,10 +153,10 @@ def test_write_manifest_entry_with_iceberg_read_with_fastavro_v1() -> None:
         assert v2_entry == fa_entry
 
 
-def test_write_manifest_entry_with_iceberg_read_with_fastavro_v2() -> None:
+def test_write_v2_manifest_entry_with_fastavro() -> None:
     data_file = DataFile.from_args(
-        content=DataFileContent.DATA,
-        file_path="s3://some-path/some-file.parquet",
+        content=DataFileContent.POSITION_DELETES,
+        file_path="s3://some-path/delete-file.parquet",
         file_format=FileFormat.PARQUET,
         partition=Record(),
         record_count=131327,
@@ -191,7 +171,9 @@ def test_write_manifest_entry_with_iceberg_read_with_fastavro_v2() -> None:
         split_offsets=[4, 133697593],
         equality_ids=[],
         sort_order_id=4,
+        referenced_data_file="s3://some-path/data-file.parquet",
     )
+
     entry = ManifestEntry.from_args(
         status=ManifestEntryStatus.ADDED,
         snapshot_id=8638475580105682862,
@@ -209,6 +191,7 @@ def test_write_manifest_entry_with_iceberg_read_with_fastavro_v2() -> 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])
@@ -224,11 +207,15 @@ def test_write_manifest_entry_with_iceberg_read_with_fastavro_v2() -> None:
 
             fa_entry = next(it)
 
-        v2_entry = todict(entry)
-        for field in ("first_row_id", "referenced_data_file", "content_offset", "content_size_in_bytes"):
-            del v2_entry["data_file"][field]
-
-        assert v2_entry == 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])
@@ -266,7 +253,7 @@ def test_write_manifest_entry_with_fastavro_read_with_iceberg(format_version: Ta
         schema = AvroSchemaConversion().iceberg_to_avro(MANIFEST_ENTRY_SCHEMAS[format_version], schema_name="manifest_entry")
 
         with open(tmp_avro_file, "wb") as out:
-            writer(out, schema, [todict(entry)])
+            writer(out, schema, [record_to_fastavro(entry)])
 
         # Read as V2
         with avro.AvroFile[ManifestEntry](

tests/conftest.py

@@ -25,13 +25,15 @@
 retrieved using `request.getfixturevalue(fixture_name)`.
 """
 
+import inspect
 import os
 import re
 import string
 import time
 import uuid
-from collections.abc import Generator
+from collections.abc import Generator, Mapping
 from datetime import date, datetime, timezone
+from enum import Enum
 from pathlib import Path
 from random import choice, randint
 from tempfile import TemporaryDirectory
@@ -72,7 +74,7 @@
 from pyiceberg.table.metadata import TableMetadataV1, TableMetadataV2, TableMetadataV3
 from pyiceberg.table.sorting import NullOrder, SortField, SortOrder
 from pyiceberg.transforms import DayTransform, IdentityTransform
-from pyiceberg.typedef import Identifier
+from pyiceberg.typedef import Identifier, Record
 from pyiceberg.types import (
     BinaryType,
     BooleanType,
@@ -103,6 +105,43 @@
 
     from pyiceberg.io.pyarrow import PyArrowFileIO
 
+
+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 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],
+                    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,
+                )
+            return result
+        return {
+            key: record_to_fastavro(value)
+            for key, value in inspect.getmembers(obj)
+            if not callable(value) and not key.startswith("_")
+        }
+    if isinstance(obj, Mapping):
+        return [{"key": key, "value": value} for key, value in obj.items()]
+    if isinstance(obj, Enum):
+        return obj.value
+    if hasattr(obj, "__iter__") and not isinstance(obj, str) and not isinstance(obj, bytes):
+        return [record_to_fastavro(value) for value in obj]
+    return obj
+
+
 # Markers for suites that run separately from the unit tests
 NON_UNIT_TEST_MARKERS = {"integration", "s3", "adls", "gcs", "notebook", "benchmark"}
 

tests/integration/test_rest_manifest.py

@@ -16,46 +16,21 @@
 # under the License.
 # pylint:disable=redefined-outer-name
 
-import inspect
-from copy import copy
-from enum import Enum
 from tempfile import TemporaryDirectory
-from typing import Any
 
 import pytest
 from fastavro import reader
 
+from conftest import record_to_fastavro
 from pyiceberg.avro.codecs import AvroCompressionCodec
 from pyiceberg.catalog import Catalog, load_catalog
 from pyiceberg.io.pyarrow import PyArrowFileIO
-from pyiceberg.manifest import DataFile, write_manifest
+from pyiceberg.manifest import (
+    data_file_with_partition,
+    manifest_entry_schema_with_data_file,
+    write_manifest,
+)
 from pyiceberg.table import Table
-from pyiceberg.typedef import Record
-from pyiceberg.utils.lazydict import LazyDict
-
-
-# helper function to serialize our objects to dicts to enable
-# direct comparison with the dicts returned by fastavro
-def todict(obj: Any, spec_keys: list[str]) -> Any:
-    if type(obj) is Record:
-        return {key: obj[pos] for key, pos in zip(spec_keys, range(len(obj)), strict=True)}
-    if isinstance(obj, dict) or isinstance(obj, LazyDict):
-        data = []
-        for k, v in obj.items():
-            data.append({"key": k, "value": v})
-        return data
-    elif isinstance(obj, Enum):
-        return obj.value
-    elif hasattr(obj, "__iter__") and not isinstance(obj, str) and not isinstance(obj, bytes):
-        return [todict(v, spec_keys) for v in obj]
-    elif hasattr(obj, "__dict__"):
-        return {
-            key: todict(value, spec_keys)
-            for key, value in inspect.getmembers(obj)
-            if not callable(value) and not key.startswith("_")
-        }
-    else:
-        return obj
 
 
 @pytest.fixture()
@@ -89,31 +64,16 @@ 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()
-    wrapped_data_file_v2_debug = DataFile.from_args(
+    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,
-        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,
-        spec_id=entry.data_file.spec_id,
     )
-    wrapped_entry_v2 = copy(entry)
-    wrapped_entry_v2.data_file = wrapped_data_file_v2_debug
-    wrapped_entry_v2_dict = todict(wrapped_entry_v2, [field.name for field in test_spec.fields])
-    for field in ("first_row_id", "referenced_data_file", "content_offset", "content_size_in_bytes"):
-        del wrapped_entry_v2_dict["data_file"][field]
+    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:
         tmp_avro_file = tmpdir + "/test_write_manifest.avro"
@@ -134,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 == wrapped_entry_v2_dict
+            assert fa_entry == record_to_fastavro(entry, record_struct=entry_v3_type, file_struct=entry_v2_type)

tests/utils/test_manifest.py

@@ -228,6 +228,40 @@ def test_fetch_manifest_entry_with_filter(generated_manifest_entry_file: str) ->
     assert len(no_match) == 0
 
 
+def test_read_manifest_entry_v2_referenced_data_file(tmp_path: Path) -> None:
+    io = PyArrowFileIO()
+    manifest_path = str(tmp_path / "manifest.avro")
+    referenced_data_file = "s3://bucket/data.parquet"
+    entry = ManifestEntry.from_args(
+        status=ManifestEntryStatus.ADDED,
+        snapshot_id=25,
+        sequence_number=1,
+        file_sequence_number=1,
+        data_file=DataFile.from_args(
+            content=DataFileContent.POSITION_DELETES,
+            file_path="s3://bucket/deletes.parquet",
+            file_format=FileFormat.PARQUET,
+            partition=Record(),
+            record_count=3,
+            file_size_in_bytes=47,
+            referenced_data_file=referenced_data_file,
+        ),
+    )
+
+    with write_manifest(
+        format_version=2,
+        spec=UNPARTITIONED_PARTITION_SPEC,
+        schema=Schema(NestedField(1, "foo", IntegerType(), required=False)),
+        output_file=io.new_output(manifest_path),
+        snapshot_id=25,
+        avro_compression="null",
+    ) as writer:
+        writer.add_entry(entry)
+
+    manifest = writer.to_manifest_file()
+    assert manifest.fetch_manifest_entry(io)[0].data_file.referenced_data_file == referenced_data_file
+
+
 def test_read_manifest_entry_v3_fields(tmp_path: Path) -> None:
     io = PyArrowFileIO()