test_manifest.py

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
# pylint: disable=redefined-outer-name,arguments-renamed,fixme
from tempfile import TemporaryDirectory

import fastavro
import pytest

import pyiceberg.manifest as manifest_module
from pyiceberg.avro.codecs import AvroCompressionCodec
from pyiceberg.io import load_file_io
from pyiceberg.io.pyarrow import PyArrowFileIO
from pyiceberg.manifest import (
    DataFile,
    DataFileContent,
    FileFormat,
    ManifestContent,
    ManifestEntry,
    ManifestEntryStatus,
    ManifestFile,
    PartitionFieldSummary,
    _manifests,
    clear_manifest_cache,
    read_manifest_list,
    write_manifest,
    write_manifest_list,
)
from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC, PartitionSpec
from pyiceberg.schema import Schema
from pyiceberg.table.snapshots import Operation, Snapshot, Summary
from pyiceberg.typedef import Record, TableVersion
from pyiceberg.types import IntegerType, NestedField


@pytest.fixture(autouse=True)
def reset_global_manifests_cache() -> None:
    clear_manifest_cache()


def _verify_metadata_with_fastavro(avro_file: str, expected_metadata: dict[str, str]) -> None:
    with open(avro_file, "rb") as f:
        reader = fastavro.reader(f)
        metadata = reader.metadata
        for k, v in expected_metadata.items():
            assert k in metadata
            assert metadata[k] == v


def test_read_manifest_entry(generated_manifest_entry_file: str) -> None:
    manifest = ManifestFile.from_args(
        manifest_path=generated_manifest_entry_file,
        manifest_length=0,
        partition_spec_id=0,
        added_snapshot_id=0,
        sequence_number=0,
        partitions=[],
    )
    manifest_entries = manifest.fetch_manifest_entry(PyArrowFileIO())
    manifest_entry = manifest_entries[0]

    assert manifest_entry.status == ManifestEntryStatus.ADDED
    assert manifest_entry.snapshot_id == 8744736658442914487
    assert manifest_entry.sequence_number == 0
    assert isinstance(manifest_entry.data_file, DataFile)

    data_file = manifest_entry.data_file

    assert data_file.content == DataFileContent.DATA
    assert data_file.file_path == (
        "/home/iceberg/warehouse/nyc/taxis_partitioned/data/VendorID=null/"
        "00000-633-d8a4223e-dc97-45a1-86e1-adaba6e8abd7-00001.parquet"
    )
    assert data_file.file_format == FileFormat.PARQUET
    assert repr(data_file.partition) == "Record[1, 1925]"
    assert data_file.record_count == 19513
    assert data_file.file_size_in_bytes == 388872
    assert data_file.column_sizes == {
        1: 53,
        2: 98153,
        3: 98693,
        4: 53,
        5: 53,
        6: 53,
        7: 17425,
        8: 18528,
        9: 53,
        10: 44788,
        11: 35571,
        12: 53,
        13: 1243,
        14: 2355,
        15: 12750,
        16: 4029,
        17: 110,
        18: 47194,
        19: 2948,
    }
    assert data_file.value_counts == {
        1: 19513,
        2: 19513,
        3: 19513,
        4: 19513,
        5: 19513,
        6: 19513,
        7: 19513,
        8: 19513,
        9: 19513,
        10: 19513,
        11: 19513,
        12: 19513,
        13: 19513,
        14: 19513,
        15: 19513,
        16: 19513,
        17: 19513,
        18: 19513,
        19: 19513,
    }
    assert data_file.null_value_counts == {
        1: 19513,
        2: 0,
        3: 0,
        4: 19513,
        5: 19513,
        6: 19513,
        7: 0,
        8: 0,
        9: 19513,
        10: 0,
        11: 0,
        12: 19513,
        13: 0,
        14: 0,
        15: 0,
        16: 0,
        17: 0,
        18: 0,
        19: 0,
    }
    assert data_file.nan_value_counts == {16: 0, 17: 0, 18: 0, 19: 0, 10: 0, 11: 0, 12: 0, 13: 0, 14: 0, 15: 0}
    assert data_file.lower_bounds == {
        2: b"\x01\x00\x00\x00\x00\x00\x00\x00",
        3: b"\x01\x00\x00\x00\x00\x00\x00\x00",
        7: b"\x03\x00\x00\x00",
        8: b"\x01\x00\x00\x00",
        10: b"\xf6(\\\x8f\xc2\x05S\xc0",
        11: b"\x00\x00\x00\x00\x00\x00\x00\x00",
        13: b"\x00\x00\x00\x00\x00\x00\x00\x00",
        14: b"\x00\x00\x00\x00\x00\x00\xe0\xbf",
        15: b")\\\x8f\xc2\xf5(\x08\xc0",
        16: b"\x00\x00\x00\x00\x00\x00\x00\x00",
        17: b"\x00\x00\x00\x00\x00\x00\x00\x00",
        18: b"\xf6(\\\x8f\xc2\xc5S\xc0",
        19: b"\x00\x00\x00\x00\x00\x00\x04\xc0",
    }
    assert data_file.upper_bounds == {
        2: b"\x06\x00\x00\x00\x00\x00\x00\x00",
        3: b"\x06\x00\x00\x00\x00\x00\x00\x00",
        7: b"\t\x01\x00\x00",
        8: b"\t\x01\x00\x00",
        10: b"\xcd\xcc\xcc\xcc\xcc,_@",
        11: b"\x1f\x85\xebQ\\\xe2\xfe@",
        13: b"\x00\x00\x00\x00\x00\x00\x12@",
        14: b"\x00\x00\x00\x00\x00\x00\xe0?",
        15: b"q=\n\xd7\xa3\xf01@",
        16: b"\x00\x00\x00\x00\x00`B@",
        17: b"333333\xd3?",
        18: b"\x00\x00\x00\x00\x00\x18b@",
        19: b"\x00\x00\x00\x00\x00\x00\x04@",
    }
    assert data_file.key_metadata is None
    assert data_file.split_offsets == [4]
    assert data_file.equality_ids is None
    assert data_file.sort_order_id == 0


def test_read_manifest_list(generated_manifest_file_file_v1: str) -> None:
    input_file = PyArrowFileIO().new_input(generated_manifest_file_file_v1)
    manifest_list = list(read_manifest_list(input_file))[0]

    assert manifest_list.manifest_length == 7989
    assert manifest_list.partition_spec_id == 0
    assert manifest_list.added_snapshot_id == 9182715666859759686
    assert manifest_list.added_files_count == 3
    assert manifest_list.existing_files_count == 0
    assert manifest_list.deleted_files_count == 0

    assert isinstance(manifest_list.partitions, list)

    partitions_summary = manifest_list.partitions[0]
    assert isinstance(partitions_summary, PartitionFieldSummary)

    assert partitions_summary.contains_null is True
    assert partitions_summary.contains_nan is False
    assert partitions_summary.lower_bound == b"\x01\x00\x00\x00"
    assert partitions_summary.upper_bound == b"\x02\x00\x00\x00"

    assert manifest_list.added_rows_count == 237993
    assert manifest_list.existing_rows_count == 0
    assert manifest_list.deleted_rows_count == 0


def test_read_manifest_v1(generated_manifest_file_file_v1: str) -> None:
    io = load_file_io()

    snapshot = Snapshot(
        snapshot_id=25,
        parent_snapshot_id=19,
        timestamp_ms=1602638573590,
        manifest_list=generated_manifest_file_file_v1,
        summary=Summary(Operation.APPEND),
        schema_id=3,
    )
    manifest_list = snapshot.manifests(io)[0]

    assert manifest_list.manifest_length == 7989
    assert manifest_list.partition_spec_id == 0
    assert manifest_list.content == ManifestContent.DATA
    assert manifest_list.sequence_number == 0
    assert manifest_list.min_sequence_number == 0
    assert manifest_list.added_snapshot_id == 9182715666859759686
    assert manifest_list.added_files_count == 3
    assert manifest_list.existing_files_count == 0
    assert manifest_list.deleted_files_count == 0
    assert manifest_list.added_rows_count == 237993
    assert manifest_list.existing_rows_count == 0
    assert manifest_list.deleted_rows_count == 0
    assert manifest_list.key_metadata is None

    assert isinstance(manifest_list.partitions, list)

    partition = manifest_list.partitions[0]

    assert isinstance(partition, PartitionFieldSummary)

    assert partition.contains_null is True
    assert partition.contains_nan is False
    assert partition.lower_bound == b"\x01\x00\x00\x00"
    assert partition.upper_bound == b"\x02\x00\x00\x00"

    entries = manifest_list.fetch_manifest_entry(io)

    assert isinstance(entries, list)

    entry = entries[0]

    assert entry.sequence_number == 0
    assert entry.file_sequence_number == 0
    assert entry.snapshot_id == 8744736658442914487
    assert entry.status == ManifestEntryStatus.ADDED


def test_read_manifest_v2(generated_manifest_file_file_v2: str) -> None:
    io = load_file_io()

    snapshot = Snapshot(
        snapshot_id=25,
        parent_snapshot_id=19,
        timestamp_ms=1602638573590,
        manifest_list=generated_manifest_file_file_v2,
        summary=Summary(Operation.APPEND),
        schema_id=3,
    )
    manifest_list = snapshot.manifests(io)[0]

    assert manifest_list.manifest_length == 7989
    assert manifest_list.partition_spec_id == 0
    assert manifest_list.content == ManifestContent.DELETES
    assert manifest_list.sequence_number == 3
    assert manifest_list.min_sequence_number == 3
    assert manifest_list.added_snapshot_id == 9182715666859759686
    assert manifest_list.added_files_count == 3
    assert manifest_list.existing_files_count == 0
    assert manifest_list.deleted_files_count == 0
    assert manifest_list.added_rows_count == 237993
    assert manifest_list.existing_rows_count == 0
    assert manifest_list.deleted_rows_count == 0
    assert manifest_list.key_metadata is None

    assert isinstance(manifest_list.partitions, list)

    partition = manifest_list.partitions[0]

    assert isinstance(partition, PartitionFieldSummary)

    assert partition.contains_null is True
    assert partition.contains_nan is False
    assert partition.lower_bound == b"\x01\x00\x00\x00"
    assert partition.upper_bound == b"\x02\x00\x00\x00"

    entries = manifest_list.fetch_manifest_entry(io)

    assert isinstance(entries, list)

    entry = entries[0]

    assert entry.sequence_number == 3
    assert entry.file_sequence_number == 3
    assert entry.snapshot_id == 8744736658442914487
    assert entry.status == ManifestEntryStatus.ADDED


def test_read_manifest_cache(generated_manifest_file_file_v2: str) -> None:
    """Test that ManifestFile objects are cached and reused across multiple reads.

    The cache now stores individual ManifestFile objects by their manifest_path,
    rather than caching entire manifest list tuples. This is more memory-efficient
    when multiple manifest lists share overlapping ManifestFile objects.
    """
    io = load_file_io()

    snapshot = Snapshot(
        snapshot_id=25,
        parent_snapshot_id=19,
        timestamp_ms=1602638573590,
        manifest_list=generated_manifest_file_file_v2,
        summary=Summary(Operation.APPEND),
        schema_id=3,
    )

    # Access the manifests property multiple times
    manifests_first_call = snapshot.manifests(io)
    manifests_second_call = snapshot.manifests(io)

    # Ensure that the same manifest list content is returned
    assert manifests_first_call == manifests_second_call

    # Verify that ManifestFile objects are the same instances (cached)
    for mf1, mf2 in zip(manifests_first_call, manifests_second_call, strict=True):
        assert mf1 is mf2, "ManifestFile objects should be the same cached instance"


def test_write_empty_manifest() -> None:
    io = load_file_io()
    test_schema = Schema(NestedField(1, "foo", IntegerType(), False))
    with TemporaryDirectory() as tmpdir:
        tmp_avro_file = tmpdir + "/test_write_manifest.avro"

        with pytest.raises(ValueError, match="An empty manifest file has been written"):
            with write_manifest(
                format_version=1,
                spec=UNPARTITIONED_PARTITION_SPEC,
                schema=test_schema,
                output_file=io.new_output(tmp_avro_file),
                snapshot_id=8744736658442914487,
                avro_compression="deflate",
            ) as _:
                pass


@pytest.mark.parametrize("format_version", [1, 2])
@pytest.mark.parametrize("compression", ["null", "deflate", "zstd"])
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()
    snapshot = Snapshot(
        snapshot_id=25,
        parent_snapshot_id=19,
        timestamp_ms=1602638573590,
        manifest_list=generated_manifest_file_file_v1 if format_version == 1 else generated_manifest_file_file_v2,
        summary=Summary(Operation.APPEND),
        schema_id=3,
    )
    demo_manifest_file = snapshot.manifests(io)[0]
    manifest_entries = demo_manifest_file.fetch_manifest_entry(io)
    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_partition_spec,
            schema=test_schema,
            output_file=output,
            snapshot_id=8744736658442914487,
            avro_compression=compression,
        ) as writer:
            for entry in manifest_entries:
                writer.add_entry(entry)
            new_manifest = writer.to_manifest_file()
            with pytest.raises(RuntimeError):
                # It is already closed
                writer.add_entry(manifest_entries[0])

        expected_metadata = {
            "schema": test_schema.model_dump_json(),
            "partition-spec": (
                '[{"source-id":1,"field-id":1000,"transform":"identity","name":"VendorID"},'
                '{"source-id":2,"field-id":1001,"transform":"day","name":"tpep_pickup_day"}]'
            ),
            "partition-spec-id": str(demo_manifest_file.partition_spec_id),
            "format-version": str(format_version),
        }
        _verify_metadata_with_fastavro(
            tmp_avro_file,
            expected_metadata,
        )
        new_manifest_entries = new_manifest.fetch_manifest_entry(io)

        manifest_entry = new_manifest_entries[0]

        assert manifest_entry.status == ManifestEntryStatus.ADDED
        assert manifest_entry.snapshot_id == 8744736658442914487
        assert manifest_entry.sequence_number == -1 if format_version == 1 else 3
        assert isinstance(manifest_entry.data_file, DataFile)

        data_file = manifest_entry.data_file

        assert data_file.content == DataFileContent.DATA
        assert data_file.file_path == (
            "/home/iceberg/warehouse/nyc/taxis_partitioned/data/VendorID=null/"
            "00000-633-d8a4223e-dc97-45a1-86e1-adaba6e8abd7-00001.parquet"
        )
        assert data_file.file_format == FileFormat.PARQUET
        assert data_file.partition == Record(1, 1925)
        assert data_file.record_count == 19513
        assert data_file.file_size_in_bytes == 388872
        assert data_file.column_sizes == {
            1: 53,
            2: 98153,
            3: 98693,
            4: 53,
            5: 53,
            6: 53,
            7: 17425,
            8: 18528,
            9: 53,
            10: 44788,
            11: 35571,
            12: 53,
            13: 1243,
            14: 2355,
            15: 12750,
            16: 4029,
            17: 110,
            18: 47194,
            19: 2948,
        }
        assert data_file.value_counts == {
            1: 19513,
            2: 19513,
            3: 19513,
            4: 19513,
            5: 19513,
            6: 19513,
            7: 19513,
            8: 19513,
            9: 19513,
            10: 19513,
            11: 19513,
            12: 19513,
            13: 19513,
            14: 19513,
            15: 19513,
            16: 19513,
            17: 19513,
            18: 19513,
            19: 19513,
        }
        assert data_file.null_value_counts == {
            1: 19513,
            2: 0,
            3: 0,
            4: 19513,
            5: 19513,
            6: 19513,
            7: 0,
            8: 0,
            9: 19513,
            10: 0,
            11: 0,
            12: 19513,
            13: 0,
            14: 0,
            15: 0,
            16: 0,
            17: 0,
            18: 0,
            19: 0,
        }
        assert data_file.nan_value_counts == {16: 0, 17: 0, 18: 0, 19: 0, 10: 0, 11: 0, 12: 0, 13: 0, 14: 0, 15: 0}
        assert data_file.lower_bounds == {
            2: b"\x01\x00\x00\x00\x00\x00\x00\x00",
            3: b"\x01\x00\x00\x00\x00\x00\x00\x00",
            7: b"\x03\x00\x00\x00",
            8: b"\x01\x00\x00\x00",
            10: b"\xf6(\\\x8f\xc2\x05S\xc0",
            11: b"\x00\x00\x00\x00\x00\x00\x00\x00",
            13: b"\x00\x00\x00\x00\x00\x00\x00\x00",
            14: b"\x00\x00\x00\x00\x00\x00\xe0\xbf",
            15: b")\\\x8f\xc2\xf5(\x08\xc0",
            16: b"\x00\x00\x00\x00\x00\x00\x00\x00",
            17: b"\x00\x00\x00\x00\x00\x00\x00\x00",
            18: b"\xf6(\\\x8f\xc2\xc5S\xc0",
            19: b"\x00\x00\x00\x00\x00\x00\x04\xc0",
        }
        assert data_file.upper_bounds == {
            2: b"\x06\x00\x00\x00\x00\x00\x00\x00",
            3: b"\x06\x00\x00\x00\x00\x00\x00\x00",
            7: b"\t\x01\x00\x00",
            8: b"\t\x01\x00\x00",
            10: b"\xcd\xcc\xcc\xcc\xcc,_@",
            11: b"\x1f\x85\xebQ\\\xe2\xfe@",
            13: b"\x00\x00\x00\x00\x00\x00\x12@",
            14: b"\x00\x00\x00\x00\x00\x00\xe0?",
            15: b"q=\n\xd7\xa3\xf01@",
            16: b"\x00\x00\x00\x00\x00`B@",
            17: b"333333\xd3?",
            18: b"\x00\x00\x00\x00\x00\x18b@",
            19: b"\x00\x00\x00\x00\x00\x00\x04@",
        }
        assert data_file.key_metadata is None
        assert data_file.split_offsets == [4]
        assert data_file.equality_ids is None
        assert data_file.sort_order_id == 0


@pytest.mark.parametrize("format_version", [1, 2])
@pytest.mark.parametrize("parent_snapshot_id", [19, None])
@pytest.mark.parametrize("compression", ["null", "deflate"])
def test_write_manifest_list(
    generated_manifest_file_file_v1: str,
    generated_manifest_file_file_v2: str,
    format_version: TableVersion,
    parent_snapshot_id: int | None,
    compression: AvroCompressionCodec,
) -> None:
    io = load_file_io()

    snapshot = Snapshot(
        snapshot_id=25,
        parent_snapshot_id=parent_snapshot_id,
        timestamp_ms=1602638573590,
        manifest_list=generated_manifest_file_file_v1 if format_version == 1 else generated_manifest_file_file_v2,
        summary=Summary(Operation.APPEND),
        schema_id=3,
    )

    demo_manifest_list = snapshot.manifests(io)
    with TemporaryDirectory() as tmp_dir:
        path = tmp_dir + "/manifest-list.avro"
        output = io.new_output(path)
        with write_manifest_list(
            format_version=format_version,
            output_file=output,
            snapshot_id=25,
            parent_snapshot_id=parent_snapshot_id,
            sequence_number=0,
            avro_compression=compression,
        ) as writer:
            writer.add_manifests(demo_manifest_list)
        new_manifest_list = list(read_manifest_list(io.new_input(path)))

        if parent_snapshot_id:
            expected_metadata = {"snapshot-id": "25", "parent-snapshot-id": "19", "format-version": str(format_version)}
        else:
            expected_metadata = {"snapshot-id": "25", "parent-snapshot-id": "null", "format-version": str(format_version)}

        if format_version == 2:
            expected_metadata["sequence-number"] = "0"
        _verify_metadata_with_fastavro(path, expected_metadata)

        manifest_file = new_manifest_list[0]

        assert manifest_file.manifest_length == 7989
        assert manifest_file.partition_spec_id == 0
        assert manifest_file.content == ManifestContent.DATA if format_version == 1 else ManifestContent.DELETES
        assert manifest_file.sequence_number == 0 if format_version == 1 else 3
        assert manifest_file.min_sequence_number == 0 if format_version == 1 else 3
        assert manifest_file.added_snapshot_id == 9182715666859759686
        assert manifest_file.added_files_count == 3
        assert manifest_file.existing_files_count == 0
        assert manifest_file.deleted_files_count == 0
        assert manifest_file.added_rows_count == 237993
        assert manifest_file.existing_rows_count == 0
        assert manifest_file.deleted_rows_count == 0
        assert manifest_file.key_metadata is None

        assert isinstance(manifest_file.partitions, list)

        partition = manifest_file.partitions[0]

        assert isinstance(partition, PartitionFieldSummary)

        assert partition.contains_null is True
        assert partition.contains_nan is False
        assert partition.lower_bound == b"\x01\x00\x00\x00"
        assert partition.upper_bound == b"\x02\x00\x00\x00"

        entries = manifest_file.fetch_manifest_entry(io)

        assert isinstance(entries, list)

        entry = entries[0]

        assert entry.sequence_number == 0 if format_version == 1 else 3
        assert entry.file_sequence_number == 0 if format_version == 1 else 3
        assert entry.snapshot_id == 8744736658442914487
        assert entry.status == ManifestEntryStatus.ADDED


@pytest.mark.parametrize(
    "raw_file_format,expected_file_format",
    [
        ("avro", FileFormat("AVRO")),
        ("AVRO", FileFormat("AVRO")),
        ("parquet", FileFormat("PARQUET")),
        ("PARQUET", FileFormat("PARQUET")),
        ("orc", FileFormat("ORC")),
        ("ORC", FileFormat("ORC")),
        ("NOT_EXISTS", None),
    ],
)
def test_file_format_case_insensitive(raw_file_format: str, expected_file_format: FileFormat) -> None:
    if expected_file_format:
        parsed_file_format = FileFormat(raw_file_format)
        assert parsed_file_format == expected_file_format, (
            f"File format {raw_file_format}: {parsed_file_format} != {expected_file_format}"
        )
    else:
        with pytest.raises(ValueError):
            _ = FileFormat(raw_file_format)


def test_manifest_cache_deduplicates_manifest_files() -> None:
    """Test that the manifest cache deduplicates ManifestFile objects across manifest lists.

    This test verifies the fix for https://github.com/apache/iceberg-python/issues/2325

    The issue was that when caching manifest lists by their path, overlapping ManifestFile
    objects were duplicated. For example:
    - ManifestList1: (ManifestFile1)
    - ManifestList2: (ManifestFile1, ManifestFile2)
    - ManifestList3: (ManifestFile1, ManifestFile2, ManifestFile3)

    With the old approach, ManifestFile1 was stored 3 times in the cache.
    With the new approach, ManifestFile objects are cached individually by their
    manifest_path, so ManifestFile1 is stored only once and reused.
    """
    io = PyArrowFileIO()

    with TemporaryDirectory() as tmp_dir:
        # Create three manifest files to simulate manifests created during appends
        manifest1_path = f"{tmp_dir}/manifest1.avro"
        manifest2_path = f"{tmp_dir}/manifest2.avro"
        manifest3_path = f"{tmp_dir}/manifest3.avro"

        schema = Schema(NestedField(field_id=1, name="id", field_type=IntegerType(), required=True))
        spec = UNPARTITIONED_PARTITION_SPEC

        # Create manifest file 1
        with write_manifest(
            format_version=2,
            spec=spec,
            schema=schema,
            output_file=io.new_output(manifest1_path),
            snapshot_id=1,
            avro_compression="zstandard",
        ) as writer:
            data_file1 = DataFile.from_args(
                content=DataFileContent.DATA,
                file_path=f"{tmp_dir}/data1.parquet",
                file_format=FileFormat.PARQUET,
                partition=Record(),
                record_count=100,
                file_size_in_bytes=1000,
            )
            writer.add_entry(
                ManifestEntry.from_args(
                    status=ManifestEntryStatus.ADDED,
                    snapshot_id=1,
                    data_file=data_file1,
                )
            )
        manifest_file1 = writer.to_manifest_file()

        # Create manifest file 2
        with write_manifest(
            format_version=2,
            spec=spec,
            schema=schema,
            output_file=io.new_output(manifest2_path),
            snapshot_id=2,
            avro_compression="zstandard",
        ) as writer:
            data_file2 = DataFile.from_args(
                content=DataFileContent.DATA,
                file_path=f"{tmp_dir}/data2.parquet",
                file_format=FileFormat.PARQUET,
                partition=Record(),
                record_count=200,
                file_size_in_bytes=2000,
            )
            writer.add_entry(
                ManifestEntry.from_args(
                    status=ManifestEntryStatus.ADDED,
                    snapshot_id=2,
                    data_file=data_file2,
                )
            )
        manifest_file2 = writer.to_manifest_file()

        # Create manifest file 3
        with write_manifest(
            format_version=2,
            spec=spec,
            schema=schema,
            output_file=io.new_output(manifest3_path),
            snapshot_id=3,
            avro_compression="zstandard",
        ) as writer:
            data_file3 = DataFile.from_args(
                content=DataFileContent.DATA,
                file_path=f"{tmp_dir}/data3.parquet",
                file_format=FileFormat.PARQUET,
                partition=Record(),
                record_count=300,
                file_size_in_bytes=3000,
            )
            writer.add_entry(
                ManifestEntry.from_args(
                    status=ManifestEntryStatus.ADDED,
                    snapshot_id=3,
                    data_file=data_file3,
                )
            )
        manifest_file3 = writer.to_manifest_file()

        # Create manifest list 1: contains only manifest1
        manifest_list1_path = f"{tmp_dir}/manifest-list1.avro"
        with write_manifest_list(
            format_version=2,
            output_file=io.new_output(manifest_list1_path),
            snapshot_id=1,
            parent_snapshot_id=None,
            sequence_number=1,
            avro_compression="zstandard",
        ) as list_writer:
            list_writer.add_manifests([manifest_file1])

        # Create manifest list 2: contains manifest1 and manifest2 (overlapping manifest1)
        manifest_list2_path = f"{tmp_dir}/manifest-list2.avro"
        with write_manifest_list(
            format_version=2,
            output_file=io.new_output(manifest_list2_path),
            snapshot_id=2,
            parent_snapshot_id=1,
            sequence_number=2,
            avro_compression="zstandard",
        ) as list_writer:
            list_writer.add_manifests([manifest_file1, manifest_file2])

        # Create manifest list 3: contains all three manifests (overlapping manifest1 and manifest2)
        manifest_list3_path = f"{tmp_dir}/manifest-list3.avro"
        with write_manifest_list(
            format_version=2,
            output_file=io.new_output(manifest_list3_path),
            snapshot_id=3,
            parent_snapshot_id=2,
            sequence_number=3,
            avro_compression="zstandard",
        ) as list_writer:
            list_writer.add_manifests([manifest_file1, manifest_file2, manifest_file3])

        # Read all three manifest lists
        manifests1 = _manifests(io, manifest_list1_path)
        manifests2 = _manifests(io, manifest_list2_path)
        manifests3 = _manifests(io, manifest_list3_path)

        # Verify the manifest files have the expected paths
        assert len(manifests1) == 1
        assert len(manifests2) == 2
        assert len(manifests3) == 3

        # Verify that ManifestFile objects with the same manifest_path are the same object (identity)
        # This is the key assertion - if caching works correctly, the same ManifestFile
        # object should be reused instead of creating duplicates

        # manifest_file1 appears in all three lists - should be the same object
        assert manifests1[0] is manifests2[0], "ManifestFile1 should be the same object instance across manifest lists"
        assert manifests2[0] is manifests3[0], "ManifestFile1 should be the same object instance across manifest lists"

        # manifest_file2 appears in lists 2 and 3 - should be the same object
        assert manifests2[1] is manifests3[1], "ManifestFile2 should be the same object instance across manifest lists"

        # Verify cache size - should only have 3 unique ManifestFile objects
        # instead of 1 + 2 + 3 = 6 objects as with the old approach
        cache = manifest_module._manifest_cache
        assert cache is not None, "Manifest cache should be enabled for this test"
        assert len(cache) == 3, f"Cache should contain exactly 3 unique ManifestFile objects, but has {len(cache)}"


def test_manifest_cache_efficiency_with_many_overlapping_lists() -> None:
    """Test that the manifest cache remains efficient with many overlapping manifest lists.

    This simulates the scenario from GitHub issue #2325 where many appends create
    manifest lists that increasingly overlap.
    """
    io = PyArrowFileIO()

    with TemporaryDirectory() as tmp_dir:
        schema = Schema(NestedField(field_id=1, name="id", field_type=IntegerType(), required=True))
        spec = UNPARTITIONED_PARTITION_SPEC

        num_manifests = 10
        manifest_files = []

        # Create N manifest files
        for i in range(num_manifests):
            manifest_path = f"{tmp_dir}/manifest{i}.avro"
            with write_manifest(
                format_version=2,
                spec=spec,
                schema=schema,
                output_file=io.new_output(manifest_path),
                snapshot_id=i + 1,
                avro_compression="zstandard",
            ) as writer:
                data_file = DataFile.from_args(
                    content=DataFileContent.DATA,
                    file_path=f"{tmp_dir}/data{i}.parquet",
                    file_format=FileFormat.PARQUET,
                    partition=Record(),
                    record_count=100 * (i + 1),
                    file_size_in_bytes=1000 * (i + 1),
                )
                writer.add_entry(
                    ManifestEntry.from_args(
                        status=ManifestEntryStatus.ADDED,
                        snapshot_id=i + 1,
                        data_file=data_file,
                    )
                )
            manifest_files.append(writer.to_manifest_file())

        # Create N manifest lists, each containing an increasing number of manifests
        # list[i] contains manifests[0:i+1]
        manifest_list_paths = []
        for i in range(num_manifests):
            list_path = f"{tmp_dir}/manifest-list{i}.avro"
            with write_manifest_list(
                format_version=2,
                output_file=io.new_output(list_path),
                snapshot_id=i + 1,
                parent_snapshot_id=i if i > 0 else None,
                sequence_number=i + 1,
                avro_compression="zstandard",
            ) as list_writer:
                list_writer.add_manifests(manifest_files[: i + 1])
            manifest_list_paths.append(list_path)

        # Read all manifest lists
        all_results = []
        for path in manifest_list_paths:
            result = _manifests(io, path)
            all_results.append(result)

        # With the old cache approach, we would have:
        # 1 + 2 + 3 + ... + N = N*(N+1)/2 ManifestFile objects in memory
        # With the new approach, we should have exactly N objects

        # Verify cache has exactly N unique entries
        cache = manifest_module._manifest_cache
        assert cache is not None, "Manifest cache should be enabled for this test"
        assert len(cache) == num_manifests, (
            f"Cache should contain exactly {num_manifests} ManifestFile objects, "
            f"but has {len(cache)}. "
            f"Old approach would have {num_manifests * (num_manifests + 1) // 2} objects."
        )

        # Verify object identity - all references to the same manifest should be the same object
        for i in range(num_manifests):
            # Find all references to this manifest across all results
            references = []
            for j, result in enumerate(all_results):
                if j >= i:  # This manifest should be in lists from i onwards
                    references.append(result[i])

            # All references should be the same object
            if len(references) > 1:
                for ref in references[1:]:
                    assert ref is references[0], f"All references to manifest {i} should be the same object instance"


@pytest.mark.parametrize("format_version", [1, 2])
def test_manifest_writer_tell(format_version: TableVersion) -> None:
    io = load_file_io()
    test_schema = Schema(NestedField(1, "foo", IntegerType(), False))

    with TemporaryDirectory() as tmpdir:
        output_file = io.new_output(f"{tmpdir}/test-manifest.avro")
        with write_manifest(
            format_version=format_version,
            spec=UNPARTITIONED_PARTITION_SPEC,
            schema=test_schema,
            output_file=output_file,
            snapshot_id=1,
            avro_compression="null",
        ) as writer:
            initial_bytes = writer.tell()
            data_file = DataFile.from_args(
                content=DataFileContent.DATA,
                file_path=f"{tmpdir}/data.parquet",
                file_format=FileFormat.PARQUET,
                partition=Record(),
                record_count=100,
                file_size_in_bytes=1000,
            )
            entry = ManifestEntry.from_args(
                status=ManifestEntryStatus.ADDED,
                snapshot_id=1,
                data_file=data_file,
            )
            writer.add_entry(entry)
            after_entry_bytes = writer.tell()

            assert after_entry_bytes > initial_bytes, "Bytes should increase after adding entry"


def test_clear_manifest_cache() -> None:
    """Test that clear_manifest_cache() clears cache entries while keeping cache enabled."""
    io = PyArrowFileIO()

    with TemporaryDirectory() as tmp_dir:
        schema = Schema(NestedField(field_id=1, name="id", field_type=IntegerType(), required=True))
        spec = UNPARTITIONED_PARTITION_SPEC

        # Create a manifest file
        manifest_path = f"{tmp_dir}/manifest.avro"
        with write_manifest(
            format_version=2,
            spec=spec,
            schema=schema,
            output_file=io.new_output(manifest_path),
            snapshot_id=1,
            avro_compression="zstandard",
        ) as writer:
            data_file = DataFile.from_args(
                content=DataFileContent.DATA,
                file_path=f"{tmp_dir}/data.parquet",
                file_format=FileFormat.PARQUET,
                partition=Record(),
                record_count=100,
                file_size_in_bytes=1000,
            )
            writer.add_entry(
                ManifestEntry.from_args(
                    status=ManifestEntryStatus.ADDED,
                    snapshot_id=1,
                    data_file=data_file,
                )
            )
        manifest_file = writer.to_manifest_file()

        # Create a manifest list
        list_path = f"{tmp_dir}/manifest-list.avro"
        with write_manifest_list(
            format_version=2,
            output_file=io.new_output(list_path),
            snapshot_id=1,
            parent_snapshot_id=None,
            sequence_number=1,
            avro_compression="zstandard",
        ) as list_writer:
            list_writer.add_manifests([manifest_file])

        # Populate the cache
        _manifests(io, list_path)

        # Verify cache has entries
        cache = manifest_module._manifest_cache
        assert cache is not None, "Cache should be enabled"
        assert len(cache) > 0, "Cache should have entries after reading manifests"

        # Clear the cache
        clear_manifest_cache()

        # Verify cache is empty but still enabled
        cache_after = manifest_module._manifest_cache
        assert cache_after is not None, "Cache should still be enabled after clear"
        assert len(cache_after) == 0, "Cache should be empty after clear"