import json

import uuid

from copy import copy

from typing import Any, Dict

import pytest

from pydantic import ValidationError

from sortedcontainers import SortedList

from pyiceberg.catalog.noop import NoopCatalog

from pyiceberg.exceptions import CommitFailedException

from pyiceberg.expressions import (

AlwaysTrue,

And,

EqualTo,

In,

)

from pyiceberg.io import PY_IO_IMPL, load_file_io

from pyiceberg.manifest import (

DataFile,

DataFileContent,

FileFormat,

ManifestEntry,

ManifestEntryStatus,

)

from pyiceberg.partitioning import PartitionField, PartitionSpec

from pyiceberg.schema import Schema

from pyiceberg.table import (

CommitTableRequest,

StaticTable,

Table,

TableIdentifier,

_match_deletes_to_data_file,

)

from pyiceberg.table.metadata import INITIAL_SEQUENCE_NUMBER, TableMetadataUtil, TableMetadataV2, _generate_snapshot_id

from pyiceberg.table.refs import SnapshotRef

from pyiceberg.table.snapshots import (

MetadataLogEntry,

Operation,

Snapshot,

SnapshotLogEntry,

Summary,

ancestors_of,

)

from pyiceberg.table.sorting import (

NullOrder,

SortDirection,

SortField,

SortOrder,

)

from pyiceberg.table.statistics import BlobMetadata, StatisticsFile

from pyiceberg.table.update import (

AddSnapshotUpdate,

AddSortOrderUpdate,

AssertCreate,

AssertCurrentSchemaId,

AssertDefaultSortOrderId,

AssertDefaultSpecId,

AssertLastAssignedFieldId,

AssertLastAssignedPartitionId,

AssertRefSnapshotId,

AssertTableUUID,

RemovePropertiesUpdate,

RemoveSnapshotRefUpdate,

RemoveSnapshotsUpdate,

RemoveStatisticsUpdate,

SetDefaultSortOrderUpdate,

SetPropertiesUpdate,

SetSnapshotRefUpdate,

SetStatisticsUpdate,

_apply_table_update,

_TableMetadataUpdateContext,

update_table_metadata,

)

from pyiceberg.table.update.schema import UpdateSchema

from pyiceberg.transforms import (

BucketTransform,

IdentityTransform,

)

from pyiceberg.types import (

BinaryType,

BooleanType,

DateType,

DecimalType,

DoubleType,

FixedType,

FloatType,

IntegerType,

ListType,

LongType,

MapType,

NestedField,

PrimitiveType,

StringType,

StructType,

TimestampType,

TimestamptzType,

TimeType,

UUIDType,

)

def test_schema(table_v2: Table) -> None:

assert table_v2.schema().schema_id == 1

def test_schemas(table_v2: Table) -> None:

assert table_v2.schemas()[1].schema_id == 1

def test_spec(table_v2: Table) -> None:

assert table_v2.spec() == PartitionSpec(

PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="x"), spec_id=0

)

def test_specs(table_v2: Table) -> None:

assert table_v2.specs() == {

0: PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="x"), spec_id=0)

}

def test_sort_order(table_v2: Table) -> None:

)

def test_sort_orders(table_v2: Table) -> None:

}

def test_location(table_v2: Table) -> None:

assert table_v2.location() == "s3://bucket/test/location"

def test_current_snapshot(table_v2: Table) -> None:

)

def test_snapshot_by_id(table_v2: Table) -> None:

)

def test_snapshot_by_timestamp(table_v2: Table) -> None:

assert table_v2.snapshot_as_of_timestamp(1515100955770, inclusive=False) is None

def test_ancestors_of(table_v2: Table) -> None:

]

def test_ancestors_of_recursive_error(table_v2_with_extensive_snapshots: Table) -> None:

)

def test_snapshot_by_id_does_not_exist(table_v2: Table) -> None:

assert table_v2.snapshot_by_id(-1) is None

def test_snapshot_by_name(table_v2: Table) -> None:

)

def test_snapshot_by_name_does_not_exist(table_v2: Table) -> None:

assert table_v2.snapshot_by_name("doesnotexist") is None

def test_repr(table_v2: Table) -> None:

assert repr(table_v2) == expected

def test_history(table_v2: Table) -> None:

]

def test_table_scan_select(table_v2: Table) -> None:

assert scan.select("a", "c").select("a").selected_fields == ("a",)

def test_table_scan_row_filter(table_v2: Table) -> None:

assert scan.filter(EqualTo("x", 10)).filter(In("y", (10, 11))).row_filter == And(EqualTo("x", 10), In("y", (10, 11)))

