Add missing referenced_data_file field to v2 manifests (#3953)

Fix v2 referenced data file schema

Add optional referenced_data_file (field 143) to v2 manifests, retain it in existing expectations, and cover a non-null reference using explicit canonical-record projection.

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

Author: kevinjqliu

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

@@ -208,6 +208,7 @@ def test_write_manifest_entry_with_iceberg_read_with_fastavro_v2() -> None:
         with avro.AvroOutputFile[ManifestEntry](
             output_file=PyArrowFileIO().new_output(tmp_avro_file),
             file_schema=MANIFEST_ENTRY_SCHEMAS[2],
+            record_schema=MANIFEST_ENTRY_SCHEMAS[3],
             schema_name="manifest_entry",
             metadata=additional_metadata,
         ) as out:
@@ -225,12 +226,44 @@ 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"):
+        for field in ("first_row_id", "content_offset", "content_size_in_bytes"):
             del v2_entry["data_file"][field]
 
         assert v2_entry == fa_entry
 
 
+def test_write_v2_referenced_data_file_with_fastavro() -> None:
+    referenced_data_file = "s3://some-path/data-file.parquet"
+    entry = ManifestEntry.from_args(
+        status=ManifestEntryStatus.ADDED,
+        snapshot_id=25,
+        data_file=DataFile.from_args(
+            content=DataFileContent.POSITION_DELETES,
+            file_path="s3://some-path/delete-file.parquet",
+            file_format=FileFormat.PARQUET,
+            partition=Record(),
+            record_count=3,
+            file_size_in_bytes=47,
+            referenced_data_file=referenced_data_file,
+        ),
+    )
+
+    with TemporaryDirectory() as tmpdir:
+        tmp_avro_file = tmpdir + "/manifest_entry.avro"
+        with avro.AvroOutputFile[ManifestEntry](
+            output_file=PyArrowFileIO().new_output(tmp_avro_file),
+            file_schema=MANIFEST_ENTRY_SCHEMAS[2],
+            record_schema=MANIFEST_ENTRY_SCHEMAS[3],
+            schema_name="manifest_entry",
+        ) as out:
+            out.write_block([entry])
+
+        with open(tmp_avro_file, "rb") as fo:
+            fa_entry = next(reader(fo))
+
+        assert fa_entry["data_file"]["referenced_data_file"] == referenced_data_file
+
+
 @pytest.mark.parametrize("format_version", [1, 2])
 def test_write_manifest_entry_with_fastavro_read_with_iceberg(format_version: TableVersion) -> None:
     data_file_dict = {

tests/integration/test_rest_manifest.py

@@ -112,7 +112,7 @@ def test_write_sample_manifest(table_test_all_types: Table, compression: AvroCom
     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"):
+    for field in ("first_row_id", "content_offset", "content_size_in_bytes"):
         del wrapped_entry_v2_dict["data_file"][field]
 
     with TemporaryDirectory() as tmpdir: