# 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
"""This contains global pytest configurations.

Fixtures contained in this file will be automatically used if provided as an argument
to any pytest function.

In the case where the fixture must be used in a pytest.mark.parametrize decorator, the string representation can be used
and the built-in pytest fixture request should be used as an additional argument in the function. The fixture can then be
retrieved using `request.getfixturevalue(fixture_name)`.
"""

import os
import re
import socket
import string
import time
import uuid
from collections.abc import Generator
from datetime import date, datetime, timezone
from pathlib import Path
from random import choice, randint
from tempfile import TemporaryDirectory
from typing import (
    TYPE_CHECKING,
    Any,
)
from unittest import mock

import boto3
import pytest
from moto import mock_aws
from pydantic_core import to_json
from pytest_lazy_fixtures import lf

from pyiceberg.catalog import Catalog, load_catalog
from pyiceberg.expressions import BoundReference
from pyiceberg.io import (
    ADLS_ACCOUNT_KEY,
    ADLS_ACCOUNT_NAME,
    ADLS_BLOB_STORAGE_AUTHORITY,
    ADLS_BLOB_STORAGE_SCHEME,
    ADLS_DFS_STORAGE_AUTHORITY,
    ADLS_DFS_STORAGE_SCHEME,
    GCS_PROJECT_ID,
    GCS_SERVICE_HOST,
    GCS_TOKEN,
    GCS_TOKEN_EXPIRES_AT_MS,
    fsspec,
    load_file_io,
)
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, TableMetadataV3
from pyiceberg.table.sorting import NullOrder, SortField, SortOrder
from pyiceberg.transforms import DayTransform, IdentityTransform
from pyiceberg.typedef import Identifier
from pyiceberg.types import (
    BinaryType,
    BooleanType,
    DateType,
    DecimalType,
    DoubleType,
    FixedType,
    FloatType,
    IntegerType,
    ListType,
    LongType,
    MapType,
    NestedField,
    StringType,
    StructType,
    TimestampType,
    TimestamptzType,
    TimeType,
    UUIDType,
)
from pyiceberg.utils.datetime import datetime_to_millis
from pyiceberg.utils.properties import property_as_bool

if TYPE_CHECKING:
    import pyarrow as pa
    from moto.server import ThreadedMotoServer
    from pyspark.sql import SparkSession

    from pyiceberg.io.pyarrow import PyArrowFileIO


def pytest_collection_modifyitems(items: list[pytest.Item]) -> None:
    for item in items:
        if not any(item.iter_markers()):
            item.add_marker("unmarked")


@pytest.fixture(autouse=True, scope="session")
def _isolate_pyiceberg_config() -> None:
    """Make test runs ignore your local PyIceberg config.

    Without this, tests will attempt to resolve a local ~/.pyiceberg.yaml while running pytest.
    This replaces the global catalog config once at session start with an env-only config.
    """
    import pyiceberg.catalog as _catalog_module
    from pyiceberg.utils.config import Config

    with mock.patch.object(Config, "_from_configuration_files", return_value=None):
        _catalog_module._ENV_CONFIG = Config()


def pytest_addoption(parser: pytest.Parser) -> None:
    # S3 options
    parser.addoption(
        "--s3.endpoint", action="store", default="http://localhost:9000", help="The S3 endpoint URL for tests marked as s3"
    )
    parser.addoption("--s3.access-key-id", action="store", default="admin", help="The AWS access key ID for tests marked as s3")
    parser.addoption(
        "--s3.secret-access-key", action="store", default="password", help="The AWS secret access key ID for tests marked as s3"
    )
    # ADLS options
    # Azurite provides default account name and key.  Those can be customized using env variables.
    # For more information, see README file at https://github.com/azure/azurite#default-storage-account
    parser.addoption(
        "--adls.endpoint",
        action="store",
        default="http://127.0.0.1:10000",
        help="The ADLS endpoint URL for tests marked as adls",
    )
    parser.addoption(
        "--adls.account-name", action="store", default="devstoreaccount1", help="The ADLS account key for tests marked as adls"
    )
    parser.addoption(
        "--adls.account-key",
        action="store",
        default="Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==",
        help="The ADLS secret account key for tests marked as adls",
    )
    parser.addoption(
        "--gcs.endpoint", action="store", default="http://0.0.0.0:4443", help="The GCS endpoint URL for tests marked gcs"
    )
    parser.addoption(
        "--gcs.oauth2.token", action="store", default="anon", help="The GCS authentication method for tests marked gcs"
    )
    parser.addoption("--gcs.project-id", action="store", default="test", help="The GCP project for tests marked gcs")


@pytest.fixture(scope="session")
def table_schema_simple() -> Schema:
    return Schema(
        NestedField(field_id=1, name="foo", field_type=StringType(), required=False),
        NestedField(field_id=2, name="bar", field_type=IntegerType(), required=True),
        NestedField(field_id=3, name="baz", field_type=BooleanType(), required=False),
        schema_id=1,
        identifier_field_ids=[2],
    )


@pytest.fixture(scope="session")
def table_schema_with_full_nested_fields() -> Schema:
    return Schema(
        NestedField(
            field_id=1,
            name="foo",
            field_type=StringType(),
            required=False,
            doc="foo doc",
            initial_default="foo initial",
            write_default="foo write",
        ),
        NestedField(
            field_id=2, name="bar", field_type=IntegerType(), required=True, doc="bar doc", initial_default=42, write_default=43
        ),
        NestedField(
            field_id=3,
            name="baz",
            field_type=BooleanType(),
            required=False,
            doc="baz doc",
            initial_default=True,
            write_default=False,
        ),
        schema_id=1,
        identifier_field_ids=[2],
    )


@pytest.fixture(scope="session")
def table_schema_nested() -> Schema:
    return Schema(
        NestedField(field_id=1, name="foo", field_type=StringType(), required=False),
        NestedField(field_id=2, name="bar", field_type=IntegerType(), required=True),
        NestedField(field_id=3, name="baz", field_type=BooleanType(), required=False),
        NestedField(
            field_id=4,
            name="qux",
            field_type=ListType(element_id=5, element_type=StringType(), element_required=True),
            required=True,
        ),
        NestedField(
            field_id=6,
            name="quux",
            field_type=MapType(
                key_id=7,
                key_type=StringType(),
                value_id=8,
                value_type=MapType(key_id=9, key_type=StringType(), value_id=10, value_type=IntegerType(), value_required=True),
                value_required=True,
            ),
            required=True,
        ),
        NestedField(
            field_id=11,
            name="location",
            field_type=ListType(
                element_id=12,
                element_type=StructType(
                    NestedField(field_id=13, name="latitude", field_type=FloatType(), required=False),
                    NestedField(field_id=14, name="longitude", field_type=FloatType(), required=False),
                ),
                element_required=True,
            ),
            required=True,
        ),
        NestedField(
            field_id=15,
            name="person",
            field_type=StructType(
                NestedField(field_id=16, name="name", field_type=StringType(), required=False),
                NestedField(field_id=17, name="age", field_type=IntegerType(), required=True),
            ),
            required=False,
        ),
        schema_id=1,
        identifier_field_ids=[2],
    )


@pytest.fixture(scope="session")
def table_schema_nested_with_struct_key_map() -> Schema:
    return Schema(
        NestedField(field_id=1, name="foo", field_type=StringType(), required=True),
        NestedField(field_id=2, name="bar", field_type=IntegerType(), required=True),
        NestedField(field_id=3, name="baz", field_type=BooleanType(), required=False),
        NestedField(
            field_id=4,
            name="qux",
            field_type=ListType(element_id=5, element_type=StringType(), element_required=True),
            required=True,
        ),
        NestedField(
            field_id=6,
            name="quux",
            field_type=MapType(
                key_id=7,
                key_type=StringType(),
                value_id=8,
                value_type=MapType(key_id=9, key_type=StringType(), value_id=10, value_type=IntegerType(), value_required=True),
                value_required=True,
            ),
            required=True,
        ),
        NestedField(
            field_id=11,
            name="location",
            field_type=MapType(
                key_id=18,
                value_id=19,
                key_type=StructType(
                    NestedField(field_id=21, name="address", field_type=StringType(), required=True),
                    NestedField(field_id=22, name="city", field_type=StringType(), required=True),
                    NestedField(field_id=23, name="zip", field_type=IntegerType(), required=True),
                ),
                value_type=StructType(
                    NestedField(field_id=13, name="latitude", field_type=FloatType(), required=True),
                    NestedField(field_id=14, name="longitude", field_type=FloatType(), required=True),
                ),
                value_required=True,
            ),
            required=True,
        ),
        NestedField(
            field_id=15,
            name="person",
            field_type=StructType(
                NestedField(field_id=16, name="name", field_type=StringType(), required=False),
                NestedField(field_id=17, name="age", field_type=IntegerType(), required=True),
            ),
            required=False,
        ),
        NestedField(
            field_id=24,
            name="points",
            field_type=ListType(
                element_id=25,
                element_type=StructType(
                    NestedField(field_id=26, name="x", field_type=LongType(), required=True),
                    NestedField(field_id=27, name="y", field_type=LongType(), required=True),
                ),
                element_required=False,
            ),
            required=False,
        ),
        NestedField(field_id=28, name="float", field_type=FloatType(), required=True),
        NestedField(field_id=29, name="double", field_type=DoubleType(), required=True),
        schema_id=1,
        identifier_field_ids=[1],
    )


@pytest.fixture(scope="session")
def table_schema_with_all_types() -> Schema:
    return Schema(
        NestedField(field_id=1, name="boolean", field_type=BooleanType(), required=True),
        NestedField(field_id=2, name="integer", field_type=IntegerType(), required=True),
        NestedField(field_id=3, name="long", field_type=LongType(), required=True),
        NestedField(field_id=4, name="float", field_type=FloatType(), required=True),
        NestedField(field_id=5, name="double", field_type=DoubleType(), required=True),
        NestedField(field_id=6, name="decimal", field_type=DecimalType(32, 3), required=True),
        NestedField(field_id=7, name="date", field_type=DateType(), required=True),
        NestedField(field_id=8, name="time", field_type=TimeType(), required=True),
        NestedField(field_id=9, name="timestamp", field_type=TimestampType(), required=True),
        NestedField(field_id=10, name="timestamptz", field_type=TimestamptzType(), required=True),
        NestedField(field_id=11, name="string", field_type=StringType(), required=True),
        NestedField(field_id=12, name="uuid", field_type=UUIDType(), required=True),
        NestedField(field_id=14, name="fixed", field_type=FixedType(12), required=True),
        NestedField(field_id=13, name="binary", field_type=BinaryType(), required=True),
        NestedField(
            field_id=15,
            name="list",
            field_type=ListType(element_id=16, element_type=StringType(), element_required=True),
            required=True,
        ),
        NestedField(
            field_id=17,
            name="map",
            field_type=MapType(
                key_id=18,
                key_type=StringType(),
                value_id=19,
                value_type=IntegerType(),
                value_required=True,
            ),
            required=True,
        ),
        NestedField(
            field_id=20,
            name="struct",
            field_type=StructType(
                NestedField(field_id=21, name="inner_string", field_type=StringType(), required=False),
                NestedField(field_id=22, name="inner_int", field_type=IntegerType(), required=True),
            ),
        ),
        schema_id=1,
        identifier_field_ids=[2],
    )


@pytest.fixture(params=["abfs", "abfss", "wasb", "wasbs"])
def adls_scheme(request: pytest.FixtureRequest) -> str:
    return request.param


@pytest.fixture(scope="session")
def pyarrow_schema_simple_without_ids() -> "pa.Schema":
    import pyarrow as pa

    return pa.schema(
        [
            pa.field("foo", pa.string(), nullable=True),
            pa.field("bar", pa.int32(), nullable=False),
            pa.field("baz", pa.bool_(), nullable=True),
        ]
    )


@pytest.fixture(scope="session")
def pyarrow_schema_nested_without_ids() -> "pa.Schema":
    import pyarrow as pa

    return pa.schema(
        [
            pa.field("foo", pa.string(), nullable=False),
            pa.field("bar", pa.int32(), nullable=False),
            pa.field("baz", pa.bool_(), nullable=True),
            pa.field("qux", pa.list_(pa.string()), nullable=False),
            pa.field(
                "quux",
                pa.map_(
                    pa.string(),
                    pa.map_(pa.string(), pa.int32()),
                ),
                nullable=False,
            ),
            pa.field(
                "location",
                pa.list_(
                    pa.struct(
                        [
                            pa.field("latitude", pa.float32(), nullable=False),
                            pa.field("longitude", pa.float32(), nullable=False),
                        ]
                    ),
                ),
                nullable=False,
            ),
            pa.field(
                "person",
                pa.struct(
                    [
                        pa.field("name", pa.string(), nullable=True),
                        pa.field("age", pa.int32(), nullable=False),
                    ]
                ),
                nullable=True,
            ),
        ]
    )


@pytest.fixture(scope="session")
def iceberg_schema_simple() -> Schema:
    return Schema(
        NestedField(field_id=1, name="foo", field_type=StringType(), required=False),
        NestedField(field_id=2, name="bar", field_type=IntegerType(), required=True),
        NestedField(field_id=3, name="baz", field_type=BooleanType(), required=False),
    )


@pytest.fixture(scope="session")
def iceberg_schema_simple_no_ids() -> Schema:
    return Schema(
        NestedField(field_id=-1, name="foo", field_type=StringType(), required=False),
        NestedField(field_id=-1, name="bar", field_type=IntegerType(), required=True),
        NestedField(field_id=-1, name="baz", field_type=BooleanType(), required=False),
    )


@pytest.fixture(scope="session")
def iceberg_schema_nested() -> Schema:
    return Schema(
        NestedField(field_id=1, name="foo", field_type=StringType(), required=True),
        NestedField(field_id=2, name="bar", field_type=IntegerType(), required=True),
        NestedField(field_id=3, name="baz", field_type=BooleanType(), required=False),
        NestedField(
            field_id=4,
            name="qux",
            field_type=ListType(element_id=5, element_type=StringType(), element_required=False),
            required=True,
        ),
        NestedField(
            field_id=6,
            name="quux",
            field_type=MapType(
                key_id=7,
                key_type=StringType(),
                value_id=8,
                value_type=MapType(key_id=9, key_type=StringType(), value_id=10, value_type=IntegerType(), value_required=False),
                value_required=False,
            ),
            required=True,
        ),
        NestedField(
            field_id=11,
            name="location",
            field_type=ListType(
                element_id=12,
                element_type=StructType(
                    NestedField(field_id=13, name="latitude", field_type=FloatType(), required=True),
                    NestedField(field_id=14, name="longitude", field_type=FloatType(), required=True),
                ),
                element_required=False,
            ),
            required=True,
        ),
        NestedField(
            field_id=15,
            name="person",
            field_type=StructType(
                NestedField(field_id=16, name="name", field_type=StringType(), required=False),
                NestedField(field_id=17, name="age", field_type=IntegerType(), required=True),
            ),
            required=False,
        ),
    )


@pytest.fixture(scope="session")
def iceberg_schema_nested_no_ids() -> Schema:
    return Schema(
        NestedField(field_id=-1, name="foo", field_type=StringType(), required=True),
        NestedField(field_id=-1, name="bar", field_type=IntegerType(), required=True),
        NestedField(field_id=-1, name="baz", field_type=BooleanType(), required=False),
        NestedField(
            field_id=-1,
            name="qux",
            field_type=ListType(element_id=-1, element_type=StringType(), element_required=False),
            required=True,
        ),
        NestedField(
            field_id=-1,
            name="quux",
            field_type=MapType(
                key_id=-1,
                key_type=StringType(),
                value_id=-1,
                value_type=MapType(key_id=-1, key_type=StringType(), value_id=-1, value_type=IntegerType(), value_required=False),
                value_required=False,
            ),
            required=True,
        ),
        NestedField(
            field_id=-1,
            name="location",
            field_type=ListType(
                element_id=-1,
                element_type=StructType(
                    NestedField(field_id=-1, name="latitude", field_type=FloatType(), required=True),
                    NestedField(field_id=-1, name="longitude", field_type=FloatType(), required=True),
                ),
                element_required=False,
            ),
            required=True,
        ),
        NestedField(
            field_id=-1,
            name="person",
            field_type=StructType(
                NestedField(field_id=-1, name="name", field_type=StringType(), required=False),
                NestedField(field_id=-1, name="age", field_type=IntegerType(), required=True),
            ),
            required=False,
        ),
    )


@pytest.fixture(scope="session")
def all_avro_types() -> dict[str, Any]:
    return {
        "type": "record",
        "name": "all_avro_types",
        "fields": [
            {"name": "primitive_string", "type": "string", "field-id": 100},
            {"name": "primitive_int", "type": "int", "field-id": 200},
            {"name": "primitive_long", "type": "long", "field-id": 300},
            {"name": "primitive_float", "type": "float", "field-id": 400},
            {"name": "primitive_double", "type": "double", "field-id": 500},
            {"name": "primitive_bytes", "type": "bytes", "field-id": 600},
            {
                "type": "record",
                "name": "Person",
                "fields": [
                    {"name": "name", "type": "string", "field-id": 701},
                    {"name": "age", "type": "long", "field-id": 702},
                    {"name": "gender", "type": ["string", "null"], "field-id": 703},
                ],
                "field-id": 700,
            },
            {
                "name": "array_with_string",
                "type": {
                    "type": "array",
                    "items": "string",
                    "default": [],
                    "element-id": 801,
                },
                "field-id": 800,
            },
            {
                "name": "array_with_optional_string",
                "type": [
                    "null",
                    {
                        "type": "array",
                        "items": ["string", "null"],
                        "default": [],
                        "element-id": 901,
                    },
                ],
                "field-id": 900,
            },
            {
                "name": "array_with_optional_record",
                "type": [
                    "null",
                    {
                        "type": "array",
                        "items": [
                            "null",
                            {
                                "type": "record",
                                "name": "person",
                                "fields": [
                                    {"name": "name", "type": "string", "field-id": 1002},
                                    {"name": "age", "type": "long", "field-id": 1003},
                                    {"name": "gender", "type": ["string", "null"], "field-id": 1004},
                                ],
                            },
                        ],
                        "element-id": 1001,
                    },
                ],
                "field-id": 1000,
            },
            {
                "name": "map_with_longs",
                "type": {
                    "type": "map",
                    "values": "long",
                    "default": {},
                    "key-id": 1101,
                    "value-id": 1102,
                },
                "field-id": 1000,
            },
        ],
    }


EXAMPLE_TABLE_METADATA_V1 = {
    "format-version": 1,
    "table-uuid": "d20125c8-7284-442c-9aea-15fee620737c",
    "location": "s3://bucket/test/location",
    "last-updated-ms": 1602638573874,
    "last-column-id": 3,
    "schema": {
        "type": "struct",
        "fields": [
            {"id": 1, "name": "x", "required": True, "type": "long"},
            {"id": 2, "name": "y", "required": True, "type": "long", "doc": "comment"},
            {"id": 3, "name": "z", "required": True, "type": "long"},
        ],
    },
    "partition-spec": [{"name": "x", "transform": "identity", "source-id": 1, "field-id": 1000}],
    "properties": {},
    "current-snapshot-id": -1,
    "snapshots": [{"snapshot-id": 1925, "timestamp-ms": 1602638573822, "manifest-list": "s3://bucket/test/manifest-list"}],
}


@pytest.fixture(scope="session")
def example_table_metadata_v1() -> dict[str, Any]:
    return EXAMPLE_TABLE_METADATA_V1


EXAMPLE_TABLE_METADATA_WITH_SNAPSHOT_V1 = {
    "format-version": 1,
    "table-uuid": "b55d9dda-6561-423a-8bfc-787980ce421f",
    "location": "s3://warehouse/database/table",
    "last-updated-ms": 1646787054459,
    "last-column-id": 2,
    "schema": {
        "type": "struct",
        "schema-id": 0,
        "fields": [
            {"id": 1, "name": "id", "required": False, "type": "int"},
            {"id": 2, "name": "data", "required": False, "type": "string"},
        ],
    },
    "current-schema-id": 0,
    "schemas": [
        {
            "type": "struct",
            "schema-id": 0,
            "fields": [
                {"id": 1, "name": "id", "required": False, "type": "int"},
                {"id": 2, "name": "data", "required": False, "type": "string"},
            ],
        }
    ],
    "partition-spec": [],
    "default-spec-id": 0,
    "partition-specs": [{"spec-id": 0, "fields": []}],
    "last-partition-id": 999,
    "default-sort-order-id": 0,
    "sort-orders": [{"order-id": 0, "fields": []}],
    "properties": {
        "owner": "bryan",
        "write.metadata.compression-codec": "gzip",
    },
    "current-snapshot-id": 3497810964824022504,
    "refs": {"main": {"snapshot-id": 3497810964824022504, "type": "branch"}},
    "snapshots": [
        {
            "snapshot-id": 3497810964824022504,
            "timestamp-ms": 1646787054459,
            "summary": {
                "operation": "append",
                "spark.app.id": "local-1646787004168",
                "added-data-files": "1",
                "added-records": "1",
                "added-files-size": "697",
                "changed-partition-count": "1",
                "total-records": "1",
                "total-files-size": "697",
                "total-data-files": "1",
                "total-delete-files": "0",
                "total-position-deletes": "0",
                "total-equality-deletes": "0",
            },
            "manifest-list": "s3://warehouse/database/table/metadata/snap-3497810964824022504-1-c4f68204-666b-4e50-a9df-b10c34bf6b82.avro",
            "schema-id": 0,
        }
    ],
    "snapshot-log": [{"timestamp-ms": 1646787054459, "snapshot-id": 3497810964824022504}],
    "metadata-log": [
        {
            "timestamp-ms": 1646787031514,
            "metadata-file": "s3://warehouse/database/table/metadata/00000-88484a1c-00e5-4a07-a787-c0e7aeffa805.gz.metadata.json",
        }
    ],
}


@pytest.fixture
def example_table_metadata_with_snapshot_v1() -> dict[str, Any]:
    return EXAMPLE_TABLE_METADATA_WITH_SNAPSHOT_V1


EXAMPLE_TABLE_METADATA_NO_SNAPSHOT_V1 = {
    "format-version": 1,
    "table-uuid": "bf289591-dcc0-4234-ad4f-5c3eed811a29",
    "location": "s3://warehouse/database/table",
    "last-updated-ms": 1657810967051,
    "last-column-id": 3,
    "schema": {
        "type": "struct",
        "schema-id": 0,
        "identifier-field-ids": [2],
        "fields": [
            {"id": 1, "name": "foo", "required": False, "type": "string"},
            {"id": 2, "name": "bar", "required": True, "type": "int"},
            {"id": 3, "name": "baz", "required": False, "type": "boolean"},
        ],
    },
    "current-schema-id": 0,
    "schemas": [
        {
            "type": "struct",
            "schema-id": 0,
            "identifier-field-ids": [2],
            "fields": [
                {"id": 1, "name": "foo", "required": False, "type": "string"},
                {"id": 2, "name": "bar", "required": True, "type": "int"},
                {"id": 3, "name": "baz", "required": False, "type": "boolean"},
            ],
        }
    ],
    "partition-spec": [],
    "default-spec-id": 0,
    "last-partition-id": 999,
    "default-sort-order-id": 0,
    "sort-orders": [{"order-id": 0, "fields": []}],
    "properties": {
        "write.delete.parquet.compression-codec": "zstd",
        "write.metadata.compression-codec": "gzip",
        "write.summary.partition-limit": "100",
        "write.parquet.compression-codec": "zstd",
    },
    "current-snapshot-id": -1,
    "refs": {},
    "snapshots": [],
    "snapshot-log": [],
    "metadata-log": [],
}


@pytest.fixture
def example_table_metadata_no_snapshot_v1() -> dict[str, Any]:
    return EXAMPLE_TABLE_METADATA_NO_SNAPSHOT_V1


@pytest.fixture
def example_table_metadata_v2_with_extensive_snapshots() -> dict[str, Any]:
    def generate_snapshot(
        snapshot_id: int,
        parent_snapshot_id: int | None = None,
        timestamp_ms: int | None = None,
        sequence_number: int = 0,
    ) -> dict[str, Any]:
        return {
            "snapshot-id": snapshot_id,
            "parent-snapshot-id": parent_snapshot_id,
            "timestamp-ms": timestamp_ms or int(time.time() * 1000),
            "sequence-number": sequence_number,
            "summary": {"operation": "append"},
            "manifest-list": f"s3://a/b/{snapshot_id}.avro",
        }

    snapshots = []
    snapshot_log = []
    initial_snapshot_id = 3051729675574597004

    for i in range(2000):
        snapshot_id = initial_snapshot_id + i
        parent_snapshot_id = snapshot_id - 1 if i > 0 else None
        timestamp_ms = int(time.time() * 1000) - randint(0, 1000000)
        snapshots.append(generate_snapshot(snapshot_id, parent_snapshot_id, timestamp_ms, i))
        snapshot_log.append({"snapshot-id": snapshot_id, "timestamp-ms": timestamp_ms})

    return {
        "format-version": 2,
        "table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c1",
        "location": "s3://bucket/test/location",
        "last-sequence-number": 34,
        "last-updated-ms": 1602638573590,
        "last-column-id": 3,
        "current-schema-id": 1,
        "schemas": [
            {"type": "struct", "schema-id": 0, "fields": [{"id": 1, "name": "x", "required": True, "type": "long"}]},
            {
                "type": "struct",
                "schema-id": 1,
                "identifier-field-ids": [1, 2],
                "fields": [
                    {"id": 1, "name": "x", "required": True, "type": "long"},
                    {"id": 2, "name": "y", "required": True, "type": "long", "doc": "comment"},
                    {"id": 3, "name": "z", "required": True, "type": "long"},
                ],
            },
        ],
        "default-spec-id": 0,
        "partition-specs": [{"spec-id": 0, "fields": [{"name": "x", "transform": "identity", "source-id": 1, "field-id": 1000}]}],
        "last-partition-id": 1000,
        "default-sort-order-id": 3,
        "sort-orders": [
            {
                "order-id": 3,
                "fields": [
                    {"transform": "identity", "source-id": 2, "direction": "asc", "null-order": "nulls-first"},
                    {"transform": "bucket[4]", "source-id": 3, "direction": "desc", "null-order": "nulls-last"},
                ],
            }
        ],
        "properties": {"read.split.target.size": "134217728"},
        "current-snapshot-id": initial_snapshot_id + 1999,
        "snapshots": snapshots,
        "snapshot-log": snapshot_log,
        "metadata-log": [{"metadata-file": "s3://bucket/.../v1.json", "timestamp-ms": 1515100}],
        "refs": {"test": {"snapshot-id": initial_snapshot_id, "type": "tag", "max-ref-age-ms": 10000000}},
    }


EXAMPLE_TABLE_METADATA_V2 = {
    "format-version": 2,
    "table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c1",
    "location": "s3://bucket/test/location",
    "last-sequence-number": 34,
    "last-updated-ms": 1602638573590,
    "last-column-id": 3,
    "current-schema-id": 1,
    "schemas": [
        {"type": "struct", "schema-id": 0, "fields": [{"id": 1, "name": "x", "required": True, "type": "long"}]},
        {
            "type": "struct",
            "schema-id": 1,
            "identifier-field-ids": [1, 2],
            "fields": [
                {"id": 1, "name": "x", "required": True, "type": "long"},
                {"id": 2, "name": "y", "required": True, "type": "long", "doc": "comment"},
                {"id": 3, "name": "z", "required": True, "type": "long"},
            ],
        },
    ],
    "default-spec-id": 0,
    "partition-specs": [{"spec-id": 0, "fields": [{"name": "x", "transform": "identity", "source-id": 1, "field-id": 1000}]}],
    "last-partition-id": 1000,
    "default-sort-order-id": 3,
    "sort-orders": [
        {
            "order-id": 3,
            "fields": [
                {"transform": "identity", "source-id": 2, "direction": "asc", "null-order": "nulls-first"},
                {"transform": "bucket[4]", "source-id": 3, "direction": "desc", "null-order": "nulls-last"},
            ],
        }
    ],
    "properties": {"read.split.target.size": "134217728"},
    "current-snapshot-id": 3055729675574597004,
    "snapshots": [
        {
            "snapshot-id": 3051729675574597004,
            "timestamp-ms": 1515100955770,
            "sequence-number": 0,
            "summary": {"operation": "append"},
            "manifest-list": "s3://a/b/1.avro",
        },
        {
            "snapshot-id": 3055729675574597004,
            "parent-snapshot-id": 3051729675574597004,
            "timestamp-ms": 1555100955770,
            "sequence-number": 1,
            "summary": {"operation": "append"},
            "manifest-list": "s3://a/b/2.avro",
            "schema-id": 1,
        },
    ],
    "snapshot-log": [
        {"snapshot-id": 3051729675574597004, "timestamp-ms": 1515100955770},
        {"snapshot-id": 3055729675574597004, "timestamp-ms": 1555100955770},
    ],
    "metadata-log": [{"metadata-file": "s3://bucket/.../v1.json", "timestamp-ms": 1515100}],
    "refs": {"test": {"snapshot-id": 3051729675574597004, "type": "tag", "max-ref-age-ms": 10000000}},
}

EXAMPLE_TABLE_METADATA_V3 = {
    "format-version": 3,
    "table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c1",
    "location": "s3://bucket/test/location",
    "last-sequence-number": 34,
    "next-row-id": 1,
    "last-updated-ms": 1602638573590,
    "last-column-id": 3,
    "current-schema-id": 1,
    "schemas": [
        {"type": "struct", "schema-id": 0, "fields": [{"id": 1, "name": "x", "required": True, "type": "long"}]},
        {
            "type": "struct",
            "schema-id": 1,
            "identifier-field-ids": [1, 2],
            "fields": [
                {"id": 1, "name": "x", "required": True, "type": "long"},
                {"id": 2, "name": "y", "required": True, "type": "long", "doc": "comment"},
                {"id": 3, "name": "z", "required": True, "type": "long"},
                {"id": 4, "name": "u", "required": True, "type": "unknown"},
                {"id": 5, "name": "ns", "required": True, "type": "timestamp_ns"},
                {"id": 6, "name": "nstz", "required": True, "type": "timestamptz_ns"},
            ],
        },
    ],
    "default-spec-id": 0,
    "partition-specs": [{"spec-id": 0, "fields": [{"name": "x", "transform": "identity", "source-ids": [1], "field-id": 1000}]}],
    "last-partition-id": 1000,
    "default-sort-order-id": 3,
    "sort-orders": [
        {
            "order-id": 3,
            "fields": [
                {"transform": "identity", "source-ids": [2], "direction": "asc", "null-order": "nulls-first"},
                {"transform": "bucket[4]", "source-ids": [3], "direction": "desc", "null-order": "nulls-last"},
            ],
        }
    ],
    "properties": {"read.split.target.size": "134217728"},
    "current-snapshot-id": 3055729675574597004,
    "snapshots": [
        {
            "snapshot-id": 3051729675574597004,
            "timestamp-ms": 1515100955770,
            "sequence-number": 0,
            "summary": {"operation": "append"},
            "manifest-list": "s3://a/b/1.avro",
        },
        {
            "snapshot-id": 3055729675574597004,
            "parent-snapshot-id": 3051729675574597004,
            "timestamp-ms": 1555100955770,
            "sequence-number": 1,
            "summary": {"operation": "append"},
            "manifest-list": "s3://a/b/2.avro",
            "schema-id": 1,
        },
    ],
    "snapshot-log": [
        {"snapshot-id": 3051729675574597004, "timestamp-ms": 1515100955770},
        {"snapshot-id": 3055729675574597004, "timestamp-ms": 1555100955770},
    ],
    "metadata-log": [{"metadata-file": "s3://bucket/.../v1.json", "timestamp-ms": 1515100}],
    "refs": {"test": {"snapshot-id": 3051729675574597004, "type": "tag", "max-ref-age-ms": 10000000}},
}

TABLE_METADATA_V2_WITH_FIXED_AND_DECIMAL_TYPES = {
    "format-version": 2,
    "table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c1",
    "location": "s3://bucket/test/location",
    "last-sequence-number": 34,
    "last-updated-ms": 1602638573590,
    "last-column-id": 7,
    "current-schema-id": 1,
    "schemas": [
        {
            "type": "struct",
            "schema-id": 1,
            "identifier-field-ids": [1],
            "fields": [
                {"id": 1, "name": "x", "required": True, "type": "long"},
                {"id": 4, "name": "a", "required": True, "type": "decimal(16, 2)"},
                {"id": 5, "name": "b", "required": True, "type": "decimal(16, 8)"},
                {"id": 6, "name": "c", "required": True, "type": "fixed[16]"},
                {"id": 7, "name": "d", "required": True, "type": "fixed[18]"},
            ],
        }
    ],
    "default-spec-id": 0,
    "partition-specs": [{"spec-id": 0, "fields": [{"name": "x", "transform": "identity", "source-id": 1, "field-id": 1000}]}],
    "last-partition-id": 1000,
    "properties": {"read.split.target.size": "134217728"},
    "current-snapshot-id": 3055729675574597004,
    "snapshots": [
        {
            "snapshot-id": 3051729675574597004,
            "timestamp-ms": 1515100955770,
            "sequence-number": 0,
            "summary": {"operation": "append"},
            "manifest-list": "s3://a/b/1.avro",
        },
        {
            "snapshot-id": 3055729675574597004,
            "parent-snapshot-id": 3051729675574597004,
            "timestamp-ms": 1555100955770,
            "sequence-number": 1,
            "summary": {"operation": "append"},
            "manifest-list": "s3://a/b/2.avro",
            "schema-id": 1,
        },
    ],
    "snapshot-log": [
        {"snapshot-id": 3051729675574597004, "timestamp-ms": 1515100955770},
        {"snapshot-id": 3055729675574597004, "timestamp-ms": 1555100955770},
    ],
    "metadata-log": [{"metadata-file": "s3://bucket/.../v1.json", "timestamp-ms": 1515100}],
    "refs": {"test": {"snapshot-id": 3051729675574597004, "type": "tag", "max-ref-age-ms": 10000000}},
}

TABLE_METADATA_V2_WITH_STATISTICS = {
    "format-version": 2,
    "table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c1",
    "location": "s3://bucket/test/location",
    "last-sequence-number": 34,
    "last-updated-ms": 1602638573590,
    "last-column-id": 3,
    "current-schema-id": 0,
    "schemas": [
        {
            "type": "struct",
            "schema-id": 0,
            "fields": [
                {
                    "id": 1,
                    "name": "x",
                    "required": True,
                    "type": "long",
                }
            ],
        }
    ],
    "default-spec-id": 0,
    "partition-specs": [{"spec-id": 0, "fields": []}],
    "last-partition-id": 1000,
    "default-sort-order-id": 0,
    "sort-orders": [{"order-id": 0, "fields": []}],
    "properties": {},
    "current-snapshot-id": 3055729675574597004,
    "snapshots": [
        {
            "snapshot-id": 3051729675574597004,
            "timestamp-ms": 1515100955770,
            "sequence-number": 0,
            "summary": {"operation": "append"},
            "manifest-list": "s3://a/b/1.avro",
        },
        {
            "snapshot-id": 3055729675574597004,
            "parent-snapshot-id": 3051729675574597004,
            "timestamp-ms": 1555100955770,
            "sequence-number": 1,
            "summary": {"operation": "append"},
            "manifest-list": "s3://a/b/2.avro",
            "schema-id": 1,
        },
    ],
    "statistics": [
        {
            "snapshot-id": 3051729675574597004,
            "statistics-path": "s3://a/b/stats.puffin",
            "file-size-in-bytes": 413,
            "file-footer-size-in-bytes": 42,
            "blob-metadata": [
                {
                    "type": "apache-datasketches-theta-v1",
                    "snapshot-id": 3051729675574597004,
                    "sequence-number": 1,
                    "fields": [1],
                }
            ],
        },
        {
            "snapshot-id": 3055729675574597004,
            "statistics-path": "s3://a/b/stats.puffin",
            "file-size-in-bytes": 413,
            "file-footer-size-in-bytes": 42,
            "blob-metadata": [
                {
                    "type": "deletion-vector-v1",
                    "snapshot-id": 3055729675574597004,
                    "sequence-number": 1,
                    "fields": [1],
                }
            ],
        },
    ],
    "snapshot-log": [],
    "metadata-log": [],
}


@pytest.fixture
def example_table_metadata_v2() -> dict[str, Any]:
    return EXAMPLE_TABLE_METADATA_V2


@pytest.fixture
def table_metadata_v2_with_fixed_and_decimal_types() -> dict[str, Any]:
    return TABLE_METADATA_V2_WITH_FIXED_AND_DECIMAL_TYPES


@pytest.fixture
def table_metadata_v2_with_statistics() -> dict[str, Any]:
    return TABLE_METADATA_V2_WITH_STATISTICS


@pytest.fixture
def example_view_metadata_v1() -> dict[str, Any]:
    return {
        "view-uuid": "a20125c8-7284-442c-9aea-15fee620737c",
        "format-version": 1,
        "location": "s3://bucket/test/location/test_view",
        "current-version-id": 1,
        "versions": [
            {
                "version-id": 1,
                "timestamp-ms": 1602638573874,
                "schema-id": 1,
                "summary": {"engine-name": "spark", "engineVersion": "3.3"},
                "representations": [
                    {
                        "type": "sql",
                        "sql": "SELECT * FROM prod.db.table",
                        "dialect": "spark",
                    }
                ],
                "default-namespace": ["default"],
            }
        ],
        "schemas": [
            {
                "type": "struct",
                "schema-id": 1,
                "fields": [
                    {"id": 1, "name": "x", "required": True, "type": "long"},
                    {"id": 2, "name": "y", "required": True, "type": "long", "doc": "comment"},
                    {"id": 3, "name": "z", "required": True, "type": "long"},
                ],
            }
        ],
        "version-log": [{"timestamp-ms": 1602638573874, "version-id": 1}],
        "properties": {"comment": "this is a test view"},
    }


@pytest.fixture
def example_table_metadata_v3() -> dict[str, Any]:
    return EXAMPLE_TABLE_METADATA_V3


def generate_table_location_with_version_hint(
    tmp_path_factory: pytest.TempPathFactory, content_in_version_hint: str, metadata_filename: str
) -> str:
    from pyiceberg.io.pyarrow import PyArrowFileIO

    metadata_location = str(tmp_path_factory.getbasetemp() / "metadata" / metadata_filename)
    version_hint_location = str(tmp_path_factory.getbasetemp() / "metadata" / "version-hint.text")
    metadata = TableMetadataV2(**EXAMPLE_TABLE_METADATA_V2)
    ToOutputFile.table_metadata(metadata, PyArrowFileIO().new_output(location=metadata_location), overwrite=True)

    with PyArrowFileIO().new_output(location=version_hint_location).create(overwrite=True) as s:
        s.write(content_in_version_hint.encode("utf-8"))

    return str(tmp_path_factory.getbasetemp())


@pytest.fixture(scope="session")
def table_location_with_version_hint_full(tmp_path_factory: pytest.TempPathFactory) -> str:
    content_in_version_hint = str(uuid.uuid4())
    metadata_filename = f"{content_in_version_hint}.metadata.json"
    return generate_table_location_with_version_hint(tmp_path_factory, content_in_version_hint, metadata_filename)


@pytest.fixture(scope="session")
def table_location_with_version_hint_numeric(tmp_path_factory: pytest.TempPathFactory) -> str:
    content_in_version_hint = "1234567890"
    metadata_filename = f"v{content_in_version_hint}.metadata.json"
    return generate_table_location_with_version_hint(tmp_path_factory, content_in_version_hint, metadata_filename)


@pytest.fixture(scope="session")
def table_location_with_version_hint_non_numeric(tmp_path_factory: pytest.TempPathFactory) -> str:
    content_in_version_hint = "non_numberic"
    metadata_filename = f"{content_in_version_hint}.metadata.json"
    return generate_table_location_with_version_hint(tmp_path_factory, content_in_version_hint, metadata_filename)


@pytest.fixture(scope="session")
def metadata_location(tmp_path_factory: pytest.TempPathFactory) -> str:
    from pyiceberg.io.pyarrow import PyArrowFileIO

    metadata_location = str(tmp_path_factory.mktemp("metadata") / f"{uuid.uuid4()}.metadata.json")
    metadata = TableMetadataV2(**EXAMPLE_TABLE_METADATA_V2)
    ToOutputFile.table_metadata(metadata, PyArrowFileIO().new_output(location=metadata_location), overwrite=True)
    return metadata_location


@pytest.fixture(scope="session")
def metadata_location_gz(tmp_path_factory: pytest.TempPathFactory) -> str:
    from pyiceberg.io.pyarrow import PyArrowFileIO

    metadata_location = str(tmp_path_factory.mktemp("metadata") / f"{uuid.uuid4()}.gz.metadata.json")
    metadata = TableMetadataV2(**EXAMPLE_TABLE_METADATA_V2)
    ToOutputFile.table_metadata(metadata, PyArrowFileIO().new_output(location=metadata_location), overwrite=True)
    return metadata_location


manifest_entry_records = [
    {
        "status": 1,
        "snapshot_id": 8744736658442914487,
        "data_file": {
            "file_path": (
                "/home/iceberg/warehouse/nyc/taxis_partitioned/data/VendorID=null/"
                "00000-633-d8a4223e-dc97-45a1-86e1-adaba6e8abd7-00001.parquet"
            ),
            "file_format": "PARQUET",
            "partition": {"VendorID": 1, "tpep_pickup_day": 1925},
            "record_count": 19513,
            "file_size_in_bytes": 388872,
            "block_size_in_bytes": 67108864,
            "column_sizes": [
                {"key": 1, "value": 53},
                {"key": 2, "value": 98153},
                {"key": 3, "value": 98693},
                {"key": 4, "value": 53},
                {"key": 5, "value": 53},
                {"key": 6, "value": 53},
                {"key": 7, "value": 17425},
                {"key": 8, "value": 18528},
                {"key": 9, "value": 53},
                {"key": 10, "value": 44788},
                {"key": 11, "value": 35571},
                {"key": 12, "value": 53},
                {"key": 13, "value": 1243},
                {"key": 14, "value": 2355},
                {"key": 15, "value": 12750},
                {"key": 16, "value": 4029},
                {"key": 17, "value": 110},
                {"key": 18, "value": 47194},
                {"key": 19, "value": 2948},
            ],
            "value_counts": [
                {"key": 1, "value": 19513},
                {"key": 2, "value": 19513},
                {"key": 3, "value": 19513},
                {"key": 4, "value": 19513},
                {"key": 5, "value": 19513},
                {"key": 6, "value": 19513},
                {"key": 7, "value": 19513},
                {"key": 8, "value": 19513},
                {"key": 9, "value": 19513},
                {"key": 10, "value": 19513},
                {"key": 11, "value": 19513},
                {"key": 12, "value": 19513},
                {"key": 13, "value": 19513},
                {"key": 14, "value": 19513},
                {"key": 15, "value": 19513},
                {"key": 16, "value": 19513},
                {"key": 17, "value": 19513},
                {"key": 18, "value": 19513},
                {"key": 19, "value": 19513},
            ],
            "null_value_counts": [
                {"key": 1, "value": 19513},
                {"key": 2, "value": 0},
                {"key": 3, "value": 0},
                {"key": 4, "value": 19513},
                {"key": 5, "value": 19513},
                {"key": 6, "value": 19513},
                {"key": 7, "value": 0},
                {"key": 8, "value": 0},
                {"key": 9, "value": 19513},
                {"key": 10, "value": 0},
                {"key": 11, "value": 0},
                {"key": 12, "value": 19513},
                {"key": 13, "value": 0},
                {"key": 14, "value": 0},
                {"key": 15, "value": 0},
                {"key": 16, "value": 0},
                {"key": 17, "value": 0},
                {"key": 18, "value": 0},
                {"key": 19, "value": 0},
            ],
            "nan_value_counts": [
                {"key": 16, "value": 0},
                {"key": 17, "value": 0},
                {"key": 18, "value": 0},
                {"key": 19, "value": 0},
                {"key": 10, "value": 0},
                {"key": 11, "value": 0},
                {"key": 12, "value": 0},
                {"key": 13, "value": 0},
                {"key": 14, "value": 0},
                {"key": 15, "value": 0},
            ],
            "lower_bounds": [
                {"key": 2, "value": b"\x01\x00\x00\x00\x00\x00\x00\x00"},
                {"key": 3, "value": b"\x01\x00\x00\x00\x00\x00\x00\x00"},
                {"key": 7, "value": b"\x03\x00\x00\x00"},
                {"key": 8, "value": b"\x01\x00\x00\x00"},
                {"key": 10, "value": b"\xf6(\\\x8f\xc2\x05S\xc0"},
                {"key": 11, "value": b"\x00\x00\x00\x00\x00\x00\x00\x00"},
                {"key": 13, "value": b"\x00\x00\x00\x00\x00\x00\x00\x00"},
                {"key": 14, "value": b"\x00\x00\x00\x00\x00\x00\xe0\xbf"},
                {"key": 15, "value": b")\\\x8f\xc2\xf5(\x08\xc0"},
                {"key": 16, "value": b"\x00\x00\x00\x00\x00\x00\x00\x00"},
                {"key": 17, "value": b"\x00\x00\x00\x00\x00\x00\x00\x00"},
                {"key": 18, "value": b"\xf6(\\\x8f\xc2\xc5S\xc0"},
                {"key": 19, "value": b"\x00\x00\x00\x00\x00\x00\x04\xc0"},
            ],
            "upper_bounds": [
                {"key": 2, "value": b"\x06\x00\x00\x00\x00\x00\x00\x00"},
                {"key": 3, "value": b"\x06\x00\x00\x00\x00\x00\x00\x00"},
                {"key": 7, "value": b"\t\x01\x00\x00"},
                {"key": 8, "value": b"\t\x01\x00\x00"},
                {"key": 10, "value": b"\xcd\xcc\xcc\xcc\xcc,_@"},
                {"key": 11, "value": b"\x1f\x85\xebQ\\\xe2\xfe@"},
                {"key": 13, "value": b"\x00\x00\x00\x00\x00\x00\x12@"},
                {"key": 14, "value": b"\x00\x00\x00\x00\x00\x00\xe0?"},
                {"key": 15, "value": b"q=\n\xd7\xa3\xf01@"},
                {"key": 16, "value": b"\x00\x00\x00\x00\x00`B@"},
                {"key": 17, "value": b"333333\xd3?"},
                {"key": 18, "value": b"\x00\x00\x00\x00\x00\x18b@"},
                {"key": 19, "value": b"\x00\x00\x00\x00\x00\x00\x04@"},
            ],
            "key_metadata": None,
            "split_offsets": [4],
            "sort_order_id": 0,
        },
    },
    {
        "status": 1,
        "snapshot_id": 8744736658442914487,
        "data_file": {
            "file_path": (
                "/home/iceberg/warehouse/nyc/taxis_partitioned/data/VendorID=1/"
                "00000-633-d8a4223e-dc97-45a1-86e1-adaba6e8abd7-00002.parquet"
            ),
            "file_format": "PARQUET",
            "partition": {"VendorID": 1, "tpep_pickup_datetime": None},
            "record_count": 95050,
            "file_size_in_bytes": 1265950,
            "block_size_in_bytes": 67108864,
            "column_sizes": [
                {"key": 1, "value": 318},
                {"key": 2, "value": 329806},
                {"key": 3, "value": 331632},
                {"key": 4, "value": 15343},
                {"key": 5, "value": 2351},
                {"key": 6, "value": 3389},
                {"key": 7, "value": 71269},
                {"key": 8, "value": 76429},
                {"key": 9, "value": 16383},
                {"key": 10, "value": 86992},
                {"key": 11, "value": 89608},
                {"key": 12, "value": 265},
                {"key": 13, "value": 19377},
                {"key": 14, "value": 1692},
                {"key": 15, "value": 76162},
                {"key": 16, "value": 4354},
                {"key": 17, "value": 759},
                {"key": 18, "value": 120650},
                {"key": 19, "value": 11804},
            ],
            "value_counts": [
                {"key": 1, "value": 95050},
                {"key": 2, "value": 95050},
                {"key": 3, "value": 95050},
                {"key": 4, "value": 95050},
                {"key": 5, "value": 95050},
                {"key": 6, "value": 95050},
                {"key": 7, "value": 95050},
                {"key": 8, "value": 95050},
                {"key": 9, "value": 95050},
                {"key": 10, "value": 95050},
                {"key": 11, "value": 95050},
                {"key": 12, "value": 95050},
                {"key": 13, "value": 95050},
                {"key": 14, "value": 95050},
                {"key": 15, "value": 95050},
                {"key": 16, "value": 95050},
                {"key": 17, "value": 95050},
                {"key": 18, "value": 95050},
                {"key": 19, "value": 95050},
            ],
            "null_value_counts": [
                {"key": 1, "value": 0},
                {"key": 2, "value": 0},
                {"key": 3, "value": 0},
                {"key": 4, "value": 0},
                {"key": 5, "value": 0},
                {"key": 6, "value": 0},
                {"key": 7, "value": 0},
                {"key": 8, "value": 0},
                {"key": 9, "value": 0},
                {"key": 10, "value": 0},
                {"key": 11, "value": 0},
                {"key": 12, "value": 95050},
                {"key": 13, "value": 0},
                {"key": 14, "value": 0},
                {"key": 15, "value": 0},
                {"key": 16, "value": 0},
                {"key": 17, "value": 0},
                {"key": 18, "value": 0},
                {"key": 19, "value": 0},
            ],
            "nan_value_counts": [
                {"key": 16, "value": 0},
                {"key": 17, "value": 0},
                {"key": 18, "value": 0},
                {"key": 19, "value": 0},
                {"key": 10, "value": 0},
                {"key": 11, "value": 0},
                {"key": 12, "value": 0},
                {"key": 13, "value": 0},
                {"key": 14, "value": 0},
                {"key": 15, "value": 0},
            ],
            "lower_bounds": [
                {"key": 1, "value": b"\x01\x00\x00\x00"},
                {"key": 2, "value": b"\x01\x00\x00\x00\x00\x00\x00\x00"},
                {"key": 3, "value": b"\x01\x00\x00\x00\x00\x00\x00\x00"},
                {"key": 4, "value": b"\x00\x00\x00\x00"},
                {"key": 5, "value": b"\x01\x00\x00\x00"},
                {"key": 6, "value": b"N"},
                {"key": 7, "value": b"\x01\x00\x00\x00"},
                {"key": 8, "value": b"\x01\x00\x00\x00"},
                {"key": 9, "value": b"\x01\x00\x00\x00"},
                {"key": 10, "value": b"\x00\x00\x00\x00\x00\x00\x00\x00"},
                {"key": 11, "value": b"\x00\x00\x00\x00\x00\x00\x00\x00"},
                {"key": 13, "value": b"\x00\x00\x00\x00\x00\x00\x00\x00"},
                {"key": 14, "value": b"\x00\x00\x00\x00\x00\x00\x00\x00"},
                {"key": 15, "value": b"\x00\x00\x00\x00\x00\x00\x00\x00"},
                {"key": 16, "value": b"\x00\x00\x00\x00\x00\x00\x00\x00"},
                {"key": 17, "value": b"\x00\x00\x00\x00\x00\x00\x00\x00"},
                {"key": 18, "value": b"\x00\x00\x00\x00\x00\x00\x00\x00"},
                {"key": 19, "value": b"\x00\x00\x00\x00\x00\x00\x00\x00"},
            ],
            "upper_bounds": [
                {"key": 1, "value": b"\x01\x00\x00\x00"},
                {"key": 2, "value": b"\x06\x00\x00\x00\x00\x00\x00\x00"},
                {"key": 3, "value": b"\x06\x00\x00\x00\x00\x00\x00\x00"},
                {"key": 4, "value": b"\x06\x00\x00\x00"},
                {"key": 5, "value": b"c\x00\x00\x00"},
                {"key": 6, "value": b"Y"},
                {"key": 7, "value": b"\t\x01\x00\x00"},
                {"key": 8, "value": b"\t\x01\x00\x00"},
                {"key": 9, "value": b"\x04\x00\x00\x00"},
                {"key": 10, "value": b"\\\x8f\xc2\xf5(8\x8c@"},
                {"key": 11, "value": b"\xcd\xcc\xcc\xcc\xcc,f@"},
                {"key": 13, "value": b"\x00\x00\x00\x00\x00\x00\x1c@"},
                {"key": 14, "value": b"\x9a\x99\x99\x99\x99\x99\xf1?"},
                {"key": 15, "value": b"\x00\x00\x00\x00\x00\x00Y@"},
                {"key": 16, "value": b"\x00\x00\x00\x00\x00\xb0X@"},
                {"key": 17, "value": b"333333\xd3?"},
                {"key": 18, "value": b"\xc3\xf5(\\\x8f:\x8c@"},
                {"key": 19, "value": b"\x00\x00\x00\x00\x00\x00\x04@"},
            ],
            "key_metadata": None,
            "split_offsets": [4],
            "sort_order_id": 0,
        },
    },
]

manifest_file_records_v1 = [
    {
        "manifest_path": "/home/iceberg/warehouse/nyc/taxis_partitioned/metadata/0125c686-8aa6-4502-bdcc-b6d17ca41a3b-m0.avro",
        "manifest_length": 7989,
        "partition_spec_id": 0,
        "added_snapshot_id": 9182715666859759686,
        "added_data_files_count": 3,
        "existing_data_files_count": 0,
        "deleted_data_files_count": 0,
        "partitions": [
            {"contains_null": True, "contains_nan": False, "lower_bound": b"\x01\x00\x00\x00", "upper_bound": b"\x02\x00\x00\x00"}
        ],
        "added_rows_count": 237993,
        "existing_rows_count": 0,
        "deleted_rows_count": 0,
    }
]

manifest_file_records_v2 = [
    {
        "manifest_path": "/home/iceberg/warehouse/nyc/taxis_partitioned/metadata/0125c686-8aa6-4502-bdcc-b6d17ca41a3b-m0.avro",
        "manifest_length": 7989,
        "partition_spec_id": 0,
        "content": 1,
        "sequence_number": 3,
        "min_sequence_number": 3,
        "added_snapshot_id": 9182715666859759686,
        "added_files_count": 3,
        "existing_files_count": 0,
        "deleted_files_count": 0,
        "added_rows_count": 237993,
        "existing_rows_count": 0,
        "deleted_rows_count": 0,
        "partitions": [
            {"contains_null": True, "contains_nan": False, "lower_bound": b"\x01\x00\x00\x00", "upper_bound": b"\x02\x00\x00\x00"}
        ],
        "key_metadata": b"\x19\x25",
    }
]


@pytest.fixture(scope="session")
def avro_schema_manifest_file_v1() -> dict[str, Any]:
    return {
        "type": "record",
        "name": "manifest_file",
        "fields": [
            {"name": "manifest_path", "type": "string", "doc": "Location URI with FS scheme", "field-id": 500},
            {"name": "manifest_length", "type": "long", "doc": "Total file size in bytes", "field-id": 501},
            {"name": "partition_spec_id", "type": "int", "doc": "Spec ID used to write", "field-id": 502},
            {
                "name": "added_snapshot_id",
                "type": ["null", "long"],
                "doc": "Snapshot ID that added the manifest",
                "default": None,
                "field-id": 503,
            },
            {
                "name": "added_data_files_count",
                "type": ["null", "int"],
                "doc": "Added entry count",
                "default": None,
                "field-id": 504,
            },
            {
                "name": "existing_data_files_count",
                "type": ["null", "int"],
                "doc": "Existing entry count",
                "default": None,
                "field-id": 505,
            },
            {
                "name": "deleted_data_files_count",
                "type": ["null", "int"],
                "doc": "Deleted entry count",
                "default": None,
                "field-id": 506,
            },
            {
                "name": "partitions",
                "type": [
                    "null",
                    {
                        "type": "array",
                        "items": {
                            "type": "record",
                            "name": "r508",
                            "fields": [
                                {
                                    "name": "contains_null",
                                    "type": "boolean",
                                    "doc": "True if any file has a null partition value",
                                    "field-id": 509,
                                },
                                {
                                    "name": "contains_nan",
                                    "type": ["null", "boolean"],
                                    "doc": "True if any file has a nan partition value",
                                    "default": None,
                                    "field-id": 518,
                                },
                                {
                                    "name": "lower_bound",
                                    "type": ["null", "bytes"],
                                    "doc": "Partition lower bound for all files",
                                    "default": None,
                                    "field-id": 510,
                                },
                                {
                                    "name": "upper_bound",
                                    "type": ["null", "bytes"],
                                    "doc": "Partition upper bound for all files",
                                    "default": None,
                                    "field-id": 511,
                                },
                            ],
                        },
                        "element-id": 508,
                    },
                ],
                "doc": "Summary for each partition",
                "default": None,
                "field-id": 507,
            },
            {"name": "added_rows_count", "type": ["null", "long"], "doc": "Added rows count", "default": None, "field-id": 512},
            {
                "name": "existing_rows_count",
                "type": ["null", "long"],
                "doc": "Existing rows count",
                "default": None,
                "field-id": 513,
            },
            {
                "name": "deleted_rows_count",
                "type": ["null", "long"],
                "doc": "Deleted rows count",
                "default": None,
                "field-id": 514,
            },
        ],
    }


@pytest.fixture(scope="session")
def avro_schema_manifest_file_v2() -> dict[str, Any]:
    return {
        "type": "record",
        "name": "manifest_file",
        "fields": [
            {"name": "manifest_path", "type": "string", "doc": "Location URI with FS scheme", "field-id": 500},
            {"name": "manifest_length", "type": "long", "doc": "Total file size in bytes", "field-id": 501},
            {"name": "partition_spec_id", "type": "int", "doc": "Spec ID used to write", "field-id": 502},
            {"name": "content", "type": "int", "doc": "Contents of the manifest: 0=data, 1=deletes", "field-id": 517},
            {
                "name": "sequence_number",
                "type": ["null", "long"],
                "doc": "Sequence number when the manifest was added",
                "field-id": 515,
            },
            {
                "name": "min_sequence_number",
                "type": ["null", "long"],
                "doc": "Lowest sequence number in the manifest",
                "field-id": 516,
            },
            {"name": "added_snapshot_id", "type": "long", "doc": "Snapshot ID that added the manifest", "field-id": 503},
            {"name": "added_files_count", "type": "int", "doc": "Added entry count", "field-id": 504},
            {"name": "existing_files_count", "type": "int", "doc": "Existing entry count", "field-id": 505},
            {"name": "deleted_files_count", "type": "int", "doc": "Deleted entry count", "field-id": 506},
            {"name": "added_rows_count", "type": "long", "doc": "Added rows count", "field-id": 512},
            {"name": "existing_rows_count", "type": "long", "doc": "Existing rows count", "field-id": 513},
            {"name": "deleted_rows_count", "type": "long", "doc": "Deleted rows count", "field-id": 514},
            {
                "name": "partitions",
                "type": [
                    "null",
                    {
                        "type": "array",
                        "items": {
                            "type": "record",
                            "name": "r508",
                            "fields": [
                                {
                                    "name": "contains_null",
                                    "type": "boolean",
                                    "doc": "True if any file has a null partition value",
                                    "field-id": 509,
                                },
                                {
                                    "name": "contains_nan",
                                    "type": ["null", "boolean"],
                                    "doc": "True if any file has a nan partition value",
                                    "default": None,
                                    "field-id": 518,
                                },
                                {
                                    "name": "lower_bound",
                                    "type": ["null", "bytes"],
                                    "doc": "Partition lower bound for all files",
                                    "default": None,
                                    "field-id": 510,
                                },
                                {
                                    "name": "upper_bound",
                                    "type": ["null", "bytes"],
                                    "doc": "Partition upper bound for all files",
                                    "default": None,
                                    "field-id": 511,
                                },
                            ],
                        },
                        "element-id": 508,
                    },
                ],
                "doc": "Summary for each partition",
                "default": None,
                "field-id": 507,
            },
        ],
    }


@pytest.fixture(scope="session")
def avro_schema_manifest_entry() -> dict[str, Any]:
    return {
        "type": "record",
        "name": "manifest_entry",
        "fields": [
            {"name": "status", "type": "int", "field-id": 0},
            {"name": "snapshot_id", "type": ["null", "long"], "default": None, "field-id": 1},
            {
                "name": "data_file",
                "type": {
                    "type": "record",
                    "name": "r2",
                    "fields": [
                        {"name": "file_path", "type": "string", "doc": "Location URI with FS scheme", "field-id": 100},
                        {
                            "name": "file_format",
                            "type": "string",
                            "doc": "File format name: avro, orc, or parquet",
                            "field-id": 101,
                        },
                        {
                            "name": "partition",
                            "type": {
                                "type": "record",
                                "name": "r102",
                                "fields": [
                                    {"field-id": 1000, "default": None, "name": "VendorID", "type": ["null", "int"]},
                                    {
                                        "field-id": 1001,
                                        "default": None,
                                        "name": "tpep_pickup_day",
                                        "type": ["null", {"type": "int", "logicalType": "date"}],
                                    },
                                ],
                            },
                            "field-id": 102,
                        },
                        {"name": "record_count", "type": "long", "doc": "Number of records in the file", "field-id": 103},
                        {"name": "file_size_in_bytes", "type": "long", "doc": "Total file size in bytes", "field-id": 104},
                        {"name": "block_size_in_bytes", "type": "long", "field-id": 105},
                        {
                            "name": "column_sizes",
                            "type": [
                                "null",
                                {
                                    "type": "array",
                                    "items": {
                                        "type": "record",
                                        "name": "k117_v118",
                                        "fields": [
                                            {"name": "key", "type": "int", "field-id": 117},
                                            {"name": "value", "type": "long", "field-id": 118},
                                        ],
                                    },
                                    "logicalType": "map",
                                },
                            ],
                            "doc": "Map of column id to total size on disk",
                            "default": None,
                            "field-id": 108,
                        },
                        {
                            "name": "value_counts",
                            "type": [
                                "null",
                                {
                                    "type": "array",
                                    "items": {
                                        "type": "record",
                                        "name": "k119_v120",
                                        "fields": [
                                            {"name": "key", "type": "int", "field-id": 119},
                                            {"name": "value", "type": "long", "field-id": 120},
                                        ],
                                    },
                                    "logicalType": "map",
                                },
                            ],
                            "doc": "Map of column id to total count, including null and NaN",
                            "default": None,
                            "field-id": 109,
                        },
                        {
                            "name": "null_value_counts",
                            "type": [
                                "null",
                                {
                                    "type": "array",
                                    "items": {
                                        "type": "record",
                                        "name": "k121_v122",
                                        "fields": [
                                            {"name": "key", "type": "int", "field-id": 121},
                                            {"name": "value", "type": "long", "field-id": 122},
                                        ],
                                    },
                                    "logicalType": "map",
                                },
                            ],
                            "doc": "Map of column id to null value count",
                            "default": None,
                            "field-id": 110,
                        },
                        {
                            "name": "nan_value_counts",
                            "type": [
                                "null",
                                {
                                    "type": "array",
                                    "items": {
                                        "type": "record",
                                        "name": "k138_v139",
                                        "fields": [
                                            {"name": "key", "type": "int", "field-id": 138},
                                            {"name": "value", "type": "long", "field-id": 139},
                                        ],
                                    },
                                    "logicalType": "map",
                                },
                            ],
                            "doc": "Map of column id to number of NaN values in the column",
                            "default": None,
                            "field-id": 137,
                        },
                        {
                            "name": "lower_bounds",
                            "type": [
                                "null",
                                {
                                    "type": "array",
                                    "items": {
                                        "type": "record",
                                        "name": "k126_v127",
                                        "fields": [
                                            {"name": "key", "type": "int", "field-id": 126},
                                            {"name": "value", "type": "bytes", "field-id": 127},
                                        ],
                                    },
                                    "logicalType": "map",
                                },
                            ],
                            "doc": "Map of column id to lower bound",
                            "default": None,
                            "field-id": 125,
                        },
                        {
                            "name": "upper_bounds",
                            "type": [
                                "null",
                                {
                                    "type": "array",
                                    "items": {
                                        "type": "record",
                                        "name": "k129_v130",
                                        "fields": [
                                            {"name": "key", "type": "int", "field-id": 129},
                                            {"name": "value", "type": "bytes", "field-id": 130},
                                        ],
                                    },
                                    "logicalType": "map",
                                },
                            ],
                            "doc": "Map of column id to upper bound",
                            "default": None,
                            "field-id": 128,
                        },
                        {
                            "name": "key_metadata",
                            "type": ["null", "bytes"],
                            "doc": "Encryption key metadata blob",
                            "default": None,
                            "field-id": 131,
                        },
                        {
                            "name": "split_offsets",
                            "type": ["null", {"type": "array", "items": "long", "element-id": 133}],
                            "doc": "Splittable offsets",
                            "default": None,
                            "field-id": 132,
                        },
                        {
                            "name": "sort_order_id",
                            "type": ["null", "int"],
                            "doc": "Sort order ID",
                            "default": None,
                            "field-id": 140,
                        },
                    ],
                },
                "field-id": 2,
            },
        ],
    }


@pytest.fixture(scope="session")
def simple_struct() -> StructType:
    return StructType(
        NestedField(id=1, name="required_field", field_type=StringType(), required=True, doc="this is a doc"),
        NestedField(id=2, name="optional_field", field_type=IntegerType()),
    )


@pytest.fixture(scope="session")
def simple_list() -> ListType:
    return ListType(element_id=22, element=StringType(), element_required=True)


@pytest.fixture(scope="session")
def simple_map() -> MapType:
    return MapType(key_id=19, key_type=StringType(), value_id=25, value_type=DoubleType(), value_required=False)


@pytest.fixture(scope="session")
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() -> PartitionSpec:
    return PartitionSpec(
        PartitionField(1, 1000, IdentityTransform(), "VendorID"),
        PartitionField(2, 1001, DayTransform(), "tpep_pickup_day"),
    )


@pytest.fixture(scope="session")
def test_sort_order() -> SortOrder:
    return SortOrder(SortField(source_id=1, transform=IdentityTransform(), null_order=NullOrder.NULLS_FIRST))


@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,
                metadata={
                    "schema": test_schema.model_dump_json(),
                    "partition-spec": to_json(test_partition_spec.fields).decode("utf-8"),
                },
            )
        yield tmp_avro_file


@pytest.fixture(scope="session")
def generated_manifest_file_file_v1(
    avro_schema_manifest_file_v1: dict[str, Any], generated_manifest_entry_file: str
) -> Generator[str, None, None]:
    from fastavro import parse_schema, writer

    parsed_schema = parse_schema(avro_schema_manifest_file_v1)

    # Make sure that a valid manifest_path is set
    manifest_file_records_v1[0]["manifest_path"] = generated_manifest_entry_file

    with TemporaryDirectory() as tmpdir:
        tmp_avro_file = tmpdir + "/manifest.avro"
        with open(tmp_avro_file, "wb") as out:
            writer(out, parsed_schema, manifest_file_records_v1)
        yield tmp_avro_file


@pytest.fixture(scope="session")
def generated_manifest_file_file_v2(
    avro_schema_manifest_file_v2: dict[str, Any], generated_manifest_entry_file: str
) -> Generator[str, None, None]:
    from fastavro import parse_schema, writer

    parsed_schema = parse_schema(avro_schema_manifest_file_v2)

    # Make sure that a valid manifest_path is set
    manifest_file_records_v2[0]["manifest_path"] = generated_manifest_entry_file

    with TemporaryDirectory() as tmpdir:
        tmp_avro_file = tmpdir + "/manifest.avro"
        with open(tmp_avro_file, "wb") as out:
            writer(out, parsed_schema, manifest_file_records_v2)
        yield tmp_avro_file


@pytest.fixture(scope="session")
def iceberg_manifest_entry_schema() -> Schema:
    return Schema(
        NestedField(field_id=0, name="status", field_type=IntegerType(), required=True),
        NestedField(field_id=1, name="snapshot_id", field_type=LongType(), required=False),
        NestedField(
            field_id=2,
            name="data_file",
            field_type=StructType(
                NestedField(
                    field_id=100,
                    name="file_path",
                    field_type=StringType(),
                    doc="Location URI with FS scheme",
                    required=True,
                ),
                NestedField(
                    field_id=101,
                    name="file_format",
                    field_type=StringType(),
                    doc="File format name: avro, orc, or parquet",
                    required=True,
                ),
                NestedField(
                    field_id=102,
                    name="partition",
                    field_type=StructType(
                        NestedField(
                            field_id=1000,
                            name="VendorID",
                            field_type=IntegerType(),
                            required=False,
                        ),
                        NestedField(
                            field_id=1001,
                            name="tpep_pickup_day",
                            field_type=DateType(),
                            required=False,
                        ),
                    ),
                    required=True,
                ),
                NestedField(
                    field_id=103,
                    name="record_count",
                    field_type=LongType(),
                    doc="Number of records in the file",
                    required=True,
                ),
                NestedField(
                    field_id=104,
                    name="file_size_in_bytes",
                    field_type=LongType(),
                    doc="Total file size in bytes",
                    required=True,
                ),
                NestedField(
                    field_id=105,
                    name="block_size_in_bytes",
                    field_type=LongType(),
                    required=True,
                ),
                NestedField(
                    field_id=108,
                    name="column_sizes",
                    field_type=MapType(
                        key_id=117,
                        key_type=IntegerType(),
                        value_id=118,
                        value_type=LongType(),
                        value_required=True,
                    ),
                    doc="Map of column id to total size on disk",
                    required=False,
                ),
                NestedField(
                    field_id=109,
                    name="value_counts",
                    field_type=MapType(
                        key_id=119,
                        key_type=IntegerType(),
                        value_id=120,
                        value_type=LongType(),
                        value_required=True,
                    ),
                    doc="Map of column id to total count, including null and NaN",
                    required=False,
                ),
                NestedField(
                    field_id=110,
                    name="null_value_counts",
                    field_type=MapType(
                        key_id=121,
                        key_type=IntegerType(),
                        value_id=122,
                        value_type=LongType(),
                        value_required=True,
                    ),
                    doc="Map of column id to null value count",
                    required=False,
                ),
                NestedField(
                    field_id=137,
                    name="nan_value_counts",
                    field_type=MapType(
                        key_id=138,
                        key_type=IntegerType(),
                        value_id=139,
                        value_type=LongType(),
                        value_required=True,
                    ),
                    doc="Map of column id to number of NaN values in the column",
                    required=False,
                ),
                NestedField(
                    field_id=125,
                    name="lower_bounds",
                    field_type=MapType(
                        key_id=126,
                        key_type=IntegerType(),
                        value_id=127,
                        value_type=BinaryType(),
                        value_required=True,
                    ),
                    doc="Map of column id to lower bound",
                    required=False,
                ),
                NestedField(
                    field_id=128,
                    name="upper_bounds",
                    field_type=MapType(
                        key_id=129,
                        key_type=IntegerType(),
                        value_id=130,
                        value_type=BinaryType(),
                        value_required=True,
                    ),
                    doc="Map of column id to upper bound",
                    required=False,
                ),
                NestedField(
                    field_id=131,
                    name="key_metadata",
                    field_type=BinaryType(),
                    doc="Encryption key metadata blob",
                    required=False,
                ),
                NestedField(
                    field_id=132,
                    name="split_offsets",
                    field_type=ListType(
                        element_id=133,
                        element_type=LongType(),
                        element_required=True,
                    ),
                    doc="Splittable offsets",
                    required=False,
                ),
                NestedField(
                    field_id=140,
                    name="sort_order_id",
                    field_type=IntegerType(),
                    doc="Sort order ID",
                    required=False,
                ),
            ),
            required=True,
        ),
        schema_id=1,
        identifier_field_ids=[],
    )


@pytest.fixture
def fsspec_fileio(request: pytest.FixtureRequest) -> FsspecFileIO:
    properties = {
        "s3.endpoint": request.config.getoption("--s3.endpoint"),
        "s3.access-key-id": request.config.getoption("--s3.access-key-id"),
        "s3.secret-access-key": request.config.getoption("--s3.secret-access-key"),
    }
    return fsspec.FsspecFileIO(properties=properties)


@pytest.fixture
def fsspec_fileio_gcs(request: pytest.FixtureRequest) -> FsspecFileIO:
    properties = {
        GCS_SERVICE_HOST: request.config.getoption("--gcs.endpoint"),
        GCS_TOKEN: request.config.getoption("--gcs.oauth2.token"),
        GCS_PROJECT_ID: request.config.getoption("--gcs.project-id"),
    }
    return fsspec.FsspecFileIO(properties=properties)


@pytest.fixture
def adls_fsspec_fileio(request: pytest.FixtureRequest) -> Generator[FsspecFileIO, None, None]:
    from azure.storage.blob import BlobServiceClient

    azurite_url = request.config.getoption("--adls.endpoint")
    azurite_account_name = request.config.getoption("--adls.account-name")
    azurite_account_key = request.config.getoption("--adls.account-key")
    azurite_connection_string = (
        f"DefaultEndpointsProtocol=http;AccountName={azurite_account_name};"
        f"AccountKey={azurite_account_key};BlobEndpoint={azurite_url}/{azurite_account_name};"
    )
    properties = {
        "adls.connection-string": azurite_connection_string,
        "adls.account-name": azurite_account_name,
    }

    bbs = BlobServiceClient.from_connection_string(conn_str=azurite_connection_string)
    bbs.create_container("tests")
    yield fsspec.FsspecFileIO(properties=properties)
    bbs.delete_container("tests")
    bbs.close()


@pytest.fixture
def pyarrow_fileio_gcs(request: pytest.FixtureRequest) -> "PyArrowFileIO":
    from pyiceberg.io.pyarrow import PyArrowFileIO

    properties = {
        GCS_SERVICE_HOST: request.config.getoption("--gcs.endpoint"),
        GCS_TOKEN: request.config.getoption("--gcs.oauth2.token"),
        GCS_PROJECT_ID: request.config.getoption("--gcs.project-id"),
        GCS_TOKEN_EXPIRES_AT_MS: datetime_to_millis(datetime.now()) + 60 * 1000,
    }
    return PyArrowFileIO(properties=properties)


@pytest.fixture
def pyarrow_fileio_adls(request: pytest.FixtureRequest) -> Generator[Any, None, None]:
    from azure.storage.blob import BlobServiceClient

    from pyiceberg.io.pyarrow import PyArrowFileIO

    azurite_url = request.config.getoption("--adls.endpoint")
    azurite_scheme, azurite_authority = azurite_url.split("://", 1)

    azurite_account_name = request.config.getoption("--adls.account-name")
    azurite_account_key = request.config.getoption("--adls.account-key")
    azurite_connection_string = (
        f"DefaultEndpointsProtocol=http;AccountName={azurite_account_name};"
        f"AccountKey={azurite_account_key};BlobEndpoint={azurite_url}/{azurite_account_name};"
    )
    properties = {
        ADLS_ACCOUNT_NAME: azurite_account_name,
        ADLS_ACCOUNT_KEY: azurite_account_key,
        ADLS_BLOB_STORAGE_AUTHORITY: azurite_authority,
        ADLS_DFS_STORAGE_AUTHORITY: azurite_authority,
        ADLS_BLOB_STORAGE_SCHEME: azurite_scheme,
        ADLS_DFS_STORAGE_SCHEME: azurite_scheme,
    }

    bbs = BlobServiceClient.from_connection_string(conn_str=azurite_connection_string)
    bbs.create_container("warehouse")
    yield PyArrowFileIO(properties=properties)
    bbs.delete_container("warehouse")
    bbs.close()


def aws_credentials() -> None:
    os.environ["AWS_ACCESS_KEY_ID"] = "testing"
    os.environ["AWS_SECRET_ACCESS_KEY"] = "testing"
    os.environ["AWS_SECURITY_TOKEN"] = "testing"
    os.environ["AWS_SESSION_TOKEN"] = "testing"
    os.environ["AWS_DEFAULT_REGION"] = "us-east-1"


@pytest.fixture(name="_aws_credentials")
def fixture_aws_credentials() -> Generator[None, None, None]:
    """Yield a mocked AWS Credentials for moto."""
    yield aws_credentials()  # type: ignore
    os.environ.pop("AWS_ACCESS_KEY_ID")
    os.environ.pop("AWS_SECRET_ACCESS_KEY")
    os.environ.pop("AWS_SECURITY_TOKEN")
    os.environ.pop("AWS_SESSION_TOKEN")
    os.environ.pop("AWS_DEFAULT_REGION")


@pytest.fixture(scope="session")
def moto_server() -> "ThreadedMotoServer":
    from moto.server import ThreadedMotoServer

    server = ThreadedMotoServer(ip_address="localhost", port=5001)

    # this will throw an exception if the port is already in use
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind((server._ip_address, server._port))

    server.start()
    yield server
    server.stop()


@pytest.fixture(scope="session")
def moto_endpoint_url(moto_server: "ThreadedMotoServer") -> str:
    _url = f"http://{moto_server._ip_address}:{moto_server._port}"
    return _url


@pytest.fixture(name="_s3", scope="function")
def fixture_s3(_aws_credentials: None, moto_endpoint_url: str) -> Generator[boto3.client, None, None]:
    """Yield a mocked S3 client."""
    with mock_aws():
        yield boto3.client("s3", region_name="us-east-1", endpoint_url=moto_endpoint_url)


@pytest.fixture(name="_glue")
def fixture_glue(_aws_credentials: None) -> Generator[boto3.client, None, None]:
    """Yield a mocked glue client."""
    with mock_aws():
        yield boto3.client("glue", region_name="us-east-1")


@pytest.fixture(name="_dynamodb")
def fixture_dynamodb(_aws_credentials: None) -> Generator[boto3.client, None, None]:
    """Yield a mocked DynamoDB client."""
    with mock_aws():
        yield boto3.client("dynamodb", region_name="us-east-1")


@pytest.fixture(scope="session")
def empty_home_dir_path(tmp_path_factory: pytest.TempPathFactory) -> str:
    home_path = str(tmp_path_factory.mktemp("home"))
    return home_path


RANDOM_LENGTH = 20
NUM_TABLES = 2


@pytest.fixture()
def table_name() -> str:
    prefix = "my_iceberg_table-"
    random_tag = "".join(choice(string.ascii_letters) for _ in range(RANDOM_LENGTH))
    return (prefix + random_tag).lower()


@pytest.fixture()
def table_list(table_name: str) -> list[str]:
    return [f"{table_name}_{idx}" for idx in range(NUM_TABLES)]


@pytest.fixture()
def view_name() -> str:
    prefix = "my_iceberg_view-"
    random_tag = "".join(choice(string.ascii_letters) for _ in range(RANDOM_LENGTH))
    return (prefix + random_tag).lower()


@pytest.fixture()
def database_name() -> str:
    prefix = "my_iceberg_database-"
    random_tag = "".join(choice(string.ascii_letters) for _ in range(RANDOM_LENGTH))
    return (prefix + random_tag).lower()


@pytest.fixture()
def gcp_dataset_name() -> str:
    prefix = "my_iceberg_database_"
    random_tag = "".join(choice(string.ascii_letters) for _ in range(RANDOM_LENGTH))
    return (prefix + random_tag).lower()


@pytest.fixture()
def database_list(database_name: str) -> list[str]:
    return [f"{database_name}_{idx}" for idx in range(NUM_TABLES)]


@pytest.fixture()
def hierarchical_namespace_name() -> str:
    prefix = "my_iceberg_ns-"
    random_tag1 = "".join(choice(string.ascii_letters) for _ in range(RANDOM_LENGTH))
    random_tag2 = "".join(choice(string.ascii_letters) for _ in range(RANDOM_LENGTH))
    return ".".join([prefix + random_tag1, prefix + random_tag2]).lower()


@pytest.fixture()
def hierarchical_namespace_list(hierarchical_namespace_name: str) -> list[str]:
    return [f"{hierarchical_namespace_name}_{idx}" for idx in range(NUM_TABLES)]


BUCKET_NAME = "test_bucket"
TABLE_METADATA_LOCATION_REGEX = re.compile(
    r"""s3://test_bucket/my_iceberg_database-[a-z]{20}.db/
    my_iceberg_table-[a-z]{20}/metadata/
    [0-9]{5}-[a-f0-9]{8}-?[a-f0-9]{4}-?4[a-f0-9]{3}-?[89ab][a-f0-9]{3}-?[a-f0-9]{12}.metadata.json""",
    re.X,
)

BQ_TABLE_METADATA_LOCATION_REGEX = re.compile(
    r"""gs://alexstephen-test-bq-bucket/my_iceberg_database_[a-z]{20}.db/
    my_iceberg_table-[a-z]{20}/metadata/
    [0-9]{5}-[a-f0-9]{8}-?[a-f0-9]{4}-?4[a-f0-9]{3}-?[89ab][a-f0-9]{3}-?[a-f0-9]{12}.metadata.json""",
    re.X,
)

UNIFIED_AWS_SESSION_PROPERTIES = {
    "client.access-key-id": "client.access-key-id",
    "client.secret-access-key": "client.secret-access-key",
    "client.region": "client.region",
    "client.session-token": "client.session-token",
}


@pytest.fixture(name="_bucket_initialize")
def fixture_s3_bucket(_s3) -> None:  # type: ignore
    _s3.create_bucket(Bucket=BUCKET_NAME)


def get_bucket_name() -> str:
    """Set the environment variable AWS_TEST_BUCKET for a default bucket to test."""
    bucket_name = os.getenv("AWS_TEST_BUCKET")
    if bucket_name is None:
        raise ValueError("Please specify a bucket to run the test by setting environment variable AWS_TEST_BUCKET")
    return bucket_name


def get_gcs_bucket_name() -> str:
    bucket_name = os.getenv("GCS_TEST_BUCKET")
    if bucket_name is None:
        raise ValueError("Please specify a bucket to run the test by setting environment variable GCS_TEST_BUCKET")
    return bucket_name


def get_glue_endpoint() -> str | None:
    """Set the optional environment variable AWS_TEST_GLUE_ENDPOINT for a glue endpoint to test."""
    return os.getenv("AWS_TEST_GLUE_ENDPOINT")


def get_s3_path(bucket_name: str, database_name: str | None = None, table_name: str | None = None) -> str:
    result_path = f"s3://{bucket_name}"
    if database_name is not None:
        result_path += f"/{database_name}.db"

    if table_name is not None:
        result_path += f"/{table_name}"
    return result_path


def get_gcs_path(bucket_name: str, database_name: str | None = None, table_name: str | None = None) -> str:
    result_path = f"gcs://{bucket_name}"
    if database_name is not None:
        result_path += f"/{database_name}.db"

    if table_name is not None:
        result_path += f"/{table_name}"
    return result_path


@pytest.fixture(name="s3", scope="module")
def fixture_s3_client() -> boto3.client:
    """Real S3 client for AWS Integration Tests."""
    yield boto3.client("s3")


def clean_up(test_catalog: Catalog) -> None:
    """Clean all databases and tables created during the integration test."""
    for database_tuple in test_catalog.list_namespaces():
        database_name = database_tuple[0]
        if "my_iceberg_database-" in database_name:
            for identifier in test_catalog.list_tables(database_name):
                test_catalog.drop_table(identifier)
            test_catalog.drop_namespace(database_name)


@pytest.fixture
def data_file(table_schema_simple: Schema, tmp_path: str) -> str:
    import pyarrow as pa
    from pyarrow import parquet as pq

    from pyiceberg.io.pyarrow import schema_to_pyarrow

    table = pa.table(
        {"foo": ["a", "b", "c"], "bar": [1, 2, 3], "baz": [True, False, None]},
        schema=schema_to_pyarrow(table_schema_simple),
    )

    file_path = f"{tmp_path}/0000-data.parquet"
    pq.write_table(table=table, where=file_path)
    return file_path


@pytest.fixture
def example_task(data_file: str) -> FileScanTask:
    datafile = DataFile.from_args(file_path=data_file, file_format=FileFormat.PARQUET, file_size_in_bytes=1925)
    datafile.spec_id = 0
    return FileScanTask(
        data_file=datafile,
    )


@pytest.fixture
def data_file_orc(table_schema_simple: Schema, tmp_path: str) -> str:
    import pyarrow as pa
    import pyarrow.orc as orc

    from pyiceberg.io.pyarrow import schema_to_pyarrow

    table = pa.table(
        {"foo": ["a", "b", "c"], "bar": [1, 2, 3], "baz": [True, False, None]},
        schema=schema_to_pyarrow(table_schema_simple),
    )

    file_path = f"{tmp_path}/0000-data.orc"
    orc.write_table(table=table, where=file_path)
    return file_path


@pytest.fixture
def example_task_orc(data_file_orc: str) -> FileScanTask:
    datafile = DataFile.from_args(file_path=data_file_orc, file_format=FileFormat.ORC, file_size_in_bytes=1925)
    datafile.spec_id = 0
    return FileScanTask(
        data_file=datafile,
    )


@pytest.fixture(scope="session")
def warehouse(tmp_path_factory: pytest.TempPathFactory) -> Path:
    return tmp_path_factory.mktemp("test_sql")


@pytest.fixture
def table_v1(example_table_metadata_v1: dict[str, Any]) -> Table:
    from pyiceberg.catalog.noop import NoopCatalog

    table_metadata = TableMetadataV1(**example_table_metadata_v1)
    return Table(
        identifier=("database", "table"),
        metadata=table_metadata,
        metadata_location=f"{table_metadata.location}/uuid.metadata.json",
        io=load_file_io(),
        catalog=NoopCatalog("NoopCatalog"),
    )


@pytest.fixture
def table_v2(example_table_metadata_v2: dict[str, Any]) -> Table:
    from pyiceberg.catalog.noop import NoopCatalog

    table_metadata = TableMetadataV2(**example_table_metadata_v2)
    return Table(
        identifier=("database", "table"),
        metadata=table_metadata,
        metadata_location=f"{table_metadata.location}/uuid.metadata.json",
        io=load_file_io(),
        catalog=NoopCatalog("NoopCatalog"),
    )


@pytest.fixture
def table_v3(example_table_metadata_v3: dict[str, Any]) -> Table:
    from pyiceberg.catalog.noop import NoopCatalog

    table_metadata = TableMetadataV3(**example_table_metadata_v3)
    return Table(
        identifier=("database", "table"),
        metadata=table_metadata,
        metadata_location=f"{table_metadata.location}/uuid.metadata.json",
        io=load_file_io(),
        catalog=NoopCatalog("NoopCatalog"),
    )


@pytest.fixture
def table_v2_orc(example_table_metadata_v2: dict[str, Any]) -> Table:
    import copy

    from pyiceberg.catalog.noop import NoopCatalog

    metadata_dict = copy.deepcopy(example_table_metadata_v2)
    if not metadata_dict["properties"]:
        metadata_dict["properties"] = {}
    metadata_dict["properties"]["write.format.default"] = "ORC"
    table_metadata = TableMetadataV2(**metadata_dict)
    return Table(
        identifier=("database", "table_orc"),
        metadata=table_metadata,
        metadata_location=f"{table_metadata.location}/uuid.metadata.json",
        io=load_file_io(),
        catalog=NoopCatalog("NoopCatalog"),
    )


@pytest.fixture
def table_v2_with_fixed_and_decimal_types(
    table_metadata_v2_with_fixed_and_decimal_types: dict[str, Any],
) -> Table:
    from pyiceberg.catalog.noop import NoopCatalog

    table_metadata = TableMetadataV2(
        **table_metadata_v2_with_fixed_and_decimal_types,
    )
    return Table(
        identifier=("database", "table"),
        metadata=table_metadata,
        metadata_location=f"{table_metadata.location}/uuid.metadata.json",
        io=load_file_io(),
        catalog=NoopCatalog("NoopCatalog"),
    )


@pytest.fixture
def table_v2_with_extensive_snapshots(example_table_metadata_v2_with_extensive_snapshots: dict[str, Any]) -> Table:
    from pyiceberg.catalog.noop import NoopCatalog

    table_metadata = TableMetadataV2(**example_table_metadata_v2_with_extensive_snapshots)
    return Table(
        identifier=("database", "table"),
        metadata=table_metadata,
        metadata_location=f"{table_metadata.location}/uuid.metadata.json",
        io=load_file_io(),
        catalog=NoopCatalog("NoopCatalog"),
    )


@pytest.fixture
def table_v2_with_statistics(table_metadata_v2_with_statistics: dict[str, Any]) -> Table:
    from pyiceberg.catalog.noop import NoopCatalog

    table_metadata = TableMetadataV2(**table_metadata_v2_with_statistics)
    return Table(
        identifier=("database", "table"),
        metadata=table_metadata,
        metadata_location=f"{table_metadata.location}/uuid.metadata.json",
        io=load_file_io(),
        catalog=NoopCatalog("NoopCatalog"),
    )


@pytest.fixture
def bound_reference_str() -> BoundReference:
    return BoundReference(field=NestedField(1, "field", StringType(), required=False), accessor=Accessor(position=0, inner=None))


@pytest.fixture
def bound_reference_binary() -> BoundReference:
    return BoundReference(field=NestedField(1, "field", BinaryType(), required=False), accessor=Accessor(position=0, inner=None))


@pytest.fixture
def bound_reference_uuid() -> BoundReference:
    return BoundReference(field=NestedField(1, "field", UUIDType(), required=False), accessor=Accessor(position=0, inner=None))


@pytest.fixture(scope="session")
def session_catalog() -> Catalog:
    return load_catalog(
        "local",
        **{
            "type": "rest",
            "uri": "http://localhost:8181",
            "s3.endpoint": "http://localhost:9000",
            "s3.access-key-id": "admin",
            "s3.secret-access-key": "password",
        },
    )


@pytest.fixture(scope="session")
def session_catalog_hive() -> Catalog:
    return load_catalog(
        "local",
        **{
            "type": "hive",
            "uri": "thrift://localhost:9083",
            "s3.endpoint": "http://localhost:9000",
            "s3.access-key-id": "admin",
            "s3.secret-access-key": "password",
        },
    )


@pytest.fixture(scope="session")
def spark() -> "SparkSession":
    from pyspark.sql import SparkSession

    # Create SparkSession against the remote Spark Connect server
    spark = SparkSession.builder.remote("sc://localhost:15002").getOrCreate()

    return spark


TEST_DATA_WITH_NULL = {
    "bool": [False, None, True],
    "string": ["a", None, "z"],
    # Go over the 16 bytes to kick in truncation
    "string_long": ["a" * 22, None, "z" * 22],
    "int": [1, None, 9],
    "long": [1, None, 9],
    "float": [0.0, None, 0.9],
    "double": [0.0, None, 0.9],
    # 'time': [1_000_000, None, 3_000_000],  # Example times: 1s, none, and 3s past midnight #Spark does not support time fields
    "timestamp": [datetime(2023, 1, 1, 19, 25, 00), None, datetime(2023, 3, 1, 19, 25, 00)],
    "timestamptz": [
        datetime(2023, 1, 1, 19, 25, 00, tzinfo=timezone.utc),
        None,
        datetime(2023, 3, 1, 19, 25, 00, tzinfo=timezone.utc),
    ],
    "date": [date(2023, 1, 1), None, date(2023, 3, 1)],
    # Not supported by Spark
    # 'time': [time(1, 22, 0), None, time(19, 25, 0)],
    # Not natively supported by Arrow
    # 'uuid': [uuid.UUID('00000000-0000-0000-0000-000000000000').bytes, None,
    #          uuid.UUID('11111111-1111-1111-1111-111111111111').bytes],
    "binary": [b"\01", None, b"\22"],
    "fixed": [
        uuid.UUID("00000000-0000-0000-0000-000000000000").bytes,
        None,
        uuid.UUID("11111111-1111-1111-1111-111111111111").bytes,
    ],
}


@pytest.fixture(scope="session")
def pa_schema() -> "pa.Schema":
    import pyarrow as pa

    return pa.schema(
        [
            ("bool", pa.bool_()),
            ("string", pa.large_string()),
            ("string_long", pa.large_string()),
            ("int", pa.int32()),
            ("long", pa.int64()),
            ("float", pa.float32()),
            ("double", pa.float64()),
            # Not supported by Spark
            # ("time", pa.time64('us')),
            ("timestamp", pa.timestamp(unit="us")),
            ("timestamptz", pa.timestamp(unit="us", tz="UTC")),
            ("date", pa.date32()),
            # Not supported by Spark
            # ("time", pa.time64("us")),
            # Not natively supported by Arrow
            # ("uuid", pa.fixed(16)),
            ("binary", pa.large_binary()),
            ("fixed", pa.binary(16)),
        ]
    )


@pytest.fixture(scope="session")
def arrow_table_with_null(pa_schema: "pa.Schema") -> "pa.Table":
    """Pyarrow table with all kinds of columns."""
    import pyarrow as pa

    return pa.Table.from_pydict(
        {
            "bool": [False, None, True],
            "string": ["a", None, "z"],
            # Go over the 16 bytes to kick in truncation
            "string_long": ["a" * 22, None, "z" * 22],
            "int": [1, None, 9],
            "long": [1, None, 9],
            "float": [0.0, None, 0.9],
            "double": [0.0, None, 0.9],
            # 'time': [1_000_000, None, 3_000_000],  # Example times: 1s, none, and 3s past midnight
            # Spark does not support time fields
            "timestamp": [datetime(2023, 1, 1, 19, 25, 00), None, datetime(2023, 3, 1, 19, 25, 00)],
            "timestamptz": [
                datetime(2023, 1, 1, 19, 25, 00, tzinfo=timezone.utc),
                None,
                datetime(2023, 3, 1, 19, 25, 00, tzinfo=timezone.utc),
            ],
            "date": [date(2023, 1, 1), None, date(2023, 3, 1)],
            # Not supported by Spark
            # 'time': [time(1, 22, 0), None, time(19, 25, 0)],
            # Not natively supported by Arrow
            # 'uuid': [uuid.UUID('00000000-0000-0000-0000-000000000000').bytes, None,
            #          uuid.UUID('11111111-1111-1111-1111-111111111111').bytes],
            "binary": [b"\01", None, b"\22"],
            "fixed": [
                uuid.UUID("00000000-0000-0000-0000-000000000000").bytes,
                None,
                uuid.UUID("11111111-1111-1111-1111-111111111111").bytes,
            ],
        },
        schema=pa_schema,
    )


@pytest.fixture(scope="session")
def arrow_table_without_data(pa_schema: "pa.Schema") -> "pa.Table":
    """Pyarrow table without data."""
    import pyarrow as pa

    return pa.Table.from_pylist([], schema=pa_schema)


@pytest.fixture(scope="session")
def arrow_table_with_only_nulls(pa_schema: "pa.Schema") -> "pa.Table":
    """Pyarrow table with only null values."""
    import pyarrow as pa

    return pa.Table.from_pylist([{}, {}], schema=pa_schema)


@pytest.fixture(scope="session")
def arrow_table_date_timestamps() -> "pa.Table":
    """Pyarrow table with only date, timestamp and timestamptz values."""
    import pyarrow as pa

    return pa.Table.from_pydict(
        {
            "date": [date(2023, 12, 31), date(2024, 1, 1), date(2024, 1, 31), date(2024, 2, 1), date(2024, 2, 1), None],
            "timestamp": [
                datetime(2023, 12, 31, 0, 0, 0),
                datetime(2024, 1, 1, 0, 0, 0),
                datetime(2024, 1, 31, 0, 0, 0),
                datetime(2024, 2, 1, 0, 0, 0),
                datetime(2024, 2, 1, 6, 0, 0),
                None,
            ],
            "timestamptz": [
                datetime(2023, 12, 31, 0, 0, 0, tzinfo=timezone.utc),
                datetime(2024, 1, 1, 0, 0, 0, tzinfo=timezone.utc),
                datetime(2024, 1, 31, 0, 0, 0, tzinfo=timezone.utc),
                datetime(2024, 2, 1, 0, 0, 0, tzinfo=timezone.utc),
                datetime(2024, 2, 1, 6, 0, 0, tzinfo=timezone.utc),
                None,
            ],
        },
        schema=pa.schema(
            [
                ("date", pa.date32()),
                ("timestamp", pa.timestamp(unit="us")),
                ("timestamptz", pa.timestamp(unit="us", tz="UTC")),
            ]
        ),
    )


@pytest.fixture(scope="session")
def table_date_timestamps_schema() -> Schema:
    """Iceberg table Schema with only date, timestamp and timestamptz values."""
    return Schema(
        NestedField(field_id=1, name="date", field_type=DateType(), required=False),
        NestedField(field_id=2, name="timestamp", field_type=TimestampType(), required=False),
        NestedField(field_id=3, name="timestamptz", field_type=TimestamptzType(), required=False),
    )


@pytest.fixture(scope="session")
def arrow_table_schema_with_all_timestamp_precisions() -> "pa.Schema":
    """Pyarrow Schema with all supported timestamp types."""
    import pyarrow as pa

    return pa.schema(
        [
            ("timestamp_s", pa.timestamp(unit="s")),
            ("timestamptz_s", pa.timestamp(unit="s", tz="UTC")),
            ("timestamp_ms", pa.timestamp(unit="ms")),
            ("timestamptz_ms", pa.timestamp(unit="ms", tz="UTC")),
            ("timestamp_us", pa.timestamp(unit="us")),
            ("timestamptz_us", pa.timestamp(unit="us", tz="UTC")),
            ("timestamp_ns", pa.timestamp(unit="ns")),
            ("timestamptz_ns", pa.timestamp(unit="ns", tz="UTC")),
            ("timestamptz_us_etc_utc", pa.timestamp(unit="us", tz="Etc/UTC")),
            ("timestamptz_ns_z", pa.timestamp(unit="ns", tz="Z")),
            ("timestamptz_s_0000", pa.timestamp(unit="s", tz="+00:00")),
        ]
    )


@pytest.fixture(scope="session")
def arrow_table_with_all_timestamp_precisions(arrow_table_schema_with_all_timestamp_precisions: "pa.Schema") -> "pa.Table":
    """Pyarrow table with all supported timestamp types."""
    import pandas as pd
    import pyarrow as pa

    test_data = pd.DataFrame(
        {
            "timestamp_s": [datetime(2023, 1, 1, 19, 25, 00), None, datetime(2023, 3, 1, 19, 25, 00)],
            "timestamptz_s": [
                datetime(2023, 1, 1, 19, 25, 00, tzinfo=timezone.utc),
                None,
                datetime(2023, 3, 1, 19, 25, 00, tzinfo=timezone.utc),
            ],
            "timestamp_ms": [datetime(2023, 1, 1, 19, 25, 00), None, datetime(2023, 3, 1, 19, 25, 00)],
            "timestamptz_ms": [
                datetime(2023, 1, 1, 19, 25, 00, tzinfo=timezone.utc),
                None,
                datetime(2023, 3, 1, 19, 25, 00, tzinfo=timezone.utc),
            ],
            "timestamp_us": [datetime(2023, 1, 1, 19, 25, 00), None, datetime(2023, 3, 1, 19, 25, 00)],
            "timestamptz_us": [
                datetime(2023, 1, 1, 19, 25, 00, tzinfo=timezone.utc),
                None,
                datetime(2023, 3, 1, 19, 25, 00, tzinfo=timezone.utc),
            ],
            "timestamp_ns": [
                pd.Timestamp(year=2024, month=7, day=11, hour=3, minute=30, second=0, microsecond=12, nanosecond=6),
                None,
                pd.Timestamp(year=2024, month=7, day=11, hour=3, minute=30, second=0, microsecond=12, nanosecond=7),
            ],
            "timestamptz_ns": [
                datetime(2023, 1, 1, 19, 25, 00, tzinfo=timezone.utc),
                None,
                datetime(2023, 3, 1, 19, 25, 00, tzinfo=timezone.utc),
            ],
            "timestamptz_us_etc_utc": [
                datetime(2023, 1, 1, 19, 25, 00, tzinfo=timezone.utc),
                None,
                datetime(2023, 3, 1, 19, 25, 00, tzinfo=timezone.utc),
            ],
            "timestamptz_ns_z": [
                pd.Timestamp(year=2024, month=7, day=11, hour=3, minute=30, second=0, microsecond=12, nanosecond=6, tz="UTC"),
                None,
                pd.Timestamp(year=2024, month=7, day=11, hour=3, minute=30, second=0, microsecond=12, nanosecond=7, tz="UTC"),
            ],
            "timestamptz_s_0000": [
                datetime(2023, 1, 1, 19, 25, 1, tzinfo=timezone.utc),
                None,
                datetime(2023, 3, 1, 19, 25, 1, tzinfo=timezone.utc),
            ],
        }
    )
    return pa.Table.from_pandas(test_data, schema=arrow_table_schema_with_all_timestamp_precisions)


@pytest.fixture(scope="session")
def arrow_table_schema_with_all_microseconds_timestamp_precisions() -> "pa.Schema":
    """Pyarrow Schema with all microseconds timestamp."""
    import pyarrow as pa

    return pa.schema(
        [
            ("timestamp_s", pa.timestamp(unit="us")),
            ("timestamptz_s", pa.timestamp(unit="us", tz="UTC")),
            ("timestamp_ms", pa.timestamp(unit="us")),
            ("timestamptz_ms", pa.timestamp(unit="us", tz="UTC")),
            ("timestamp_us", pa.timestamp(unit="us")),
            ("timestamptz_us", pa.timestamp(unit="us", tz="UTC")),
            ("timestamp_ns", pa.timestamp(unit="us")),
            ("timestamptz_ns", pa.timestamp(unit="us", tz="UTC")),
            ("timestamptz_us_etc_utc", pa.timestamp(unit="us", tz="UTC")),
            ("timestamptz_ns_z", pa.timestamp(unit="us", tz="UTC")),
            ("timestamptz_s_0000", pa.timestamp(unit="us", tz="UTC")),
        ]
    )


@pytest.fixture(scope="session")
def table_schema_with_all_microseconds_timestamp_precision() -> Schema:
    """Iceberg table Schema with only date, timestamp and timestamptz values."""
    return Schema(
        NestedField(field_id=1, name="timestamp_s", field_type=TimestampType(), required=False),
        NestedField(field_id=2, name="timestamptz_s", field_type=TimestamptzType(), required=False),
        NestedField(field_id=3, name="timestamp_ms", field_type=TimestampType(), required=False),
        NestedField(field_id=4, name="timestamptz_ms", field_type=TimestamptzType(), required=False),
        NestedField(field_id=5, name="timestamp_us", field_type=TimestampType(), required=False),
        NestedField(field_id=6, name="timestamptz_us", field_type=TimestamptzType(), required=False),
        NestedField(field_id=7, name="timestamp_ns", field_type=TimestampType(), required=False),
        NestedField(field_id=8, name="timestamptz_ns", field_type=TimestamptzType(), required=False),
        NestedField(field_id=9, name="timestamptz_us_etc_utc", field_type=TimestamptzType(), required=False),
        NestedField(field_id=10, name="timestamptz_ns_z", field_type=TimestamptzType(), required=False),
        NestedField(field_id=11, name="timestamptz_s_0000", field_type=TimestamptzType(), required=False),
    )


@pytest.fixture(scope="session")
def table_schema_with_promoted_types() -> Schema:
    """Iceberg table Schema with longs, doubles and uuid in simple and nested types."""
    return Schema(
        NestedField(field_id=1, name="long", field_type=LongType(), required=False),
        NestedField(
            field_id=2,
            name="list",
            field_type=ListType(element_id=4, element_type=LongType(), element_required=False),
            required=True,
        ),
        NestedField(
            field_id=3,
            name="map",
            field_type=MapType(
                key_id=5,
                key_type=StringType(),
                value_id=6,
                value_type=LongType(),
                value_required=False,
            ),
            required=True,
        ),
        NestedField(field_id=7, name="double", field_type=DoubleType(), required=False),
        NestedField(field_id=8, name="uuid", field_type=UUIDType(), required=False),
    )


@pytest.fixture(scope="session")
def pyarrow_schema_with_promoted_types() -> "pa.Schema":
    """Pyarrow Schema with longs, doubles and uuid in simple and nested types."""
    import pyarrow as pa

    return pa.schema(
        (
            pa.field("long", pa.int32(), nullable=True),  # can support upcasting integer to long
            pa.field("list", pa.list_(pa.int32()), nullable=False),  # can support upcasting integer to long
            pa.field("map", pa.map_(pa.string(), pa.int32()), nullable=False),  # can support upcasting integer to long
            pa.field("double", pa.float32(), nullable=True),  # can support upcasting float to double
            pa.field("uuid", pa.binary(length=16), nullable=True),  # can support upcasting fixed to uuid
        )
    )


@pytest.fixture(scope="session")
def pyarrow_table_with_promoted_types(pyarrow_schema_with_promoted_types: "pa.Schema") -> "pa.Table":
    """Pyarrow table with longs, doubles and uuid in simple and nested types."""
    import pyarrow as pa

    return pa.Table.from_pydict(
        {
            "long": [1, 9],
            "list": [[1, 1], [2, 2]],
            "map": [{"a": 1}, {"b": 2}],
            "double": [1.1, 9.2],
            "uuid": [
                uuid.UUID("00000000-0000-0000-0000-000000000000").bytes,
                uuid.UUID("11111111-1111-1111-1111-111111111111").bytes,
            ],
        },
        schema=pyarrow_schema_with_promoted_types,
    )


def pytest_configure(config: pytest.Config) -> None:
    # Disable Ray's automatic uv environment propagation to workers.
    # When tests are invoked via `uv run`, Ray detects the wrapper and tries
    # to package the working directory for remote workers, which is unnecessary
    # and problematic in local single-node test mode
    # See https://docs.ray.io/en/latest/_modules/ray/_private/worker.html
    os.environ["RAY_ENABLE_UV_RUN_RUNTIME_ENV"] = "0"


@pytest.fixture(scope="session")
def ray_session() -> Generator[Any, None, None]:
    """Fixture to manage Ray initialization and shutdown for tests."""
    import ray

    ray.init(
        ignore_reinit_error=True,
    )
    yield ray
    ray.shutdown()


# Catalog fixtures


def _create_memory_catalog(name: str, warehouse: Path) -> Catalog:
    from pyiceberg.catalog.memory import InMemoryCatalog

    return InMemoryCatalog(name, warehouse=f"file://{warehouse}")


def _create_sql_catalog(name: str, warehouse: Path) -> Catalog:
    from pyiceberg.catalog.sql import SqlCatalog

    catalog = SqlCatalog(
        name,
        uri="sqlite:///:memory:",
        warehouse=f"file://{warehouse}",
    )
    catalog.create_tables()
    return catalog


def _create_sql_without_rowcount_catalog(name: str, warehouse: Path) -> Catalog:
    from pyiceberg.catalog.sql import SqlCatalog

    props = {
        "uri": f"sqlite:////{warehouse}/sql-catalog",
        "warehouse": f"file://{warehouse}",
    }
    catalog = SqlCatalog(name, **props)
    catalog.engine.dialect.supports_sane_rowcount = False
    catalog.create_tables()
    return catalog


_CATALOG_FACTORIES = {
    "memory": _create_memory_catalog,
    "sql": _create_sql_catalog,
    "sql_without_rowcount": _create_sql_without_rowcount_catalog,
}


@pytest.fixture(params=list(_CATALOG_FACTORIES.keys()))
def catalog(request: pytest.FixtureRequest, tmp_path: Path) -> Generator[Catalog, None, None]:
    """Parameterized fixture that yields catalogs listed in _CATALOG_FACTORIES."""
    catalog_type = request.param
    factory = _CATALOG_FACTORIES[catalog_type]
    cat = factory("test_catalog", tmp_path)
    yield cat
    if hasattr(cat, "destroy_tables"):
        cat.destroy_tables()


@pytest.fixture(params=list(_CATALOG_FACTORIES.keys()))
def catalog_with_warehouse(
    request: pytest.FixtureRequest,
    warehouse: Path,
) -> Generator[Catalog, None, None]:
    factory = _CATALOG_FACTORIES[request.param]
    cat = factory("test_catalog", warehouse)
    yield cat
    if hasattr(cat, "destroy_tables"):
        cat.destroy_tables()


@pytest.fixture(name="random_table_identifier")
def fixture_random_table_identifier(warehouse: Path, database_name: str, table_name: str) -> Identifier:
    os.makedirs(f"{warehouse}/{database_name}/{table_name}/metadata/", exist_ok=True)
    return database_name, table_name


@pytest.fixture(name="another_random_table_identifier")
def fixture_another_random_table_identifier(warehouse: Path, database_name: str, table_name: str) -> Identifier:
    database_name = database_name + "_new"
    table_name = table_name + "_new"
    os.makedirs(f"{warehouse}/{database_name}/{table_name}/metadata/", exist_ok=True)
    return database_name, table_name


@pytest.fixture(name="random_hierarchical_identifier")
def fixture_random_hierarchical_identifier(warehouse: Path, hierarchical_namespace_name: str, table_name: str) -> Identifier:
    os.makedirs(f"{warehouse}/{hierarchical_namespace_name}/{table_name}/metadata/", exist_ok=True)
    return Catalog.identifier_to_tuple(".".join((hierarchical_namespace_name, table_name)))


@pytest.fixture(name="another_random_hierarchical_identifier")
def fixture_another_random_hierarchical_identifier(
    warehouse: Path, hierarchical_namespace_name: str, table_name: str
) -> Identifier:
    hierarchical_namespace_name = hierarchical_namespace_name + "_new"
    table_name = table_name + "_new"
    os.makedirs(f"{warehouse}/{hierarchical_namespace_name}/{table_name}/metadata/", exist_ok=True)
    return Catalog.identifier_to_tuple(".".join((hierarchical_namespace_name, table_name)))


@pytest.fixture(scope="session")
def fixed_test_table_identifier() -> Identifier:
    return "com", "organization", "department", "my_table"


@pytest.fixture(scope="session")
def another_fixed_test_table_identifier() -> Identifier:
    return "com", "organization", "department_alt", "my_another_table"


@pytest.fixture(scope="session")
def fixed_test_table_namespace() -> Identifier:
    return "com", "organization", "department"


@pytest.fixture(
    params=[
        lf("fixed_test_table_identifier"),
        lf("random_table_identifier"),
        lf("random_hierarchical_identifier"),
    ],
)
def test_table_identifier(request: pytest.FixtureRequest) -> Identifier:
    return request.param


@pytest.fixture(
    params=[
        lf("another_fixed_test_table_identifier"),
        lf("another_random_table_identifier"),
        lf("another_random_hierarchical_identifier"),
    ],
)
def another_table_identifier(request: pytest.FixtureRequest) -> Identifier:
    return request.param


@pytest.fixture(
    params=[
        lf("database_name"),
        lf("hierarchical_namespace_name"),
        lf("fixed_test_table_namespace"),
    ],
)
def test_namespace(request: pytest.FixtureRequest) -> Identifier:
    ns = request.param
    if isinstance(ns, tuple):
        return ns
    if "." in ns:
        return tuple(ns.split("."))
    return (ns,)


@pytest.fixture(scope="session")
def test_namespace_properties() -> dict[str, str]:
    return {"key1": "value1", "key2": "value2"}


@pytest.fixture(scope="session")
def test_table_properties() -> dict[str, str]:
    return {
        "key1": "value1",
        "key2": "value2",
    }


def does_support_purge_table(catalog: Catalog) -> bool:
    from pyiceberg.catalog.noop import NoopCatalog
    from pyiceberg.catalog.rest import RestCatalog

    if isinstance(catalog, RestCatalog):
        return property_as_bool(catalog.properties, "supports_purge_table", True)
    from pyiceberg.catalog.hive import HiveCatalog

    if isinstance(catalog, (HiveCatalog, NoopCatalog)):
        return False
    return True


def does_support_atomic_concurrent_updates(catalog: Catalog) -> bool:
    from pyiceberg.catalog.noop import NoopCatalog
    from pyiceberg.catalog.rest import RestCatalog

    if isinstance(catalog, RestCatalog):
        return property_as_bool(catalog.properties, "supports_atomic_concurrent_updates", True)
    from pyiceberg.catalog.hive import HiveCatalog

    if isinstance(catalog, (HiveCatalog, NoopCatalog)):
        return False
    return True


def does_support_nested_namespaces(catalog: Catalog) -> bool:
    from pyiceberg.catalog.dynamodb import DynamoDbCatalog
    from pyiceberg.catalog.glue import GlueCatalog
    from pyiceberg.catalog.noop import NoopCatalog
    from pyiceberg.catalog.rest import RestCatalog

    if isinstance(catalog, RestCatalog):
        return property_as_bool(catalog.properties, "supports_nested_namespaces", True)
    from pyiceberg.catalog.bigquery_metastore import BigQueryMetastoreCatalog
    from pyiceberg.catalog.hive import HiveCatalog

    if isinstance(catalog, (HiveCatalog, BigQueryMetastoreCatalog, NoopCatalog, GlueCatalog, DynamoDbCatalog)):
        return False
    return True


def does_support_schema_evolution(catalog: Catalog) -> bool:
    from pyiceberg.catalog.noop import NoopCatalog
    from pyiceberg.catalog.rest import RestCatalog

    if isinstance(catalog, RestCatalog):
        return property_as_bool(catalog.properties, "supports_schema_evolution", True)
    from pyiceberg.catalog.hive import HiveCatalog

    if isinstance(catalog, (HiveCatalog, NoopCatalog)):
        return False
    return True


def does_support_slash_in_identifier(catalog: Catalog) -> bool:
    from pyiceberg.catalog.noop import NoopCatalog
    from pyiceberg.catalog.rest import RestCatalog
    from pyiceberg.catalog.sql import SqlCatalog

    if isinstance(catalog, RestCatalog):
        return property_as_bool(catalog.properties, "supports_slash_in_identifier", True)
    from pyiceberg.catalog.hive import HiveCatalog

    if isinstance(catalog, (HiveCatalog, NoopCatalog, SqlCatalog)):
        return False
    return True


def does_support_dot_in_identifier(catalog: Catalog) -> bool:
    from pyiceberg.catalog.noop import NoopCatalog
    from pyiceberg.catalog.rest import RestCatalog
    from pyiceberg.catalog.sql import SqlCatalog

    if isinstance(catalog, RestCatalog):
        return property_as_bool(catalog.properties, "supports_dot_in_identifier", True)
    from pyiceberg.catalog.hive import HiveCatalog

    if isinstance(catalog, (HiveCatalog, NoopCatalog, SqlCatalog)):
        return False
    return True