def test_table_scan_ref(table_v2: Table) -> None:

scan = table_v2.scan()

assert scan.use_ref("test").snapshot_id == 3051729675574597004

def test_table_scan_ref_does_not_exists(table_v2: Table) -> None:

assert "Cannot scan unknown ref=boom" in str(exc_info.value)

def test_table_scan_projection_full_schema(table_v2: Table) -> None:

assert projection_schema.schema_id == 1

def test_table_scan_projection_single_column(table_v2: Table) -> None:

assert projection_schema.schema_id == 1

def test_table_scan_projection_single_column_case_sensitive(table_v2: Table) -> None:

assert projection_schema.schema_id == 1

def test_table_scan_projection_unknown_column(table_v2: Table) -> None:

assert "Could not find column: 'a'" in str(exc_info.value)

def test_static_table_same_as_table(table_v2: Table, metadata_location: str) -> None:

static_table = StaticTable.from_metadata(metadata_location)

assert isinstance(static_table, Table)

assert static_table.metadata == table_v2.metadata

def test_static_table_gz_same_as_table(table_v2: Table, metadata_location_gz: str) -> None:

static_table = StaticTable.from_metadata(metadata_location_gz)

assert isinstance(static_table, Table)

assert static_table.metadata == table_v2.metadata

def test_static_table_io_does_not_exist(metadata_location: str) -> None:

with pytest.raises(ValueError):

StaticTable.from_metadata(metadata_location, {PY_IO_IMPL: "pyiceberg.does.not.exist.FileIO"})

def test_match_deletes_to_datafile() -> None:

}

def test_match_deletes_to_datafile_duplicate_number() -> None:

}

def test_serialize_set_properties_updates() -> None:

assert (

SetPropertiesUpdate(updates={"abc": "🤪"}).model_dump_json() == """{"action":"set-properties","updates":{"abc":"🤪"}}"""

)

def test_add_column(table_v2: Table) -> None:

assert apply_schema.highest_field_id == 4

def test_update_column(table_v1: Table, table_v2: Table) -> None:

), "failed to update existing field required with union_by_name"

def test_add_primitive_type_column(table_v2: Table) -> None:

assert field.doc == f"new_column_{name}"

def test_add_nested_type_column(table_v2: Table) -> None:

assert schema_.highest_field_id == 6

def test_add_nested_map_type_column(table_v2: Table) -> None:

assert new_schema.highest_field_id == 6

def test_add_nested_list_type_column(table_v2: Table) -> None:

assert new_schema.highest_field_id == 7

def test_apply_set_properties_update(table_v2: Table) -> None:

assert new_metadata_add_only.last_updated_ms > base_metadata.last_updated_ms

def test_apply_remove_properties_update(table_v2: Table) -> None:

assert new_metadata.properties == {"read.split.target.size": "134217728", "test_b": "test_b", "test_d": "test_d"}

def test_apply_add_schema_update(table_v2: Table) -> None:

assert test_context.is_added_schema(2)

def test_update_metadata_table_schema(table_v2: Table) -> None:

assert new_metadata.current_schema_id == 2

def test_update_metadata_add_snapshot(table_v2: Table) -> None:

assert new_metadata.last_updated_ms == new_snapshot.timestamp_ms

def test_update_metadata_set_ref_snapshot(table_v2: Table) -> None:

)

def test_update_metadata_set_snapshot_ref(table_v2: Table) -> None:

)

def test_update_remove_snapshots(table_v2: Table) -> None:

assert new_metadata.refs["main"].snapshot_id == KEEP_SNAPSHOT

def test_update_remove_snapshots_doesnt_exist(table_v2: Table) -> None:

update_table_metadata(table_v2.metadata, (update,))

def test_update_remove_snapshots_remove_current_snapshot_id(table_v2: Table) -> None:

assert new_metadata.current_snapshot_id is None

def test_update_remove_snapshot_ref(table_v2: Table) -> None:

assert new_metadata.refs["main"].snapshot_id == 3055729675574597004

def test_update_metadata_add_update_sort_order(table_v2: Table) -> None:

assert new_metadata.last_updated_ms > table_v2.metadata.last_updated_ms

def test_update_metadata_update_sort_order_invalid(table_v2: Table) -> None:

update_table_metadata(table_v2.metadata, (SetDefaultSortOrderUpdate(sort_order_id=invalid_order_id),))

def test_update_metadata_with_multiple_updates(table_v1: Table) -> None:

assert new_metadata.properties == {"owner": "test", "test_a": "test_a1"}

def test_update_metadata_schema_immutability(

)

def test_metadata_isolation_from_illegal_updates(table_v1: Table) -> None: