Avro: Fix tests and add missing content header

While working on #2004 I've noticed some small discrepancies
that I think would be good to address in a separate PR.

Author: Fokko

pyiceberg/manifest.py

@@ -1279,6 +1279,7 @@ def __init__(
                 "parent-snapshot-id": str(parent_snapshot_id) if parent_snapshot_id is not None else "null",
                 "sequence-number": str(sequence_number),
                 "format-version": "2",
+                "content": "data",
                 AVRO_CODEC_KEY: compression,
             },
         )

tests/conftest.py

@@ -47,6 +47,7 @@
 import boto3
 import pytest
 from moto import mock_aws
+from pydantic_core import to_json
 
 from pyiceberg.catalog import Catalog, load_catalog
 from pyiceberg.catalog.noop import NoopCatalog
@@ -67,10 +68,12 @@
 )
 from pyiceberg.io.fsspec import FsspecFileIO
 from pyiceberg.manifest import DataFile, FileFormat
+from pyiceberg.partitioning import PartitionField, PartitionSpec
 from pyiceberg.schema import Accessor, Schema
 from pyiceberg.serializers import ToOutputFile
 from pyiceberg.table import FileScanTask, Table
 from pyiceberg.table.metadata import TableMetadataV1, TableMetadataV2
+from pyiceberg.transforms import DayTransform, IdentityTransform
 from pyiceberg.types import (
     BinaryType,
     BooleanType,
@@ -1858,15 +1861,40 @@ def simple_map() -> MapType:
 
 
 @pytest.fixture(scope="session")
-def generated_manifest_entry_file(avro_schema_manifest_entry: Dict[str, Any]) -> Generator[str, None, None]:
+def test_schema() -> Schema:
+    return Schema(
+        NestedField(1, "VendorID", IntegerType(), False), NestedField(2, "tpep_pickup_datetime", TimestampType(), False)
+    )
+
+
+@pytest.fixture(scope="session")
+def test_partition_spec() -> Schema:
+    return PartitionSpec(
+        PartitionField(1, 1000, IdentityTransform(), "VendorID"),
+        PartitionField(2, 1001, DayTransform(), "tpep_pickup_datetime"),
+    )
+
+
+@pytest.fixture(scope="session")
+def generated_manifest_entry_file(
+    avro_schema_manifest_entry: Dict[str, Any], test_schema: Schema, test_partition_spec: PartitionSpec
+) -> Generator[str, None, None]:
     from fastavro import parse_schema, writer
 
     parsed_schema = parse_schema(avro_schema_manifest_entry)
 
     with TemporaryDirectory() as tmpdir:
         tmp_avro_file = tmpdir + "/manifest.avro"
         with open(tmp_avro_file, "wb") as out:
-            writer(out, parsed_schema, manifest_entry_records)
+            writer(
+                out,
+                parsed_schema,
+                manifest_entry_records,
+                metadata={
+                    "schema": test_schema.model_dump_json(),
+                    "partition-spec": to_json(test_partition_spec.fields).decode("utf-8"),
+                },
+            )
         yield tmp_avro_file
 
 

tests/utils/test_manifest.py

@@ -38,10 +38,9 @@
     write_manifest,
     write_manifest_list,
 )
-from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC, PartitionField, PartitionSpec
+from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC, PartitionSpec
 from pyiceberg.schema import Schema
 from pyiceberg.table.snapshots import Operation, Snapshot, Summary
-from pyiceberg.transforms import IdentityTransform
 from pyiceberg.typedef import Record, TableVersion
 from pyiceberg.types import IntegerType, NestedField
 
@@ -363,6 +362,8 @@ def test_write_manifest(
     generated_manifest_file_file_v1: str,
     generated_manifest_file_file_v2: str,
     format_version: TableVersion,
+    test_schema: Schema,
+    test_partition_spec: PartitionSpec,
     compression: AvroCompressionCodec,
 ) -> None:
     io = load_file_io()
@@ -376,20 +377,12 @@ def test_write_manifest(
     )
     demo_manifest_file = snapshot.manifests(io)[0]
     manifest_entries = demo_manifest_file.fetch_manifest_entry(io)
-    test_schema = Schema(
-        NestedField(1, "VendorID", IntegerType(), False), NestedField(2, "tpep_pickup_datetime", IntegerType(), False)
-    )
-    test_spec = PartitionSpec(
-        PartitionField(source_id=1, field_id=1, transform=IdentityTransform(), name="VendorID"),
-        PartitionField(source_id=2, field_id=2, transform=IdentityTransform(), name="tpep_pickup_datetime"),
-        spec_id=demo_manifest_file.partition_spec_id,
-    )
     with TemporaryDirectory() as tmpdir:
         tmp_avro_file = tmpdir + "/test_write_manifest.avro"
         output = io.new_output(tmp_avro_file)
         with write_manifest(
             format_version=format_version,
-            spec=test_spec,
+            spec=test_partition_spec,
             schema=test_schema,
             output_file=output,
             snapshot_id=8744736658442914487,
@@ -404,7 +397,7 @@ def test_write_manifest(
 
         expected_metadata = {
             "schema": test_schema.model_dump_json(),
-            "partition-spec": """[{"source-id":1,"field-id":1,"transform":"identity","name":"VendorID"},{"source-id":2,"field-id":2,"transform":"identity","name":"tpep_pickup_datetime"}]""",
+            "partition-spec": """[{"source-id":1,"field-id":1000,"transform":"identity","name":"VendorID"},{"source-id":2,"field-id":1001,"transform":"day","name":"tpep_pickup_datetime"}]""",
             "partition-spec-id": str(demo_manifest_file.partition_spec_id),
             "format-version": str(format_version),
         }