Google Git Google Git
Sign in
apache/iceberg-python/76eb1d4f25c3ff8b8e96a8951cc3cebb6da170f2/./tests/table/test_init.py
blob: 739039debb21568c9ad66f7e70243e691057c950 [file]
# 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
import json
import uuid
from copy import copy
from typing import Any
import pytest
from pydantic import BaseModel, ValidationError
from pytest_lazy_fixtures import lf
from pyiceberg.catalog import Catalog
from pyiceberg.catalog.noop import NoopCatalog
from pyiceberg.exceptions import CommitFailedException
from pyiceberg.expressions import (
AlwaysTrue,
And,
EqualTo,
In,
)
from pyiceberg.expressions.visitors import bind
from pyiceberg.io import PY_IO_IMPL, FileIO, load_file_io
from pyiceberg.partitioning import PartitionField, PartitionSpec
from pyiceberg.schema import Schema
from pyiceberg.table import (
CommitTableRequest,
StagedTable,
StaticTable,
Table,
TableIdentifier,
)
from pyiceberg.table.metadata import TableMetadataUtil, TableMetadataV1, TableMetadataV2, _generate_snapshot_id
from pyiceberg.table.refs import MAIN_BRANCH, SnapshotRef, SnapshotRefType
from pyiceberg.table.snapshots import (
MetadataLogEntry,
Operation,
Snapshot,
SnapshotLogEntry,
Summary,
)
from pyiceberg.table.sorting import (
NullOrder,
SortDirection,
SortField,
SortOrder,
)
from pyiceberg.table.statistics import BlobMetadata, PartitionStatisticsFile, StatisticsFile
from pyiceberg.table.update import (
AddPartitionSpecUpdate,
AddSnapshotUpdate,
AddSortOrderUpdate,
AssertCreate,
AssertCurrentSchemaId,
AssertDefaultSortOrderId,
AssertDefaultSpecId,
AssertLastAssignedFieldId,
AssertLastAssignedPartitionId,
AssertRefSnapshotId,
AssertTableUUID,
RemovePartitionSpecsUpdate,
RemovePartitionStatisticsUpdate,
RemovePropertiesUpdate,
RemoveSchemasUpdate,
RemoveSnapshotRefUpdate,
RemoveSnapshotsUpdate,
RemoveStatisticsUpdate,
SetDefaultSortOrderUpdate,
SetPartitionStatisticsUpdate,
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.typedef import Record
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(
NestedField(field_id=1, name="x", field_type=LongType(), required=True),
NestedField(field_id=2, name="y", field_type=LongType(), required=True, doc="comment"),
NestedField(field_id=3, name="z", field_type=LongType(), required=True),
identifier_field_ids=[1, 2],
)
assert table_v2.schema().schema_id == 1
def test_schemas(table_v2: Table) -> None:
assert table_v2.schemas() == {
0: Schema(
NestedField(field_id=1, name="x", field_type=LongType(), required=True),
identifier_field_ids=[],
),
1: Schema(
NestedField(field_id=1, name="x", field_type=LongType(), required=True),
NestedField(field_id=2, name="y", field_type=LongType(), required=True, doc="comment"),
NestedField(field_id=3, name="z", field_type=LongType(), required=True),
identifier_field_ids=[1, 2],
),
}
assert table_v2.schemas()[0].schema_id == 0
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:
assert table_v2.sort_order() == SortOrder(
SortField(source_id=2, transform=IdentityTransform(), direction=SortDirection.ASC, null_order=NullOrder.NULLS_FIRST),
SortField(
source_id=3,
transform=BucketTransform(num_buckets=4),
direction=SortDirection.DESC,
null_order=NullOrder.NULLS_LAST,
),
order_id=3,
)
def test_sort_orders(table_v2: Table) -> None:
assert table_v2.sort_orders() == {
3: SortOrder(
SortField(source_id=2, transform=IdentityTransform(), direction=SortDirection.ASC, null_order=NullOrder.NULLS_FIRST),
SortField(
source_id=3,
transform=BucketTransform(num_buckets=4),
direction=SortDirection.DESC,
null_order=NullOrder.NULLS_LAST,
),
order_id=3,
)
}
def test_location(table_v2: Table) -> None:
assert table_v2.location() == "s3://bucket/test/location"
def test_current_snapshot(table_v2: Table) -> None:
assert table_v2.current_snapshot() == Snapshot(
snapshot_id=3055729675574597004,
parent_snapshot_id=3051729675574597004,
sequence_number=1,
timestamp_ms=1555100955770,
manifest_list="s3://a/b/2.avro",
summary=Summary(operation=Operation.APPEND),
schema_id=1,
)
def test_snapshot_by_id(table_v2: Table) -> None:
assert table_v2.snapshot_by_id(3055729675574597004) == Snapshot(
snapshot_id=3055729675574597004,
parent_snapshot_id=3051729675574597004,
sequence_number=1,
timestamp_ms=1555100955770,
manifest_list="s3://a/b/2.avro",
summary=Summary(operation=Operation.APPEND),
schema_id=1,
)
def test_snapshot_by_timestamp(table_v2: Table) -> None:
assert table_v2.snapshot_as_of_timestamp(1515100955770) == Snapshot(
snapshot_id=3051729675574597004,
parent_snapshot_id=None,
sequence_number=0,
timestamp_ms=1515100955770,
manifest_list="s3://a/b/1.avro",
summary=Summary(Operation.APPEND),
schema_id=None,
)
assert table_v2.snapshot_as_of_timestamp(1515100955770, inclusive=False) is 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:
assert table_v2.snapshot_by_name("test") == Snapshot(
snapshot_id=3051729675574597004,
parent_snapshot_id=None,
sequence_number=0,
timestamp_ms=1515100955770,
manifest_list="s3://a/b/1.avro",
summary=Summary(operation=Operation.APPEND),
schema_id=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:
expected = """table(
1: x: required long,
2: y: required long (comment),
3: z: required long
),
partition by: [x],
sort order: [2 ASC NULLS FIRST, bucket[4](3) DESC NULLS LAST],
snapshot: Operation.APPEND: id=3055729675574597004, parent_id=3051729675574597004, schema_id=1"""
assert repr(table_v2) == expected
def test_history(table_v2: Table) -> None:
assert table_v2.history() == [
SnapshotLogEntry(snapshot_id=3051729675574597004, timestamp_ms=1515100955770),
SnapshotLogEntry(snapshot_id=3055729675574597004, timestamp_ms=1555100955770),
]
@pytest.mark.parametrize(
"table_fixture",
[
pytest.param(lf("table_v2"), id="parquet"),
pytest.param(lf("table_v2_orc"), id="orc"),
],
)
def test_table_scan_select(table_fixture: Table) -> None:
scan = table_fixture.scan()
assert scan.selected_fields == ("*",)
assert scan.select("a", "b").selected_fields == ("a", "b")
assert scan.select("a", "c").select("a").selected_fields == ("a",)
def test_table_scan_row_filter(table_v2: Table) -> None:
scan = table_v2.scan()
assert scan.row_filter == AlwaysTrue()
assert scan.filter(EqualTo("x", 10)).row_filter == EqualTo("x", 10)
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:
scan = table_v2.scan()
with pytest.raises(ValueError) as exc_info:
_ = scan.use_ref("boom")
assert "Cannot scan unknown ref=boom" in str(exc_info.value)
def test_table_scan_projection_full_schema(table_v2: Table) -> None:
scan = table_v2.scan()
projection_schema = scan.select("x", "y", "z").projection()
assert projection_schema == Schema(
NestedField(field_id=1, name="x", field_type=LongType(), required=True),
NestedField(field_id=2, name="y", field_type=LongType(), required=True, doc="comment"),
NestedField(field_id=3, name="z", field_type=LongType(), required=True),
identifier_field_ids=[1, 2],
)
assert projection_schema.schema_id == 1
def test_table_scan_projection_single_column(table_v2: Table) -> None:
scan = table_v2.scan()
projection_schema = scan.select("y").projection()
assert projection_schema == Schema(
NestedField(field_id=2, name="y", field_type=LongType(), required=True, doc="comment"),
identifier_field_ids=[2],
)
assert projection_schema.schema_id == 1
def test_table_scan_projection_single_column_case_sensitive(table_v2: Table) -> None:
scan = table_v2.scan()
projection_schema = scan.with_case_sensitive(False).select("Y").projection()
assert projection_schema == Schema(
NestedField(field_id=2, name="y", field_type=LongType(), required=True, doc="comment"),
identifier_field_ids=[2],
)
assert projection_schema.schema_id == 1
def test_table_scan_projection_unknown_column(table_v2: Table) -> None:
scan = table_v2.scan()
with pytest.raises(ValueError) as exc_info:
_ = scan.select("a").projection()
assert "Could not find column: 'a'" in str(exc_info.value)
def test_data_scan_plan_files_no_current_snapshot(example_table_metadata_no_snapshot_v1: dict[str, Any]) -> None:
table = Table(
identifier=("default", "test_no_snapshot"),
metadata=TableMetadataV1(**example_table_metadata_no_snapshot_v1),
metadata_location="s3://bucket/test/metadata.json",
io=load_file_io(),
catalog=NoopCatalog("noop"),
)
assert table.current_snapshot() is None
scan = table.scan()
assert list(scan.plan_files()) == []
assert scan.count() == 0
assert len(scan.to_arrow()) == 0
def test_incremental_append_scan_default(table_v2: Table) -> None:
scan = table_v2.incremental_append_scan()
assert scan.row_filter == AlwaysTrue()
assert scan.selected_fields == ("*",)
assert scan.case_sensitive is True
assert scan.from_snapshot_id is None
assert scan.from_snapshot_inclusive is False
assert scan.to_snapshot_id is None
def test_incremental_append_scan_no_current_snapshot(example_table_metadata_no_snapshot_v1: dict[str, Any]) -> None:
table = Table(
identifier=("default", "test_no_snapshot"),
metadata=TableMetadataV1(**example_table_metadata_no_snapshot_v1),
metadata_location="s3://bucket/test/metadata.json",
io=load_file_io(),
catalog=NoopCatalog("noop"),
)
assert table.current_snapshot() is None
# With no range set and no current snapshot, there is nothing to scan.
assert list(table.incremental_append_scan().plan_files()) == []
# The guard doesn't mask a missing end: an explicit start with no resolvable end still errors.
with pytest.raises(ValueError, match="End snapshot is not set and table has no current snapshot"):
list(table.incremental_append_scan(from_snapshot_id_exclusive=42).plan_files())
def test_incremental_append_scan_factory_maps_to_exclusive_from(table_v2: Table) -> None:
older_snapshot_id, newer_snapshot_id = 3051729675574597004, 3055729675574597004
scan = table_v2.incremental_append_scan(
from_snapshot_id_exclusive=older_snapshot_id,
to_snapshot_id_inclusive=newer_snapshot_id,
)
assert scan.from_snapshot_id == older_snapshot_id
assert scan.from_snapshot_inclusive is False
assert scan.to_snapshot_id == newer_snapshot_id
def test_incremental_append_scan_builders(table_v2: Table) -> None:
older_snapshot_id, newer_snapshot_id = 3051729675574597004, 3055729675574597004
# Builders return refined copies; the snapshot-independent helpers still chain and must
# round-trip the range through `update()`.
scan = (
table_v2.incremental_append_scan()
.from_snapshot_id_exclusive(older_snapshot_id)
.to_snapshot_id_inclusive(newer_snapshot_id)
.filter(EqualTo("x", 1))
.select("x", "y")
.with_case_sensitive(False)
)
assert scan.from_snapshot_id == older_snapshot_id
assert scan.from_snapshot_inclusive is False
assert scan.to_snapshot_id == newer_snapshot_id
assert scan.row_filter == EqualTo("x", 1)
assert set(scan.selected_fields) == {"x", "y"}
assert scan.case_sensitive is False
# `from_snapshot_id_inclusive` flips the flag on the same slot.
inclusive_scan = table_v2.incremental_append_scan().from_snapshot_id_inclusive(older_snapshot_id)
assert inclusive_scan.from_snapshot_id == older_snapshot_id
assert inclusive_scan.from_snapshot_inclusive is True
def test_incremental_append_scan_projection_uses_current_schema(table_v2: Table) -> None:
scan = table_v2.incremental_append_scan(selected_fields=("x", "y"))
projection_schema = scan.projection()
assert projection_schema.schema_id == 1
assert {f.name for f in projection_schema.fields} == {"x", "y"}
def test_incremental_append_scan_unset_from_resolves_to_oldest_ancestor(table_v2: Table) -> None:
older_snapshot_id, current_snapshot_id = 3051729675574597004, 3055729675574597004
# `older` is the oldest ancestor of `current`; an unset start resolves to a None exclusive
# boundary (i.e. the whole lineage of the end snapshot is in range).
scan = table_v2.incremental_append_scan(to_snapshot_id_inclusive=current_snapshot_id)
assert scan._validate_and_resolve_snapshots() == (None, current_snapshot_id)
assert table_v2.snapshot_by_id(older_snapshot_id).parent_snapshot_id is None # type: ignore[union-attr]
def test_incremental_append_scan_inclusive_from_resolves_to_parent(table_v2: Table) -> None:
older_snapshot_id, current_snapshot_id = 3051729675574597004, 3055729675574597004
# An inclusive start resolves to its parent as the exclusive boundary; `older` is the root,
# so its parent (None) becomes the boundary and the start snapshot itself stays in range.
scan = table_v2.incremental_append_scan().from_snapshot_id_inclusive(older_snapshot_id)
assert scan._validate_and_resolve_snapshots() == (None, current_snapshot_id)
# An inclusive start that isn't an ancestor of the end snapshot is rejected.
bad_scan = table_v2.incremental_append_scan(to_snapshot_id_inclusive=older_snapshot_id).from_snapshot_id_inclusive(
current_snapshot_id
)
with pytest.raises(ValueError, match="Starting snapshot .inclusive. .* is not an ancestor"):
bad_scan._validate_and_resolve_snapshots()
# Unlike an exclusive start (which admits an expired snapshot), an inclusive start must be
# present in the metadata, since its parent is needed to form the exclusive boundary.
missing_scan = table_v2.incremental_append_scan().from_snapshot_id_inclusive(42)
with pytest.raises(ValueError, match="Start snapshot .inclusive. not found in table metadata: 42"):
missing_scan._validate_and_resolve_snapshots()
def test_incremental_append_scan_invalid_to_snapshot(table_v2: Table) -> None:
scan = table_v2.incremental_append_scan(
from_snapshot_id_exclusive=3051729675574597004,
to_snapshot_id_inclusive=42,
)
with pytest.raises(ValueError, match="End snapshot not found in table metadata: 42"):
list(scan.plan_files())
@pytest.mark.parametrize(
"from_snapshot_id, to_snapshot_id, expected",
[
# Fabricated 'from' snapshot ID, unrelated to the table.
(42, 3055729675574597004, "Starting snapshot .exclusive. 42 is not a parent ancestor"),
# 'from' and 'to' reversed (newer used as exclusive start).
(3055729675574597004, 3051729675574597004, "is not a parent ancestor of end snapshot"),
# Equal from/to: a snapshot is never its own parent ancestor.
(3055729675574597004, 3055729675574597004, "is not a parent ancestor of end snapshot"),
],
)
def test_incremental_append_scan_rejects_non_parent_ancestor_from(
table_v2: Table, from_snapshot_id: int, to_snapshot_id: int, expected: str
) -> None:
scan = table_v2.incremental_append_scan(
from_snapshot_id_exclusive=from_snapshot_id,
to_snapshot_id_inclusive=to_snapshot_id,
)
with pytest.raises(ValueError, match=expected):
list(scan.plan_files())
def test_incremental_append_scan_to_snapshot_defaults_to_current(table_v2: Table) -> None:
older_snapshot_id, current_snapshot_id = 3051729675574597004, 3055729675574597004
assert table_v2.current_snapshot().snapshot_id == current_snapshot_id # type: ignore[union-attr]
# Omitting `to_snapshot_id_inclusive` resolves to the table's current snapshot.
scan = table_v2.incremental_append_scan(from_snapshot_id_exclusive=older_snapshot_id)
assert scan._validate_and_resolve_snapshots() == (older_snapshot_id, current_snapshot_id)
def test_incremental_append_scan_admits_expired_from_snapshot(example_table_metadata_v2: dict[str, Any]) -> None:
# The exclusive `from` snapshot does not need to be present in the table metadata, as long as
# some ancestor of `to` references it as a parent. Here we drop the oldest snapshot from the
# metadata (simulating snapshot expiration) and verify that the remaining-but-now-expired
# snapshot ID is still accepted as the exclusive `from`.
expired_snapshot_id = 3051729675574597004
remaining_snapshot_id = 3055729675574597004
metadata_dict = {**example_table_metadata_v2}
metadata_dict["snapshots"] = [
snapshot for snapshot in metadata_dict["snapshots"] if snapshot["snapshot-id"] != expired_snapshot_id
]
# The remaining snapshot still references the expired one as its parent.
table = Table(
identifier=("dummy", "test"),
metadata=TableMetadataV2(**metadata_dict),
metadata_location="s3://bucket/test/metadata.json",
io=load_file_io(),
catalog=NoopCatalog("noop"),
)
assert table.snapshot_by_id(expired_snapshot_id) is None
assert table.snapshot_by_id(remaining_snapshot_id) is not None
scan = table.incremental_append_scan(
from_snapshot_id_exclusive=expired_snapshot_id,
to_snapshot_id_inclusive=remaining_snapshot_id,
)
# The private validation method must accept this combination; we don't call `plan_files()` here
# because that would require real manifest files.
assert scan._validate_and_resolve_snapshots() == (expired_snapshot_id, remaining_snapshot_id)
def test_incremental_append_scan_update_preserves_type(table_v2: Table) -> None:
# `update()` lives on BaseScan and uses `type(self)(...)` to construct the copy;
# verify the concrete type is preserved through both `update` and the fluent helpers.
base_scan = table_v2.incremental_append_scan(from_snapshot_id_exclusive=123)
base_type = type(base_scan)
assert base_type.__name__ == "IncrementalAppendScan"
assert type(base_scan.update(from_snapshot_id=456)) is base_type
assert type(base_scan.from_snapshot_id_exclusive(456)) is base_type
assert type(base_scan.from_snapshot_id_inclusive(456)) is base_type
assert type(base_scan.to_snapshot_id_inclusive(789)) is base_type
assert type(base_scan.filter(EqualTo("x", 1))) is base_type
assert type(base_scan.select("x")) is base_type
assert type(base_scan.with_case_sensitive(False)) is base_type
# The snapshot range round-trips through the public attributes (update() reconstructs by keyword).
assert base_scan.from_snapshot_id_inclusive(456).from_snapshot_inclusive is True
assert base_scan.to_snapshot_id_inclusive(789).to_snapshot_id == 789
assert base_scan.update(to_snapshot_id=789).update(to_snapshot_id=None).to_snapshot_id is None
def test_incremental_append_scan_staged_table_raises(table_v2: Table) -> None:
# Mirrors StagedTable.scan: a staged table has no committed metadata to scan against.
staged_table = StagedTable(
identifier=table_v2._identifier,
metadata=table_v2.metadata,
metadata_location=table_v2.metadata_location,
io=table_v2.io,
catalog=table_v2.catalog,
)
with pytest.raises(ValueError, match="Cannot scan a staged table"):
staged_table.incremental_append_scan()
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_version_hint_same_as_table(
table_v2: Table,
table_location_with_version_hint_full: str,
table_location_with_version_hint_numeric: str,
table_location_with_version_hint_non_numeric: str,
) -> None:
for table_location in [
table_location_with_version_hint_full,
table_location_with_version_hint_numeric,
table_location_with_version_hint_non_numeric,
]:
static_table = StaticTable.from_metadata(table_location)
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_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:
update = UpdateSchema(transaction=table_v2.transaction())
update.add_column(path="b", field_type=IntegerType())
apply_schema: Schema = update._apply() # pylint: disable=W0212
assert len(apply_schema.fields) == 4
assert apply_schema == Schema(
NestedField(field_id=1, name="x", field_type=LongType(), required=True),
NestedField(field_id=2, name="y", field_type=LongType(), required=True, doc="comment"),
NestedField(field_id=3, name="z", field_type=LongType(), required=True),
NestedField(field_id=4, name="b", field_type=IntegerType(), required=False),
identifier_field_ids=[1, 2],
)
assert apply_schema.schema_id == 2
assert apply_schema.highest_field_id == 4
def test_update_column(table_v1: Table, table_v2: Table) -> None:
"""
Table should be able to update existing property `doc`
Table should also be able to update property `required`, if the field is not an identifier field.
"""
COMMENT2 = "comment2"
for table in [table_v1, table_v2]:
original_schema = table.schema()
# update existing doc to a new doc
assert original_schema.find_field("y").doc == "comment"
new_schema = table.transaction().update_schema().update_column("y", doc=COMMENT2)._apply()
assert new_schema.find_field("y").doc == COMMENT2, "failed to update existing field doc"
# update existing doc to an empty string
assert new_schema.find_field("y").doc == COMMENT2
new_schema2 = table.transaction().update_schema().update_column("y", doc="")._apply()
assert new_schema2.find_field("y").doc == "", "failed to remove existing field doc"
# update required to False
assert original_schema.find_field("z").required is True
new_schema3 = table.transaction().update_schema().update_column("z", required=False)._apply()
assert new_schema3.find_field("z").required is False, "failed to update existing field required"
# assert the above two updates also works with union_by_name
assert table.update_schema().union_by_name(new_schema)._apply() == new_schema, (
"failed to update existing field doc with union_by_name"
)
assert table.update_schema().union_by_name(new_schema2)._apply() == new_schema2, (
"failed to remove existing field doc with union_by_name"
)
assert table.update_schema().union_by_name(new_schema3)._apply() == new_schema3, (
"failed to update existing field required with union_by_name"
)
def test_add_primitive_type_column(table_v2: Table) -> None:
primitive_type: dict[str, PrimitiveType] = {
"boolean": BooleanType(),
"int": IntegerType(),
"long": LongType(),
"float": FloatType(),
"double": DoubleType(),
"date": DateType(),
"time": TimeType(),
"timestamp": TimestampType(),
"timestamptz": TimestamptzType(),
"string": StringType(),
"uuid": UUIDType(),
"binary": BinaryType(),
}
for name, type_ in primitive_type.items():
field_name = f"new_column_{name}"
update = UpdateSchema(transaction=table_v2.transaction())
update.add_column(path=field_name, field_type=type_, doc=f"new_column_{name}")
new_schema = update._apply() # pylint: disable=W0212
field: NestedField = new_schema.find_field(field_name)
assert field.field_type == type_
assert field.doc == f"new_column_{name}"
def test_add_nested_type_column(table_v2: Table) -> None:
# add struct type column
field_name = "new_column_struct"
update = UpdateSchema(transaction=table_v2.transaction())
struct_ = StructType(
NestedField(1, "lat", DoubleType()),
NestedField(2, "long", DoubleType()),
)
update.add_column(path=field_name, field_type=struct_)
schema_ = update._apply() # pylint: disable=W0212
field: NestedField = schema_.find_field(field_name)
assert field.field_type == StructType(
NestedField(5, "lat", DoubleType()),
NestedField(6, "long", DoubleType()),
)
assert schema_.highest_field_id == 6
def test_add_nested_map_type_column(table_v2: Table) -> None:
# add map type column
field_name = "new_column_map"
update = UpdateSchema(transaction=table_v2.transaction())
map_ = MapType(1, StringType(), 2, IntegerType(), False)
update.add_column(path=field_name, field_type=map_)
new_schema = update._apply() # pylint: disable=W0212
field: NestedField = new_schema.find_field(field_name)
assert field.field_type == MapType(5, StringType(), 6, IntegerType(), False)
assert new_schema.highest_field_id == 6
def test_add_nested_list_type_column(table_v2: Table) -> None:
# add list type column
field_name = "new_column_list"
update = UpdateSchema(transaction=table_v2.transaction())
list_ = ListType(
element_id=101,
element_type=StructType(
NestedField(102, "lat", DoubleType()),
NestedField(103, "long", DoubleType()),
),
element_required=False,
)
update.add_column(path=field_name, field_type=list_)
new_schema = update._apply() # pylint: disable=W0212
field: NestedField = new_schema.find_field(field_name)
assert field.field_type == ListType(
element_id=5,
element_type=StructType(
NestedField(6, "lat", DoubleType()),
NestedField(7, "long", DoubleType()),
),
element_required=False,
)
assert new_schema.highest_field_id == 7
def test_update_list_element_required(table_v2: Table) -> None:
"""Test that update_column can change list element's required property."""
# Add a list column with optional elements
update = UpdateSchema(transaction=table_v2.transaction())
update.add_column(path="tags", field_type=ListType(element_id=1, element_type=StringType(), element_required=False))
schema_with_list = update._apply()
# Verify initial state
field = schema_with_list.find_field("tags")
assert isinstance(field.field_type, ListType)
assert field.field_type.element_required is False
# Update element to required
update2 = UpdateSchema(transaction=table_v2.transaction(), allow_incompatible_changes=True, schema=schema_with_list)
new_schema = update2.update_column(("tags", "element"), required=True)._apply()
# Verify the update
field = new_schema.find_field("tags")
assert isinstance(field.field_type, ListType)
assert field.field_type.element_required is True
def test_update_map_value_required(table_v2: Table) -> None:
"""Test that update_column can change map value's required property."""
# Add a map column with optional values
update = UpdateSchema(transaction=table_v2.transaction())
update.add_column(
path="metadata",
field_type=MapType(key_id=1, key_type=StringType(), value_id=2, value_type=IntegerType(), value_required=False),
)
schema_with_map = update._apply()
# Verify initial state
field = schema_with_map.find_field("metadata")
assert isinstance(field.field_type, MapType)
assert field.field_type.value_required is False
# Update value to required
update2 = UpdateSchema(transaction=table_v2.transaction(), allow_incompatible_changes=True, schema=schema_with_map)
new_schema = update2.update_column(("metadata", "value"), required=True)._apply()
# Verify the update
field = new_schema.find_field("metadata")
assert isinstance(field.field_type, MapType)
assert field.field_type.value_required is True
def test_update_list_element_required_to_optional(table_v2: Table) -> None:
"""Test that update_column can change list element from required to optional (safe direction)."""
# Add a list column with required elements
update = UpdateSchema(transaction=table_v2.transaction())
update.add_column(path="tags", field_type=ListType(element_id=1, element_type=StringType(), element_required=True))
schema_with_list = update._apply()
# Verify initial state
field = schema_with_list.find_field("tags")
assert isinstance(field.field_type, ListType)
assert field.field_type.element_required is True
# Update element to optional - should work without allow_incompatible_changes
update2 = UpdateSchema(transaction=table_v2.transaction(), schema=schema_with_list)
new_schema = update2.update_column(("tags", "element"), required=False)._apply()
# Verify the update
field = new_schema.find_field("tags")
assert isinstance(field.field_type, ListType)
assert field.field_type.element_required is False
def test_update_list_element_required_fails_without_allow_incompatible(table_v2: Table) -> None:
"""Test that optional -> required fails without allow_incompatible_changes."""
# Add a list column with optional elements
update = UpdateSchema(transaction=table_v2.transaction())
update.add_column(path="tags", field_type=ListType(element_id=1, element_type=StringType(), element_required=False))
schema_with_list = update._apply()
# Try to update element to required without allow_incompatible_changes - should fail
update2 = UpdateSchema(transaction=table_v2.transaction(), schema=schema_with_list)
with pytest.raises(ValueError, match="Cannot change column nullability"):
update2.update_column(("tags", "element"), required=True)._apply()
def test_update_map_value_required_fails_without_allow_incompatible(table_v2: Table) -> None:
"""Test that optional -> required for map value fails without allow_incompatible_changes."""
# Add a map column with optional values
update = UpdateSchema(transaction=table_v2.transaction())
update.add_column(
path="metadata",
field_type=MapType(key_id=1, key_type=StringType(), value_id=2, value_type=IntegerType(), value_required=False),
)
schema_with_map = update._apply()
# Try to update value to required without allow_incompatible_changes - should fail
update2 = UpdateSchema(transaction=table_v2.transaction(), schema=schema_with_map)
with pytest.raises(ValueError, match="Cannot change column nullability"):
update2.update_column(("metadata", "value"), required=True)._apply()
def test_update_map_key_fails(table_v2: Table) -> None:
"""Test that updating map keys is not allowed."""
# Add a map column
update = UpdateSchema(transaction=table_v2.transaction())
update.add_column(
path="metadata",
field_type=MapType(key_id=1, key_type=StringType(), value_id=2, value_type=IntegerType(), value_required=False),
)
schema_with_map = update._apply()
# Try to update the key - should fail even with allow_incompatible_changes
update2 = UpdateSchema(transaction=table_v2.transaction(), allow_incompatible_changes=True, schema=schema_with_map)
with pytest.raises(ValueError, match="Cannot update map keys"):
update2.update_column(("metadata", "key"), required=False)._apply()
def test_update_map_value_required_to_optional(table_v2: Table) -> None:
"""Test that update_column can change map value from required to optional (safe direction)."""
# Add a map column with required values
update = UpdateSchema(transaction=table_v2.transaction())
update.add_column(
path="metadata",
field_type=MapType(key_id=1, key_type=StringType(), value_id=2, value_type=IntegerType(), value_required=True),
)
schema_with_map = update._apply()
# Verify initial state
field = schema_with_map.find_field("metadata")
assert isinstance(field.field_type, MapType)
assert field.field_type.value_required is True
# Update value to optional - should work without allow_incompatible_changes
update2 = UpdateSchema(transaction=table_v2.transaction(), schema=schema_with_map)
new_schema = update2.update_column(("metadata", "value"), required=False)._apply()
# Verify the update
field = new_schema.find_field("metadata")
assert isinstance(field.field_type, MapType)
assert field.field_type.value_required is False
def test_update_list_and_map_in_single_schema_change(table_v2: Table) -> None:
"""Test updating both list element and map value required properties in a single schema change."""
# Add both a list and a map column with optional elements/values
update = UpdateSchema(transaction=table_v2.transaction())
update.add_column(path="tags", field_type=ListType(element_id=1, element_type=StringType(), element_required=False))
update.add_column(
path="metadata",
field_type=MapType(key_id=1, key_type=StringType(), value_id=2, value_type=IntegerType(), value_required=False),
)
schema_with_both = update._apply()
# Verify initial state
list_field = schema_with_both.find_field("tags")
assert isinstance(list_field.field_type, ListType)
assert list_field.field_type.element_required is False
map_field = schema_with_both.find_field("metadata")
assert isinstance(map_field.field_type, MapType)
assert map_field.field_type.value_required is False
# Update both in a single schema change
update2 = UpdateSchema(transaction=table_v2.transaction(), allow_incompatible_changes=True, schema=schema_with_both)
update2.update_column(("tags", "element"), required=True)
update2.update_column(("metadata", "value"), required=True)
new_schema = update2._apply()
# Verify both updates
list_field = new_schema.find_field("tags")
assert isinstance(list_field.field_type, ListType)
assert list_field.field_type.element_required is True
map_field = new_schema.find_field("metadata")
assert isinstance(map_field.field_type, MapType)
assert map_field.field_type.value_required is True
def test_update_nested_list_in_struct_required(table_v2: Table) -> None:
"""Test updating nested list element required property inside a struct."""
# Add a struct column containing a list
update = UpdateSchema(transaction=table_v2.transaction())
struct_type = StructType(
NestedField(
field_id=1,
name="coordinates",
field_type=ListType(element_id=2, element_type=DoubleType(), element_required=False),
required=False,
)
)
update.add_column(path="location", field_type=struct_type)
schema_with_nested = update._apply()
# Verify initial state
field = schema_with_nested.find_field("location")
assert isinstance(field.field_type, StructType)
nested_list = field.field_type.fields[0].field_type
assert isinstance(nested_list, ListType)
assert nested_list.element_required is False
# Update nested list element to required
update2 = UpdateSchema(transaction=table_v2.transaction(), allow_incompatible_changes=True, schema=schema_with_nested)
new_schema = update2.update_column(("location", "coordinates", "element"), required=True)._apply()
# Verify the update
field = new_schema.find_field("location")
nested_list = field.field_type.fields[0].field_type
assert isinstance(nested_list, ListType)
assert nested_list.element_required is True
def test_apply_set_properties_update(table_v2: Table) -> None:
base_metadata = table_v2.metadata
new_metadata_no_update = update_table_metadata(base_metadata, (SetPropertiesUpdate(updates={}),))
assert new_metadata_no_update == base_metadata
new_metadata = update_table_metadata(
base_metadata, (SetPropertiesUpdate(updates={"read.split.target.size": "123", "test_a": "test_a", "test_b": "test_b"}),)
)
assert base_metadata.properties == {"read.split.target.size": "134217728"}
assert new_metadata.properties == {"read.split.target.size": "123", "test_a": "test_a", "test_b": "test_b"}
new_metadata_add_only = update_table_metadata(new_metadata, (SetPropertiesUpdate(updates={"test_c": "test_c"}),))
assert new_metadata_add_only.properties == {
"read.split.target.size": "123",
"test_a": "test_a",
"test_b": "test_b",
"test_c": "test_c",
}
assert new_metadata_add_only.last_updated_ms > base_metadata.last_updated_ms
def test_apply_remove_properties_update(table_v2: Table) -> None:
base_metadata = update_table_metadata(
table_v2.metadata,
(SetPropertiesUpdate(updates={"test_a": "test_a", "test_b": "test_b", "test_c": "test_c", "test_d": "test_d"}),),
)
new_metadata_no_removal = update_table_metadata(base_metadata, (RemovePropertiesUpdate(removals=[]),))
assert base_metadata == new_metadata_no_removal
new_metadata = update_table_metadata(base_metadata, (RemovePropertiesUpdate(removals=["test_a", "test_c"]),))
assert base_metadata.properties == {
"read.split.target.size": "134217728",
"test_a": "test_a",
"test_b": "test_b",
"test_c": "test_c",
"test_d": "test_d",
}
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:
transaction = table_v2.transaction()
update = transaction.update_schema()
update.add_column(path="b", field_type=IntegerType())
update.commit()
test_context = _TableMetadataUpdateContext()
new_table_metadata = _apply_table_update(transaction._updates[0], base_metadata=table_v2.metadata, context=test_context) # pylint: disable=W0212
assert len(new_table_metadata.schemas) == 3
assert new_table_metadata.current_schema_id == 1
assert len(test_context._updates) == 1
assert test_context._updates[0] == transaction._updates[0] # pylint: disable=W0212
assert test_context.is_added_schema(2)
new_table_metadata = _apply_table_update(transaction._updates[1], base_metadata=new_table_metadata, context=test_context) # pylint: disable=W0212
assert len(new_table_metadata.schemas) == 3
assert new_table_metadata.current_schema_id == 2
assert len(test_context._updates) == 2
assert test_context._updates[1] == transaction._updates[1] # pylint: disable=W0212
assert test_context.is_added_schema(2)
def test_update_metadata_table_schema(table_v2: Table) -> None:
transaction = table_v2.transaction()
update = transaction.update_schema()
update.add_column(path="b", field_type=IntegerType())
update.commit()
new_metadata = update_table_metadata(table_v2.metadata, transaction._updates) # pylint: disable=W0212
apply_schema: Schema = next(schema for schema in new_metadata.schemas if schema.schema_id == 2)
assert len(apply_schema.fields) == 4
assert apply_schema == Schema(
NestedField(field_id=1, name="x", field_type=LongType(), required=True),
NestedField(field_id=2, name="y", field_type=LongType(), required=True, doc="comment"),
NestedField(field_id=3, name="z", field_type=LongType(), required=True),
NestedField(field_id=4, name="b", field_type=IntegerType(), required=False),
identifier_field_ids=[1, 2],
)
assert apply_schema.schema_id == 2
assert apply_schema.highest_field_id == 4
assert new_metadata.current_schema_id == 2
def test_update_metadata_add_snapshot(table_v2: Table) -> None:
new_snapshot = Snapshot(
snapshot_id=25,
parent_snapshot_id=19,
sequence_number=200,
timestamp_ms=1602638593590,
manifest_list="s3:/a/b/c.avro",
summary=Summary(Operation.APPEND),
schema_id=3,
)
new_metadata = update_table_metadata(table_v2.metadata, (AddSnapshotUpdate(snapshot=new_snapshot),))
assert len(new_metadata.snapshots) == 3
assert new_metadata.snapshots[-1] == new_snapshot
assert new_metadata.last_sequence_number == new_snapshot.sequence_number
assert new_metadata.last_updated_ms == new_snapshot.timestamp_ms
def test_update_metadata_set_ref_snapshot(table_v2: Table) -> None:
update, _ = table_v2.transaction()._set_ref_snapshot(
snapshot_id=3051729675574597004,
ref_name="main",
type="branch",
max_ref_age_ms=123123123,
max_snapshot_age_ms=12312312312,
min_snapshots_to_keep=1,
)
new_metadata = update_table_metadata(table_v2.metadata, update)
assert len(new_metadata.snapshot_log) == 3
assert new_metadata.snapshot_log[2].snapshot_id == 3051729675574597004
assert new_metadata.current_snapshot_id == 3051729675574597004
assert new_metadata.last_updated_ms > table_v2.metadata.last_updated_ms
assert new_metadata.refs["main"] == SnapshotRef(
snapshot_id=3051729675574597004,
snapshot_ref_type="branch",
min_snapshots_to_keep=1,
max_snapshot_age_ms=12312312312,
max_ref_age_ms=123123123,
)
def test_update_metadata_set_snapshot_ref(table_v2: Table) -> None:
update = SetSnapshotRefUpdate(
ref_name="main",
type="branch",
snapshot_id=3051729675574597004,
max_ref_age_ms=123123123,
max_snapshot_age_ms=12312312312,
min_snapshots_to_keep=1,
)
new_metadata = update_table_metadata(table_v2.metadata, (update,))
assert len(new_metadata.snapshot_log) == 3
assert new_metadata.snapshot_log[2].snapshot_id == 3051729675574597004
assert new_metadata.current_snapshot_id == 3051729675574597004
assert new_metadata.last_updated_ms > table_v2.metadata.last_updated_ms
assert new_metadata.refs[update.ref_name] == SnapshotRef(
snapshot_id=3051729675574597004,
snapshot_ref_type="branch",
min_snapshots_to_keep=1,
max_snapshot_age_ms=12312312312,
max_ref_age_ms=123123123,
)
def test_update_remove_snapshots(table_v2: Table) -> None:
REMOVE_SNAPSHOT = 3051729675574597004
KEEP_SNAPSHOT = 3055729675574597004
# assert fixture data to easily understand the test assumptions
assert len(table_v2.metadata.snapshots) == 2
assert len(table_v2.metadata.snapshot_log) == 2
assert len(table_v2.metadata.refs) == 2
update = RemoveSnapshotsUpdate(snapshot_ids=[REMOVE_SNAPSHOT])
new_metadata = update_table_metadata(table_v2.metadata, (update,))
assert len(new_metadata.snapshots) == 1
assert new_metadata.snapshots[0].snapshot_id == KEEP_SNAPSHOT
assert new_metadata.snapshots[0].parent_snapshot_id is None
assert new_metadata.current_snapshot_id == KEEP_SNAPSHOT
assert new_metadata.last_updated_ms > table_v2.metadata.last_updated_ms
assert len(new_metadata.snapshot_log) == 1
assert new_metadata.snapshot_log[0].snapshot_id == KEEP_SNAPSHOT
assert len(new_metadata.refs) == 1
assert new_metadata.refs["main"].snapshot_id == KEEP_SNAPSHOT
def test_update_remove_snapshots_doesnt_exist(table_v2: Table) -> None:
update = RemoveSnapshotsUpdate(
snapshot_ids=[123],
)
with pytest.raises(ValueError, match="Snapshot with snapshot id 123 does not exist"):
update_table_metadata(table_v2.metadata, (update,))
def test_update_remove_snapshots_remove_current_snapshot_id(table_v2: Table) -> None:
update = RemoveSnapshotsUpdate(snapshot_ids=[3055729675574597004])
new_metadata = update_table_metadata(table_v2.metadata, (update,))
assert len(new_metadata.refs) == 1
assert new_metadata.refs["test"].snapshot_id == 3051729675574597004
assert new_metadata.current_snapshot_id is None
def test_update_remove_snapshot_ref(table_v2: Table) -> None:
# assert fixture data to easily understand the test assumptions
assert len(table_v2.metadata.refs) == 2
update = RemoveSnapshotRefUpdate(ref_name="test")
new_metadata = update_table_metadata(table_v2.metadata, (update,))
assert len(new_metadata.refs) == 1
assert new_metadata.refs["main"].snapshot_id == 3055729675574597004
def test_update_metadata_add_update_sort_order(table_v2: Table) -> None:
new_sort_order = SortOrder(order_id=table_v2.sort_order().order_id + 1)
new_metadata = update_table_metadata(
table_v2.metadata,
(AddSortOrderUpdate(sort_order=new_sort_order), SetDefaultSortOrderUpdate(sort_order_id=-1)),
)
assert len(new_metadata.sort_orders) == 2
assert new_metadata.sort_orders[-1] == new_sort_order
assert new_metadata.default_sort_order_id == new_sort_order.order_id
assert new_metadata.last_updated_ms > table_v2.metadata.last_updated_ms
def test_update_metadata_update_sort_order_invalid(table_v2: Table) -> None:
with pytest.raises(ValueError, match="Cannot set current sort order to the last added one when no sort order has been added"):
update_table_metadata(table_v2.metadata, (SetDefaultSortOrderUpdate(sort_order_id=-1),))
invalid_order_id = 10
with pytest.raises(ValueError, match=f"Sort order with id {invalid_order_id} does not exist"):
update_table_metadata(table_v2.metadata, (SetDefaultSortOrderUpdate(sort_order_id=invalid_order_id),))
def test_update_metadata_with_multiple_updates(table_v1: Table) -> None:
base_metadata = table_v1.metadata
transaction = table_v1.transaction()
transaction.upgrade_table_version(format_version=2)
schema_update_1 = transaction.update_schema()
schema_update_1.add_column(path="b", field_type=IntegerType())
schema_update_1.commit()
transaction.set_properties(owner="test", test_a="test_a", test_b="test_b", test_c="test_c")
test_updates = transaction._updates # pylint: disable=W0212
new_snapshot = Snapshot(
snapshot_id=25,
parent_snapshot_id=19,
sequence_number=200,
timestamp_ms=1602638573590,
manifest_list="s3:/a/b/c.avro",
summary=Summary(Operation.APPEND),
schema_id=3,
)
test_updates += (
AddSnapshotUpdate(snapshot=new_snapshot),
SetPropertiesUpdate(updates={"test_a": "test_a1"}),
SetSnapshotRefUpdate(
ref_name="main",
type="branch",
snapshot_id=25,
max_ref_age_ms=123123123,
max_snapshot_age_ms=12312312312,
min_snapshots_to_keep=1,
),
RemovePropertiesUpdate(removals=["test_c", "test_b"]),
)
new_metadata = update_table_metadata(base_metadata, test_updates)
# rebuild the metadata to trigger validation
new_metadata = TableMetadataUtil.parse_obj(copy(new_metadata.model_dump()))
# UpgradeFormatVersionUpdate
assert new_metadata.format_version == 2
assert isinstance(new_metadata, TableMetadataV2)
# UpdateSchema
assert len(new_metadata.schemas) == 2
assert new_metadata.current_schema_id == 1
assert new_metadata.schema_by_id(new_metadata.current_schema_id).highest_field_id == 4 # type: ignore
# AddSchemaUpdate
assert len(new_metadata.snapshots) == 2
assert new_metadata.snapshots[-1] == new_snapshot
assert new_metadata.last_sequence_number == new_snapshot.sequence_number
assert new_metadata.last_updated_ms == new_snapshot.timestamp_ms
# SetSnapshotRefUpdate
assert len(new_metadata.snapshot_log) == 1
assert new_metadata.snapshot_log[0].snapshot_id == 25
assert new_metadata.current_snapshot_id == 25
assert new_metadata.last_updated_ms == 1602638573590
assert new_metadata.refs["main"] == SnapshotRef(
snapshot_id=25,
snapshot_ref_type="branch",
min_snapshots_to_keep=1,
max_snapshot_age_ms=12312312312,
max_ref_age_ms=123123123,
)
# Set/RemovePropertiesUpdate
assert new_metadata.properties == {"owner": "test", "test_a": "test_a1"}
def test_update_metadata_schema_immutability(
table_v2_with_fixed_and_decimal_types: TableMetadataV2,
) -> None:
update = SetSnapshotRefUpdate(
ref_name="main",
type="branch",
snapshot_id=3051729675574597004,
max_ref_age_ms=123123123,
max_snapshot_age_ms=12312312312,
min_snapshots_to_keep=1,
)
new_metadata = update_table_metadata(
table_v2_with_fixed_and_decimal_types.metadata,
(update,),
)
assert new_metadata.schemas[0].fields == (
NestedField(field_id=1, name="x", field_type=LongType(), required=True),
NestedField(field_id=4, name="a", field_type=DecimalType(precision=16, scale=2), required=True),
NestedField(field_id=5, name="b", field_type=DecimalType(precision=16, scale=8), required=True),
NestedField(field_id=6, name="c", field_type=FixedType(length=16), required=True),
NestedField(field_id=7, name="d", field_type=FixedType(length=18), required=True),
)
def test_metadata_isolation_from_illegal_updates(table_v1: Table) -> None:
base_metadata = table_v1.metadata
base_metadata_backup = base_metadata.model_copy(deep=True)
# Apply legal updates on the table metadata
transaction = table_v1.transaction()
schema_update_1 = transaction.update_schema()
schema_update_1.add_column(path="b", field_type=IntegerType())
schema_update_1.commit()
test_updates = transaction._updates # pylint: disable=W0212
new_snapshot = Snapshot(
snapshot_id=25,
parent_snapshot_id=19,
sequence_number=200,
timestamp_ms=1602638573590,
manifest_list="s3:/a/b/c.avro",
summary=Summary(Operation.APPEND),
schema_id=3,
)
test_updates += (
AddSnapshotUpdate(snapshot=new_snapshot),
SetSnapshotRefUpdate(
ref_name="main",
type="branch",
snapshot_id=25,
max_ref_age_ms=123123123,
max_snapshot_age_ms=12312312312,
min_snapshots_to_keep=1,
),
)
new_metadata = update_table_metadata(base_metadata, test_updates)
# Check that the original metadata is not modified
assert base_metadata == base_metadata_backup
# Perform illegal update on the new metadata:
# TableMetadata should be immutable, but the pydantic's frozen config cannot prevent
# operations such as list append.
new_metadata.partition_specs.append(PartitionSpec(spec_id=0))
assert len(new_metadata.partition_specs) == 2
# The original metadata should not be affected by the illegal update on the new metadata
assert len(base_metadata.partition_specs) == 1
def test_generate_snapshot_id(table_v2: Table) -> None:
assert isinstance(_generate_snapshot_id(), int)
assert isinstance(table_v2.metadata.new_snapshot_id(), int)
def test_assert_create(table_v2: Table) -> None:
AssertCreate().validate(None)
with pytest.raises(CommitFailedException, match="Table already exists"):
AssertCreate().validate(table_v2.metadata)
def test_assert_table_uuid(table_v2: Table) -> None:
base_metadata = table_v2.metadata
AssertTableUUID(uuid=base_metadata.table_uuid).validate(base_metadata)
with pytest.raises(CommitFailedException, match="Requirement failed: current table metadata is missing"):
AssertTableUUID(uuid=uuid.UUID("9c12d441-03fe-4693-9a96-a0705ddf69c2")).validate(None)
with pytest.raises(
CommitFailedException,
match="Table UUID does not match: 9c12d441-03fe-4693-9a96-a0705ddf69c2 != 9c12d441-03fe-4693-9a96-a0705ddf69c1",
):
AssertTableUUID(uuid=uuid.UUID("9c12d441-03fe-4693-9a96-a0705ddf69c2")).validate(base_metadata)
def test_assert_ref_snapshot_id(table_v2: Table) -> None:
base_metadata = table_v2.metadata
AssertRefSnapshotId(ref=MAIN_BRANCH, snapshot_id=base_metadata.current_snapshot_id).validate(base_metadata)
with pytest.raises(CommitFailedException, match="Requirement failed: current table metadata is missing"):
AssertRefSnapshotId(ref=MAIN_BRANCH, snapshot_id=1).validate(None)
with pytest.raises(
CommitFailedException,
match=f"Requirement failed: branch {MAIN_BRANCH} was created concurrently",
):
AssertRefSnapshotId(ref=MAIN_BRANCH, snapshot_id=None).validate(base_metadata)
with pytest.raises(
CommitFailedException,
match=f"Requirement failed: branch {MAIN_BRANCH} has changed: expected id 1, found 3055729675574597004",
):
AssertRefSnapshotId(ref=MAIN_BRANCH, snapshot_id=1).validate(base_metadata)
non_existing_ref = "not_exist_branch_or_tag"
assert table_v2.refs().get("not_exist_branch_or_tag") is None
with pytest.raises(
CommitFailedException,
match=f"Requirement failed: branch or tag {non_existing_ref} is missing, expected 1",
):
AssertRefSnapshotId(ref=non_existing_ref, snapshot_id=1).validate(base_metadata)
# existing Tag in metadata: test
ref_tag = table_v2.refs().get("test")
assert ref_tag is not None
assert ref_tag.snapshot_ref_type == SnapshotRefType.TAG, "TAG test should be present in table to be tested"
with pytest.raises(
CommitFailedException,
match="Requirement failed: tag test has changed: expected id 3055729675574597004, found 3051729675574597004",
):
AssertRefSnapshotId(ref="test", snapshot_id=3055729675574597004).validate(base_metadata)
expected_json = '{"type":"assert-ref-snapshot-id","ref":"main","snapshot-id":null}'
assert AssertRefSnapshotId(ref="main").model_dump_json() == expected_json
def test_assert_last_assigned_field_id(table_v2: Table) -> None:
base_metadata = table_v2.metadata
AssertLastAssignedFieldId(last_assigned_field_id=base_metadata.last_column_id).validate(base_metadata)
with pytest.raises(CommitFailedException, match="Requirement failed: current table metadata is missing"):
AssertLastAssignedFieldId(last_assigned_field_id=1).validate(None)
with pytest.raises(
CommitFailedException,
match="Requirement failed: last assigned field id has changed: expected 1, found 3",
):
AssertLastAssignedFieldId(last_assigned_field_id=1).validate(base_metadata)
def test_assert_current_schema_id(table_v2: Table) -> None:
base_metadata = table_v2.metadata
AssertCurrentSchemaId(current_schema_id=base_metadata.current_schema_id).validate(base_metadata)
with pytest.raises(CommitFailedException, match="Requirement failed: current table metadata is missing"):
AssertCurrentSchemaId(current_schema_id=1).validate(None)
with pytest.raises(
CommitFailedException,
match="Requirement failed: current schema id has changed: expected 2, found 1",
):
AssertCurrentSchemaId(current_schema_id=2).validate(base_metadata)
def test_last_assigned_partition_id(table_v2: Table) -> None:
base_metadata = table_v2.metadata
AssertLastAssignedPartitionId(last_assigned_partition_id=base_metadata.last_partition_id).validate(base_metadata)
with pytest.raises(CommitFailedException, match="Requirement failed: current table metadata is missing"):
AssertLastAssignedPartitionId(last_assigned_partition_id=1).validate(None)
with pytest.raises(
CommitFailedException,
match="Requirement failed: last assigned partition id has changed: expected 1, found 1000",
):
AssertLastAssignedPartitionId(last_assigned_partition_id=1).validate(base_metadata)
def test_assert_default_spec_id(table_v2: Table) -> None:
base_metadata = table_v2.metadata
AssertDefaultSpecId(default_spec_id=base_metadata.default_spec_id).validate(base_metadata)
with pytest.raises(CommitFailedException, match="Requirement failed: current table metadata is missing"):
AssertDefaultSpecId(default_spec_id=1).validate(None)
with pytest.raises(
CommitFailedException,
match="Requirement failed: default spec id has changed: expected 1, found 0",
):
AssertDefaultSpecId(default_spec_id=1).validate(base_metadata)
def test_assert_default_sort_order_id(table_v2: Table) -> None:
base_metadata = table_v2.metadata
AssertDefaultSortOrderId(default_sort_order_id=base_metadata.default_sort_order_id).validate(base_metadata)
with pytest.raises(CommitFailedException, match="Requirement failed: current table metadata is missing"):
AssertDefaultSortOrderId(default_sort_order_id=1).validate(None)
with pytest.raises(
CommitFailedException,
match="Requirement failed: default sort order id has changed: expected 1, found 3",
):
AssertDefaultSortOrderId(default_sort_order_id=1).validate(base_metadata)
def test_correct_schema() -> None:
table_metadata = TableMetadataV2(
**{
"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"},
{"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": 0,
"sort-orders": [],
"current-snapshot-id": 123,
"snapshots": [
{
"snapshot-id": 234,
"timestamp-ms": 1515100955770,
"sequence-number": 0,
"summary": {"operation": "append"},
"manifest-list": "s3://a/b/1.avro",
"schema-id": 10,
},
{
"snapshot-id": 123,
"timestamp-ms": 1515100955770,
"sequence-number": 0,
"summary": {"operation": "append"},
"manifest-list": "s3://a/b/1.avro",
"schema-id": 0,
},
],
}
)
t = Table(
identifier=("default", "t1"),
metadata=table_metadata,
metadata_location="s3://../..",
io=load_file_io(),
catalog=NoopCatalog("NoopCatalog"),
)
# Should use the current schema, instead the one from the snapshot
projection_schema = t.scan().projection()
assert projection_schema == Schema(
NestedField(field_id=1, name="x", field_type=LongType(), required=True),
NestedField(field_id=2, name="y", field_type=LongType(), required=True),
NestedField(field_id=3, name="z", field_type=LongType(), required=True),
identifier_field_ids=[1, 2],
)
assert projection_schema.schema_id == 1
# When we explicitly filter on the commit, we want to have the schema that's linked to the snapshot
projection_schema = t.scan(snapshot_id=123).projection()
assert projection_schema == Schema(
NestedField(field_id=1, name="x", field_type=LongType(), required=True),
identifier_field_ids=[],
)
assert projection_schema.schema_id == 0
with pytest.warns(UserWarning, match="Metadata does not contain schema with id: 10"):
t.scan(snapshot_id=234).projection()
# Invalid snapshot
with pytest.raises(ValueError) as exc_info:
_ = t.scan(snapshot_id=-1).projection()
assert "Snapshot not found: -1" in str(exc_info.value)
def test_table_properties(example_table_metadata_v2: dict[str, Any]) -> None:
# metadata properties are all strings
for k, v in example_table_metadata_v2["properties"].items():
assert isinstance(k, str)
assert isinstance(v, str)
metadata = TableMetadataV2(**example_table_metadata_v2)
for k, v in metadata.properties.items():
assert isinstance(k, str)
assert isinstance(v, str)
# property can be set to int, but still serialized as string
property_with_int = {"property_name": 42}
new_example_table_metadata_v2 = {**example_table_metadata_v2, "properties": property_with_int}
assert isinstance(new_example_table_metadata_v2["properties"]["property_name"], int)
new_metadata = TableMetadataV2(**new_example_table_metadata_v2)
assert isinstance(new_metadata.properties["property_name"], str)
def test_table_properties_raise_for_none_value(example_table_metadata_v2: dict[str, Any]) -> None:
property_with_none = {"property_name": None}
example_table_metadata_v2 = {**example_table_metadata_v2, "properties": property_with_none}
with pytest.raises(ValidationError) as exc_info:
TableMetadataV2(**example_table_metadata_v2)
assert "None type is not a supported value in properties: property_name" in str(exc_info.value)
def test_serialize_commit_table_request() -> None:
request = CommitTableRequest(
requirements=(AssertTableUUID(uuid="4bfd18a3-74c6-478e-98b1-71c4c32f4163"),),
identifier=TableIdentifier(namespace=["a"], name="b"),
)
deserialized_request = CommitTableRequest.model_validate_json(request.model_dump_json())
assert request == deserialized_request
def test_update_metadata_log(table_v2: Table) -> None:
new_snapshot = Snapshot(
snapshot_id=25,
parent_snapshot_id=19,
sequence_number=200,
timestamp_ms=1602638593590,
manifest_list="s3:/a/b/c.avro",
summary=Summary(Operation.APPEND),
schema_id=3,
)
new_metadata = update_table_metadata(
table_v2.metadata, (AddSnapshotUpdate(snapshot=new_snapshot),), False, table_v2.metadata_location
)
assert len(new_metadata.metadata_log) == 2
def test_update_metadata_log_overflow(table_v2: Table) -> None:
metadata_log = [
MetadataLogEntry(
timestamp_ms=1602638593590 + i,
metadata_file=f"/path/to/metadata/{i}.json",
)
for i in range(10)
]
table_v2.metadata = table_v2.metadata.model_copy(update={"metadata_log": metadata_log, "last_updated_ms": 1602638593600})
table_v2.metadata_location = "/path/to/metadata/10.json"
assert len(table_v2.metadata.metadata_log) == 10
base_metadata = table_v2.metadata
new_metadata = update_table_metadata(
base_metadata,
(SetPropertiesUpdate(updates={"write.metadata.previous-versions-max": "5"}),),
False,
table_v2.metadata_location,
)
assert len(new_metadata.metadata_log) == 5
assert new_metadata.metadata_log[-1].metadata_file == "/path/to/metadata/10.json"
# check invalid value of write.metadata.previous-versions-max
new_metadata = update_table_metadata(
base_metadata,
(SetPropertiesUpdate(updates={"write.metadata.previous-versions-max": "0"}),),
False,
table_v2.metadata_location,
)
assert len(new_metadata.metadata_log) == 1
def test_remove_partition_spec_update(table_v2: Table) -> None:
base_metadata = table_v2.metadata
new_spec = PartitionSpec(PartitionField(source_id=2, field_id=1001, transform=IdentityTransform(), name="y"), spec_id=1)
metadata_with_new_spec = update_table_metadata(base_metadata, (AddPartitionSpecUpdate(spec=new_spec),))
assert len(metadata_with_new_spec.partition_specs) == 2
update = RemovePartitionSpecsUpdate(spec_ids=[1])
updated_metadata = update_table_metadata(
metadata_with_new_spec,
(update,),
)
assert len(updated_metadata.partition_specs) == 1
def test_remove_partition_spec_update_spec_does_not_exist(table_v2: Table) -> None:
update = RemovePartitionSpecsUpdate(
spec_ids=[123],
)
with pytest.raises(ValueError, match="Partition spec with id 123 does not exist"):
update_table_metadata(table_v2.metadata, (update,))
def test_remove_partition_spec_update_default_spec(table_v2: Table) -> None:
update = RemovePartitionSpecsUpdate(
spec_ids=[0],
)
with pytest.raises(ValueError, match="Cannot remove default partition spec: 0"):
update_table_metadata(table_v2.metadata, (update,))
def test_remove_schemas_update(table_v2: Table) -> None:
base_metadata = table_v2.metadata
assert len(base_metadata.schemas) == 2
update = RemoveSchemasUpdate(schema_ids=[0])
updated_metadata = update_table_metadata(
base_metadata,
(update,),
)
assert len(updated_metadata.schemas) == 1
def test_remove_schemas_update_schema_does_not_exist(table_v2: Table) -> None:
update = RemoveSchemasUpdate(
schema_ids=[123],
)
with pytest.raises(ValueError, match="Schema with schema id 123 does not exist"):
update_table_metadata(table_v2.metadata, (update,))
def test_remove_schemas_update_current_schema(table_v2: Table) -> None:
update = RemoveSchemasUpdate(
schema_ids=[1],
)
with pytest.raises(ValueError, match="Cannot remove current schema with id 1"):
update_table_metadata(table_v2.metadata, (update,))
def test_set_statistics_update(table_v2_with_statistics: Table) -> None:
snapshot_id = table_v2_with_statistics.metadata.current_snapshot_id
blob_metadata = BlobMetadata(
type="apache-datasketches-theta-v1",
snapshot_id=snapshot_id,
sequence_number=2,
fields=[1],
properties={"prop-key": "prop-value"},
)
statistics_file = StatisticsFile(
snapshot_id=snapshot_id,
statistics_path="s3://bucket/warehouse/stats.puffin",
file_size_in_bytes=124,
file_footer_size_in_bytes=27,
blob_metadata=[blob_metadata],
)
update = SetStatisticsUpdate(
snapshot_id=snapshot_id,
statistics=statistics_file,
)
assert model_roundtrips(update)
new_metadata = update_table_metadata(
table_v2_with_statistics.metadata,
(update,),
)
expected = """
{
"snapshot-id": 3055729675574597004,
"statistics-path": "s3://bucket/warehouse/stats.puffin",
"file-size-in-bytes": 124,
"file-footer-size-in-bytes": 27,
"blob-metadata": [
{
"type": "apache-datasketches-theta-v1",
"snapshot-id": 3055729675574597004,
"sequence-number": 2,
"fields": [
1
],
"properties": {
"prop-key": "prop-value"
}
}
]
}"""
assert len(new_metadata.statistics) == 2
updated_statistics = [stat for stat in new_metadata.statistics if stat.snapshot_id == snapshot_id]
assert len(updated_statistics) == 1
assert json.loads(updated_statistics[0].model_dump_json()) == json.loads(expected)
def test_set_statistics_update_handles_deprecated_snapshot_id(table_v2_with_statistics: Table) -> None:
snapshot_id = table_v2_with_statistics.metadata.current_snapshot_id
blob_metadata = BlobMetadata(
type="apache-datasketches-theta-v1",
snapshot_id=snapshot_id,
sequence_number=2,
fields=[1],
properties={"prop-key": "prop-value"},
)
statistics_file = StatisticsFile(
snapshot_id=snapshot_id,
statistics_path="s3://bucket/warehouse/stats.puffin",
file_size_in_bytes=124,
file_footer_size_in_bytes=27,
blob_metadata=[blob_metadata],
)
update_with_model = SetStatisticsUpdate(statistics=statistics_file)
assert model_roundtrips(update_with_model)
assert update_with_model.snapshot_id == snapshot_id
update_with_dict = SetStatisticsUpdate.model_validate({"statistics": statistics_file.model_dump()})
assert model_roundtrips(update_with_dict)
assert update_with_dict.snapshot_id == snapshot_id
update_json = """
{
"statistics":
{
"snapshot-id": 3055729675574597004,
"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": 3055729675574597004,
"sequence-number": 1,
"fields": [1]
}
]
}
}
"""
update_with_json = SetStatisticsUpdate.model_validate_json(update_json)
assert model_roundtrips(update_with_json)
assert update_with_json.snapshot_id == snapshot_id
def test_remove_statistics_update(table_v2_with_statistics: Table) -> None:
update = RemoveStatisticsUpdate(
snapshot_id=3055729675574597004,
)
remove_metadata = update_table_metadata(
table_v2_with_statistics.metadata,
(update,),
)
assert len(remove_metadata.statistics) == 1
with pytest.raises(
ValueError,
match="Statistics with snapshot id 123456789 does not exist",
):
update_table_metadata(
table_v2_with_statistics.metadata,
(RemoveStatisticsUpdate(snapshot_id=123456789),),
)
def test_set_partition_statistics_update(table_v2_with_statistics: Table) -> None:
snapshot_id = table_v2_with_statistics.metadata.current_snapshot_id
partition_statistics_file = PartitionStatisticsFile(
snapshot_id=snapshot_id,
statistics_path="s3://bucket/warehouse/stats.parquet",
file_size_in_bytes=124,
)
update = SetPartitionStatisticsUpdate(
partition_statistics=partition_statistics_file,
)
# Verify that serialization uses 'partition-statistics' alias
dumped_with_alias = update.model_dump(by_alias=True)
assert "partition-statistics" in dumped_with_alias
new_metadata = update_table_metadata(
table_v2_with_statistics.metadata,
(update,),
)
expected = """
{
"snapshot-id": 3055729675574597004,
"statistics-path": "s3://bucket/warehouse/stats.parquet",
"file-size-in-bytes": 124
}"""
assert len(new_metadata.partition_statistics) == 1
updated_statistics = [stat for stat in new_metadata.partition_statistics if stat.snapshot_id == snapshot_id]
assert len(updated_statistics) == 1
assert json.loads(updated_statistics[0].model_dump_json()) == json.loads(expected)
def test_remove_partition_statistics_update(table_v2_with_statistics: Table) -> None:
# Add partition statistics file.
snapshot_id = table_v2_with_statistics.metadata.current_snapshot_id
partition_statistics_file = PartitionStatisticsFile(
snapshot_id=snapshot_id,
statistics_path="s3://bucket/warehouse/stats.parquet",
file_size_in_bytes=124,
)
update = SetPartitionStatisticsUpdate(
partition_statistics=partition_statistics_file,
)
new_metadata = update_table_metadata(
table_v2_with_statistics.metadata,
(update,),
)
assert len(new_metadata.partition_statistics) == 1
# Remove the same partition statistics file.
remove_update = RemovePartitionStatisticsUpdate(snapshot_id=snapshot_id)
remove_metadata = update_table_metadata(
new_metadata,
(remove_update,),
)
assert len(remove_metadata.partition_statistics) == 0
def test_remove_partition_statistics_update_with_invalid_snapshot_id(table_v2_with_statistics: Table) -> None:
# Remove the same partition statistics file.
with pytest.raises(
ValueError,
match="Partition Statistics with snapshot id 123456789 does not exist",
):
update_table_metadata(
table_v2_with_statistics.metadata,
(RemovePartitionStatisticsUpdate(snapshot_id=123456789),),
)
def test_add_snapshot_update_fails_without_first_row_id(table_v3: Table) -> None:
new_snapshot = Snapshot(
snapshot_id=25,
parent_snapshot_id=19,
sequence_number=200,
timestamp_ms=1602638593590,
manifest_list="s3:/a/b/c.avro",
summary=Summary(Operation.APPEND),
schema_id=3,
)
with pytest.raises(
ValueError,
match="Cannot add snapshot without first row id",
):
update_table_metadata(table_v3.metadata, (AddSnapshotUpdate(snapshot=new_snapshot),))
def test_add_snapshot_update_fails_with_smaller_first_row_id(table_v3: Table) -> None:
new_snapshot = Snapshot(
snapshot_id=25,
parent_snapshot_id=19,
sequence_number=200,
timestamp_ms=1602638593590,
manifest_list="s3:/a/b/c.avro",
summary=Summary(Operation.APPEND),
schema_id=3,
first_row_id=0,
)
with pytest.raises(
ValueError,
match="Cannot add a snapshot with first row id smaller than the table's next-row-id",
):
update_table_metadata(table_v3.metadata, (AddSnapshotUpdate(snapshot=new_snapshot),))
def test_add_snapshot_update_updates_next_row_id(table_v3: Table) -> None:
new_snapshot = Snapshot(
snapshot_id=25,
parent_snapshot_id=19,
sequence_number=200,
timestamp_ms=1602638593590,
manifest_list="s3:/a/b/c.avro",
summary=Summary(Operation.APPEND),
schema_id=3,
first_row_id=2,
added_rows=10,
)
new_metadata = update_table_metadata(table_v3.metadata, (AddSnapshotUpdate(snapshot=new_snapshot),))
assert new_metadata.next_row_id == 11
def model_roundtrips(model: BaseModel) -> bool:
"""Helper assertion that tests if a pydantic model roundtrips
successfully.
"""
__tracebackhide__ = True
model_data = model.model_dump()
if model != type(model).model_validate(model_data):
pytest.fail(f"model {type(model)} did not roundtrip successfully")
return True
def test_check_uuid_raises_when_mismatch(table_v2: Table, example_table_metadata_v2: dict[str, Any]) -> None:
different_uuid = "550e8400-e29b-41d4-a716-446655440000"
metadata_with_different_uuid = {**example_table_metadata_v2, "table-uuid": different_uuid}
new_metadata = TableMetadataV2(**metadata_with_different_uuid)
with pytest.raises(ValueError) as exc_info:
Table._check_uuid(table_v2.metadata, new_metadata)
assert "Table UUID does not match" in str(exc_info.value)
assert different_uuid in str(exc_info.value)
def test_check_uuid_passes_when_match(table_v2: Table, example_table_metadata_v2: dict[str, Any]) -> None:
new_metadata = TableMetadataV2(**example_table_metadata_v2)
# Should not raise with same uuid
Table._check_uuid(table_v2.metadata, new_metadata)
def test_build_large_partition_predicate(table_v2: Table) -> None:
with table_v2.transaction() as tx:
schema = table_v2.schema()
spec = table_v2.spec()
expr = tx._build_partition_predicate(
partition_records={Record(i) for i in range(5000)},
partition_fields=[(schema.find_field(field.source_id).name) for field in spec.fields],
)
bind(table_v2.metadata.schema(), expr, case_sensitive=True)
def test_overwrite_delete_data_file_on_bucket_partition(catalog: Catalog) -> None:
"""Delete a data file without projecting its already-transformed partition value again."""
import pyarrow as pa
from pyiceberg.io.pyarrow import schema_to_pyarrow
catalog.create_namespace("default")
schema = Schema(
NestedField(1, "tenant_id", StringType(), required=False),
NestedField(2, "value", IntegerType(), required=False),
)
spec = PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=BucketTransform(8), name="tenant_id_bucket"))
table = catalog.create_table("default.bucket_file_delete", schema=schema, partition_spec=spec)
table.append(pa.Table.from_pylist([{"tenant_id": "tenant-a", "value": 1}], schema=schema_to_pyarrow(schema)))
data_file = next(iter(table.scan().plan_files())).file
with table.transaction() as transaction:
with transaction.update_snapshot().overwrite() as overwrite:
overwrite.delete_data_file(data_file)
assert table.scan().to_arrow().num_rows == 0
def test_static_table_forwards_location_to_table_file_io(metadata_location: str, monkeypatch: pytest.MonkeyPatch) -> None:
seen_locations: list[str | None] = []
real_load_file_io = load_file_io
def _spy(*args: Any, **kwargs: Any) -> FileIO:
if "location" in kwargs:
seen_locations.append(kwargs["location"])
elif len(args) >= 2:
seen_locations.append(args[1])
else:
seen_locations.append(None)
return real_load_file_io(*args, **kwargs)
monkeypatch.setattr("pyiceberg.table.load_file_io", _spy)
StaticTable.from_metadata(metadata_location)
assert seen_locations, "expected at least one load_file_io call"
assert all(loc is not None for loc in seen_locations), f"load_file_io called without a location: {seen_locations}"