smaheshwar-pltr commented on code in PR #3220:
URL: https://github.com/apache/iceberg-python/pull/3220#discussion_r3261049398
##########
pyiceberg/catalog/__init__.py:
##########
@@ -444,6 +463,136 @@ def create_table_if_not_exists(
except TableAlreadyExistsError:
return self.load_table(identifier)
+ def replace_table(
+ self,
+ identifier: str | Identifier,
+ schema: Schema | pa.Schema,
+ location: str | None = None,
+ partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
+ sort_order: SortOrder = UNSORTED_SORT_ORDER,
+ properties: Properties = EMPTY_DICT,
+ ) -> Table:
+ """Atomically replace a table's schema, spec, sort order, location,
and properties.
+
+ The table UUID and history (snapshots, schemas, specs, sort orders)
are preserved.
+ The current snapshot is cleared (main branch ref is removed).
+
+ Args:
+ identifier (str | Identifier): Table identifier.
+ schema (Schema): New table schema.
+ location (str | None): New table location. Defaults to the
existing location.
+ partition_spec (PartitionSpec): New partition spec.
+ sort_order (SortOrder): New sort order.
+ properties (Properties): Properties to apply. Merged on top of the
existing
+ table properties: keys present here override existing values;
existing keys
+ not present here are preserved. To remove a property, follow
up with a
+ transaction that removes it explicitly.
+
+ Returns:
+ Table: the replaced table instance.
+
+ Raises:
+ NoSuchTableError: If the table does not exist.
+ TableAlreadyExistsError: If a view exists at the same identifier.
+ """
+ return self.replace_table_transaction(
+ identifier, schema, location, partition_spec, sort_order,
properties
+ ).commit_transaction()
+
+ def replace_table_transaction(
+ self,
+ identifier: str | Identifier,
+ schema: Schema | pa.Schema,
+ location: str | None = None,
+ partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
+ sort_order: SortOrder = UNSORTED_SORT_ORDER,
+ properties: Properties = EMPTY_DICT,
+ ) -> ReplaceTableTransaction:
+ """Create a ReplaceTableTransaction.
+
+ The transaction can be used to stage additional changes (schema
evolution,
+ partition evolution, etc.) before committing.
+
+ Args:
+ identifier (str | Identifier): Table identifier.
+ schema (Schema): New table schema.
+ location (str | None): New table location. Defaults to the
existing location.
+ partition_spec (PartitionSpec): New partition spec.
+ sort_order (SortOrder): New sort order.
+ properties (Properties): Properties to apply. Merged on top of the
existing
+ table properties: keys present here override existing values;
existing keys
+ not present here are preserved. To remove a property, follow
up with a
+ transaction that removes it explicitly.
+
+ Returns:
+ ReplaceTableTransaction: A transaction for the replace operation.
+
+ Raises:
+ NoSuchTableError: If the table does not exist.
+ TableAlreadyExistsError: If a view exists at the same identifier.
+ """
+ raise NotImplementedError("replace_table_transaction is not supported
for this catalog type")
+
+ def _replace_staged_table(
+ self,
+ identifier: str | Identifier,
+ schema: Schema | pa.Schema,
+ location: str | None,
+ partition_spec: PartitionSpec,
+ sort_order: SortOrder,
+ properties: Properties,
+ ) -> tuple[StagedTable, Schema, PartitionSpec, SortOrder, str]:
+ """Load the existing table and build fresh schema/spec/sort-order for
replacement.
+
+ - reuses existing field IDs by name (from the current schema)
+ - reuses partition field IDs by `(source, transform)` across all specs
(v2+),
+ or carries forward the current spec with `VoidTransform`s (v1)
+ - reassigns sort field IDs against the fresh schema
+ - resolves `location` to the existing table's location when omitted
+
+ Returns:
+ A tuple `(staged_table, fresh_schema, fresh_partition_spec,
fresh_sort_order, resolved_location)`.
+ """
+ existing_table = self.load_table(identifier)
+ existing_metadata = existing_table.metadata
+
+ requested_format_version =
properties.get(TableProperties.FORMAT_VERSION)
+ if requested_format_version is not None and
int(requested_format_version) < existing_metadata.format_version:
+ raise ValueError(
+ f"Cannot downgrade format-version from
{existing_metadata.format_version} to {requested_format_version}"
Review Comment:
Java's
[`buildReplacement`](https://github.com/apache/iceberg/blob/2f6606a247e2b16be46ca6c02fc4cfc2e17691e6/core/src/main/java/org/apache/iceberg/TableMetadata.java#L733-L738)
reads `format-version` from properties and only upgrades. Rejecting downgrade
explicitly here — otherwise `_convert_schema_if_needed` would run with v1
semantics while the actual upgrade silently drops, producing a confusing
mismatch.
##########
pyiceberg/catalog/__init__.py:
##########
@@ -444,6 +449,100 @@ def create_table_if_not_exists(
except TableAlreadyExistsError:
return self.load_table(identifier)
+ @abstractmethod
+ def replace_table_transaction(
+ self,
+ identifier: str | Identifier,
+ schema: Schema | pa.Schema,
+ location: str | None = None,
+ partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
+ sort_order: SortOrder = UNSORTED_SORT_ORDER,
+ properties: Properties = EMPTY_DICT,
+ ) -> ReplaceTableTransaction:
+ """Create a ReplaceTableTransaction.
+
+ The transaction can be used to stage additional changes (schema
evolution,
+ partition evolution, etc.) before committing.
+
+ Args:
+ identifier (str | Identifier): Table identifier.
+ schema (Schema): New table schema.
+ location (str | None): New table location. Defaults to the
existing location.
+ partition_spec (PartitionSpec): New partition spec.
+ sort_order (SortOrder): New sort order.
+ properties (Properties): Properties to apply. Merged on top of the
existing
+ table properties: keys present here override existing values;
existing keys
+ not present here are preserved. To remove a property, follow
up with a
+ transaction that removes it explicitly.
+
+ Returns:
+ ReplaceTableTransaction: A transaction for the replace operation.
+
+ Raises:
+ NoSuchTableError: If the table does not exist.
+ """
+
+ def _replace_staged_table(
Review Comment:
@/var/folders/bq/gpty1dt579d_lh4mfg191y_508jd0k/T/tmp.fKjI183X2T
##########
pyiceberg/catalog/__init__.py:
##########
@@ -444,6 +449,135 @@ def create_table_if_not_exists(
except TableAlreadyExistsError:
return self.load_table(identifier)
+ def replace_table(
+ self,
+ identifier: str | Identifier,
+ schema: Schema | pa.Schema,
+ location: str | None = None,
+ partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
+ sort_order: SortOrder = UNSORTED_SORT_ORDER,
+ properties: Properties = EMPTY_DICT,
+ ) -> Table:
+ """Atomically replace a table's schema, spec, sort order, location,
and properties.
+
+ The table UUID and history (snapshots, schemas, specs, sort orders)
are preserved.
+ The current snapshot is cleared (main branch ref is removed).
+
+ Args:
+ identifier (str | Identifier): Table identifier.
+ schema (Schema): New table schema.
+ location (str | None): New table location. Defaults to the
existing location.
+ partition_spec (PartitionSpec): New partition spec.
+ sort_order (SortOrder): New sort order.
+ properties (Properties): Properties to apply. Merged on top of the
existing
+ table properties: keys present here override existing values;
existing keys
+ not present here are preserved. To remove a property, follow
up with a
+ transaction that removes it explicitly.
+
+ Returns:
+ Table: the replaced table instance.
+
+ Raises:
+ NoSuchTableError: If the table does not exist.
+ """
+ return self.replace_table_transaction(
+ identifier, schema, location, partition_spec, sort_order,
properties
+ ).commit_transaction()
+
+ @abstractmethod
+ def replace_table_transaction(
+ self,
+ identifier: str | Identifier,
+ schema: Schema | pa.Schema,
+ location: str | None = None,
+ partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
+ sort_order: SortOrder = UNSORTED_SORT_ORDER,
+ properties: Properties = EMPTY_DICT,
+ ) -> ReplaceTableTransaction:
+ """Create a ReplaceTableTransaction.
+
+ The transaction can be used to stage additional changes (schema
evolution,
+ partition evolution, etc.) before committing.
+
+ Args:
+ identifier (str | Identifier): Table identifier.
+ schema (Schema): New table schema.
+ location (str | None): New table location. Defaults to the
existing location.
+ partition_spec (PartitionSpec): New partition spec.
+ sort_order (SortOrder): New sort order.
+ properties (Properties): Properties to apply. Merged on top of the
existing
+ table properties: keys present here override existing values;
existing keys
+ not present here are preserved. To remove a property, follow
up with a
+ transaction that removes it explicitly.
+
+ Returns:
+ ReplaceTableTransaction: A transaction for the replace operation.
+
+ Raises:
+ NoSuchTableError: If the table does not exist.
+ """
+
+ def _replace_staged_table(
+ self,
+ identifier: str | Identifier,
+ schema: Schema | pa.Schema,
+ location: str | None,
+ partition_spec: PartitionSpec,
+ sort_order: SortOrder,
+ properties: Properties,
+ ) -> tuple[StagedTable, Schema, PartitionSpec, SortOrder, str]:
+ """Load the existing table and build fresh schema/spec/sort-order for
replacement.
+
+ - reuses existing field IDs by name (from the current schema)
+ - reuses partition field IDs by `(source, transform)` across all specs
(v2+),
+ or carries forward the current spec with `VoidTransform`s (v1)
+ - reassigns sort field IDs against the fresh schema
+ - resolves `location` to the existing table's location when omitted
+
+ Returns:
+ A tuple `(staged_table, fresh_schema, fresh_partition_spec,
fresh_sort_order, resolved_location)`.
+ """
+ existing_table = self.load_table(identifier)
+ existing_metadata = existing_table.metadata
+
+ requested_format_version =
properties.get(TableProperties.FORMAT_VERSION)
+ if requested_format_version is not None and
int(requested_format_version) < existing_metadata.format_version:
+ raise ValueError(
+ f"Cannot downgrade format-version from
{existing_metadata.format_version} to {requested_format_version}"
+ )
+ resolved_format_version = (
+ int(requested_format_version) if requested_format_version is not
None else existing_metadata.format_version
+ )
+ iceberg_schema = self._convert_schema_if_needed(schema,
cast(TableVersion, resolved_format_version))
+ iceberg_schema.check_format_version_compatibility(cast(TableVersion,
resolved_format_version))
Review Comment:
Same call `new_table_metadata` makes
([metadata.py:597](https://github.com/apache/iceberg-python/blob/main/pyiceberg/table/metadata.py#L597)),
and the same check Java's Builder runs inside
[`addSchemaInternal`](https://github.com/apache/iceberg/blob/2f6606a247e2b16be46ca6c02fc4cfc2e17691e6/core/src/main/java/org/apache/iceberg/TableMetadata.java#L1605).
Catches v1-incompatible types up front rather than failing later inside
`AddSchemaUpdate`'s apply path.
##########
pyiceberg/catalog/__init__.py:
##########
@@ -444,6 +449,100 @@ def create_table_if_not_exists(
except TableAlreadyExistsError:
return self.load_table(identifier)
+ @abstractmethod
+ def replace_table_transaction(
+ self,
+ identifier: str | Identifier,
+ schema: Schema | pa.Schema,
+ location: str | None = None,
+ partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
+ sort_order: SortOrder = UNSORTED_SORT_ORDER,
+ properties: Properties = EMPTY_DICT,
+ ) -> ReplaceTableTransaction:
+ """Create a ReplaceTableTransaction.
+
+ The transaction can be used to stage additional changes (schema
evolution,
+ partition evolution, etc.) before committing.
+
+ Args:
+ identifier (str | Identifier): Table identifier.
+ schema (Schema): New table schema.
+ location (str | None): New table location. Defaults to the
existing location.
+ partition_spec (PartitionSpec): New partition spec.
+ sort_order (SortOrder): New sort order.
+ properties (Properties): Properties to apply. Merged on top of the
existing
+ table properties: keys present here override existing values;
existing keys
+ not present here are preserved. To remove a property, follow
up with a
+ transaction that removes it explicitly.
+
+ Returns:
+ ReplaceTableTransaction: A transaction for the replace operation.
+
+ Raises:
+ NoSuchTableError: If the table does not exist.
+ """
+
+ def _replace_staged_table(
+ self,
+ identifier: str | Identifier,
+ schema: Schema | pa.Schema,
+ location: str | None,
+ partition_spec: PartitionSpec,
+ sort_order: SortOrder,
+ properties: Properties,
+ ) -> tuple[StagedTable, Schema, PartitionSpec, SortOrder, str]:
+ """Load the existing table and build fresh schema/spec/sort-order for
replacement.
+
+ - reuses existing field IDs by name (from the current schema)
+ - reuses partition field IDs by `(source, transform)` across all specs
(v2+),
+ or carries forward the current spec with `VoidTransform`s (v1)
+ - reassigns sort field IDs against the fresh schema
+ - resolves `location` to the existing table's location when omitted
+
+ Returns:
+ A tuple `(staged_table, fresh_schema, fresh_partition_spec,
fresh_sort_order, resolved_location)`.
+ """
+ existing_table = self.load_table(identifier)
+ existing_metadata = existing_table.metadata
+
+ requested_format_version =
properties.get(TableProperties.FORMAT_VERSION)
+ if requested_format_version is not None and
int(requested_format_version) < existing_metadata.format_version:
+ raise ValueError(
+ f"Cannot downgrade format-version from
{existing_metadata.format_version} to {requested_format_version}"
Review Comment:
@/var/folders/bq/gpty1dt579d_lh4mfg191y_508jd0k/T/tmp.5HmHlfSl0U
##########
pyiceberg/catalog/__init__.py:
##########
@@ -444,6 +449,135 @@ def create_table_if_not_exists(
except TableAlreadyExistsError:
return self.load_table(identifier)
+ def replace_table(
+ self,
+ identifier: str | Identifier,
+ schema: Schema | pa.Schema,
+ location: str | None = None,
+ partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
+ sort_order: SortOrder = UNSORTED_SORT_ORDER,
+ properties: Properties = EMPTY_DICT,
+ ) -> Table:
+ """Atomically replace a table's schema, spec, sort order, location,
and properties.
+
+ The table UUID and history (snapshots, schemas, specs, sort orders)
are preserved.
+ The current snapshot is cleared (main branch ref is removed).
+
+ Args:
+ identifier (str | Identifier): Table identifier.
+ schema (Schema): New table schema.
+ location (str | None): New table location. Defaults to the
existing location.
+ partition_spec (PartitionSpec): New partition spec.
+ sort_order (SortOrder): New sort order.
+ properties (Properties): Properties to apply. Merged on top of the
existing
+ table properties: keys present here override existing values;
existing keys
+ not present here are preserved. To remove a property, follow
up with a
+ transaction that removes it explicitly.
+
+ Returns:
+ Table: the replaced table instance.
+
+ Raises:
+ NoSuchTableError: If the table does not exist.
+ """
+ return self.replace_table_transaction(
+ identifier, schema, location, partition_spec, sort_order,
properties
+ ).commit_transaction()
+
+ @abstractmethod
+ def replace_table_transaction(
+ self,
+ identifier: str | Identifier,
+ schema: Schema | pa.Schema,
+ location: str | None = None,
+ partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
+ sort_order: SortOrder = UNSORTED_SORT_ORDER,
+ properties: Properties = EMPTY_DICT,
+ ) -> ReplaceTableTransaction:
+ """Create a ReplaceTableTransaction.
+
+ The transaction can be used to stage additional changes (schema
evolution,
+ partition evolution, etc.) before committing.
+
+ Args:
+ identifier (str | Identifier): Table identifier.
+ schema (Schema): New table schema.
+ location (str | None): New table location. Defaults to the
existing location.
+ partition_spec (PartitionSpec): New partition spec.
+ sort_order (SortOrder): New sort order.
+ properties (Properties): Properties to apply. Merged on top of the
existing
+ table properties: keys present here override existing values;
existing keys
+ not present here are preserved. To remove a property, follow
up with a
+ transaction that removes it explicitly.
+
+ Returns:
+ ReplaceTableTransaction: A transaction for the replace operation.
+
+ Raises:
+ NoSuchTableError: If the table does not exist.
+ """
+
+ def _replace_staged_table(
+ self,
+ identifier: str | Identifier,
+ schema: Schema | pa.Schema,
+ location: str | None,
+ partition_spec: PartitionSpec,
+ sort_order: SortOrder,
+ properties: Properties,
+ ) -> tuple[StagedTable, Schema, PartitionSpec, SortOrder, str]:
+ """Load the existing table and build fresh schema/spec/sort-order for
replacement.
+
+ - reuses existing field IDs by name (from the current schema)
+ - reuses partition field IDs by `(source, transform)` across all specs
(v2+),
+ or carries forward the current spec with `VoidTransform`s (v1)
+ - reassigns sort field IDs against the fresh schema
+ - resolves `location` to the existing table's location when omitted
+
+ Returns:
+ A tuple `(staged_table, fresh_schema, fresh_partition_spec,
fresh_sort_order, resolved_location)`.
+ """
+ existing_table = self.load_table(identifier)
+ existing_metadata = existing_table.metadata
+
+ requested_format_version =
properties.get(TableProperties.FORMAT_VERSION)
+ if requested_format_version is not None and
int(requested_format_version) < existing_metadata.format_version:
+ raise ValueError(
+ f"Cannot downgrade format-version from
{existing_metadata.format_version} to {requested_format_version}"
+ )
+ resolved_format_version = (
+ int(requested_format_version) if requested_format_version is not
None else existing_metadata.format_version
+ )
+ iceberg_schema = self._convert_schema_if_needed(schema,
cast(TableVersion, resolved_format_version))
+ iceberg_schema.check_format_version_compatibility(cast(TableVersion,
resolved_format_version))
Review Comment:
Same call `new_table_metadata` makes
([metadata.py:597](https://github.com/apache/iceberg-python/blob/main/pyiceberg/table/metadata.py#L597)),
and the same check Java's Builder runs inside
[`addSchemaInternal`](https://github.com/apache/iceberg/blob/2f6606a247e2b16be46ca6c02fc4cfc2e17691e6/core/src/main/java/org/apache/iceberg/TableMetadata.java#L1605).
Catches v1-incompatible types up front rather than failing later inside
`AddSchemaUpdate`'s apply path.
##########
pyiceberg/catalog/__init__.py:
##########
@@ -924,6 +1057,28 @@ def create_table_transaction(
self._create_staged_table(identifier, schema, location,
partition_spec, sort_order, properties)
)
+ @override
+ def replace_table_transaction(
+ self,
+ identifier: str | Identifier,
+ schema: Schema | pa.Schema,
+ location: str | None = None,
+ partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
+ sort_order: SortOrder = UNSORTED_SORT_ORDER,
+ properties: Properties = EMPTY_DICT,
+ ) -> ReplaceTableTransaction:
+ staged_table, fresh_schema, fresh_spec, fresh_sort_order,
resolved_location = self._replace_staged_table(
Review Comment:
Note: Java's
[`RESTSessionCatalog.replaceTransaction`](https://github.com/apache/iceberg/blob/2f6606a247e2b16be46ca6c02fc4cfc2e17691e6/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java#L1035-L1037)
does a view-existence check before replacing:
https://github.com/apache/iceberg/pull/9012.
I intend on implementing that (+ tests) as a follow-up as it feels like a
nice-to-have, and keeps this PR more concise. I'd be happy to implement it now
if folks would prefer though?
##########
pyiceberg/catalog/__init__.py:
##########
@@ -444,6 +449,100 @@ def create_table_if_not_exists(
except TableAlreadyExistsError:
return self.load_table(identifier)
+ @abstractmethod
+ def replace_table_transaction(
+ self,
+ identifier: str | Identifier,
+ schema: Schema | pa.Schema,
+ location: str | None = None,
+ partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
+ sort_order: SortOrder = UNSORTED_SORT_ORDER,
+ properties: Properties = EMPTY_DICT,
+ ) -> ReplaceTableTransaction:
+ """Create a ReplaceTableTransaction.
+
+ The transaction can be used to stage additional changes (schema
evolution,
+ partition evolution, etc.) before committing.
+
+ Args:
+ identifier (str | Identifier): Table identifier.
+ schema (Schema): New table schema.
+ location (str | None): New table location. Defaults to the
existing location.
+ partition_spec (PartitionSpec): New partition spec.
+ sort_order (SortOrder): New sort order.
+ properties (Properties): Properties to apply. Merged on top of the
existing
+ table properties: keys present here override existing values;
existing keys
+ not present here are preserved. To remove a property, follow
up with a
+ transaction that removes it explicitly.
+
+ Returns:
+ ReplaceTableTransaction: A transaction for the replace operation.
+
+ Raises:
+ NoSuchTableError: If the table does not exist.
+ """
+
+ def _replace_staged_table(
+ self,
+ identifier: str | Identifier,
+ schema: Schema | pa.Schema,
+ location: str | None,
+ partition_spec: PartitionSpec,
+ sort_order: SortOrder,
+ properties: Properties,
+ ) -> tuple[StagedTable, Schema, PartitionSpec, SortOrder, str]:
+ """Load the existing table and build fresh schema/spec/sort-order for
replacement.
+
+ - reuses existing field IDs by name (from the current schema)
+ - reuses partition field IDs by `(source, transform)` across all specs
(v2+),
+ or carries forward the current spec with `VoidTransform`s (v1)
+ - reassigns sort field IDs against the fresh schema
+ - resolves `location` to the existing table's location when omitted
+
+ Returns:
+ A tuple `(staged_table, fresh_schema, fresh_partition_spec,
fresh_sort_order, resolved_location)`.
+ """
+ existing_table = self.load_table(identifier)
+ existing_metadata = existing_table.metadata
+
+ requested_format_version =
properties.get(TableProperties.FORMAT_VERSION)
+ if requested_format_version is not None and
int(requested_format_version) < existing_metadata.format_version:
+ raise ValueError(
+ f"Cannot downgrade format-version from
{existing_metadata.format_version} to {requested_format_version}"
+ )
+ resolved_format_version = (
+ int(requested_format_version) if requested_format_version is not
None else existing_metadata.format_version
+ )
+ iceberg_schema = self._convert_schema_if_needed(schema,
cast(TableVersion, resolved_format_version))
+ iceberg_schema.check_format_version_compatibility(cast(TableVersion,
resolved_format_version))
Review Comment:
@/var/folders/bq/gpty1dt579d_lh4mfg191y_508jd0k/T/tmp.FtSuhtVKHl
##########
tests/catalog/test_catalog_behaviors.py:
##########
@@ -387,6 +387,298 @@ def test_load_table_from_self_identifier(
assert table.metadata == loaded_table.metadata
+_SIMPLE_SCHEMA = Schema(
+ NestedField(field_id=1, name="id", field_type=LongType(), required=False),
+ NestedField(field_id=2, name="data", field_type=StringType(),
required=False),
+)
+
+
+def _create_simple_table(
+ catalog: Catalog,
+ identifier: Identifier,
+ *,
+ schema: Schema = _SIMPLE_SCHEMA,
+ format_version: int = 2,
+ partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
+ properties: dict[str, str] | None = None,
+) -> tuple[Identifier, Schema]:
+ namespace = Catalog.namespace_from(identifier)
+ catalog.create_namespace_if_not_exists(namespace)
+ merged_properties = {"format-version": str(format_version), **(properties
or {})}
+ catalog.create_table(identifier, schema=schema,
partition_spec=partition_spec, properties=merged_properties)
+ return identifier, schema
+
+
+def _simple_data(num_rows: int = 2) -> pa.Table:
+ return pa.Table.from_pydict(
+ {"id": list(range(num_rows)), "data": [chr(ord("a") + i) for i in
range(num_rows)]},
+ schema=pa.schema([pa.field("id", pa.int64()), pa.field("data",
pa.large_string())]),
+ )
+
+
+_REPLACE_SCHEMA = Schema(
+ NestedField(field_id=1, name="id", field_type=LongType(), required=False),
+ NestedField(field_id=2, name="data", field_type=StringType(),
required=False),
+ NestedField(field_id=3, name="extra", field_type=BooleanType(),
required=False),
+)
+
+
+def test_replace_transaction(catalog: Catalog, test_table_identifier:
Identifier) -> None:
+ _, original_schema = _create_simple_table(catalog, test_table_identifier)
+ original = catalog.load_table(test_table_identifier)
+ original.append(_simple_data())
+ original = catalog.load_table(test_table_identifier)
+ old_snapshot_id = original.current_snapshot().snapshot_id # type:
ignore[union-attr]
+ snapshot_log_before = list(original.metadata.snapshot_log)
+ assert len(snapshot_log_before) == 1
+
+ catalog.replace_table(test_table_identifier, schema=_REPLACE_SCHEMA)
+ replaced = catalog.load_table(test_table_identifier)
+
+ # UUID + history preserved, current snapshot cleared, current schema
swapped.
+ assert replaced.metadata.table_uuid == original.metadata.table_uuid
+ assert replaced.metadata.current_snapshot_id is None
+ assert {f.name for f in replaced.schema().fields} == {"id", "data",
"extra"}
+ # Old snapshot kept by identity (not just count), and snapshot_log entries
from before survive.
+ assert any(s.snapshot_id == old_snapshot_id for s in
replaced.metadata.snapshots)
+ assert all(entry in replaced.metadata.snapshot_log for entry in
snapshot_log_before)
+ # Old schema is still in the schemas list alongside the new one.
+ schema_ids = sorted(s.schema_id for s in replaced.metadata.schemas)
+ assert schema_ids == [0, 1]
+ assert replaced.metadata.current_schema_id == 1
+ # Time-travel back to the pre-replace snapshot returns the rows that were
there before.
+ assert
replaced.scan(snapshot_id=old_snapshot_id).to_arrow().equals(_simple_data())
+
+
+def test_complete_replace_transaction(catalog: Catalog, test_table_identifier:
Identifier, tmp_path: Path) -> None:
+ _create_simple_table(catalog, test_table_identifier, properties={"keep":
"yes", "override": "old"})
+ catalog.load_table(test_table_identifier).append(_simple_data())
+ original = catalog.load_table(test_table_identifier)
+ old_snapshot_id = original.current_snapshot().snapshot_id # type:
ignore[union-attr]
+ original_data = original.scan().to_arrow()
+
+ new_location = f"file://{tmp_path}/replaced"
+ new_schema = Schema(
+ NestedField(field_id=1, name="id", field_type=LongType(),
required=False),
+ NestedField(field_id=2, name="data", field_type=StringType(),
required=False),
+ NestedField(field_id=3, name="extra", field_type=BooleanType(),
required=False),
+ )
+ new_spec = PartitionSpec(PartitionField(source_id=1, field_id=1000,
name="id_part", transform=IdentityTransform()))
+ new_sort = SortOrder(SortField(source_id=1, transform=IdentityTransform(),
direction=SortDirection.ASC))
+ new_data = pa.Table.from_pydict(
+ {"id": [10, 20], "data": ["alice", "bob"], "extra": [True, False]},
+ schema=pa.schema([pa.field("id", pa.int64()), pa.field("data",
pa.large_string()), pa.field("extra", pa.bool_())]),
+ )
+
+ with catalog.replace_table_transaction(
+ test_table_identifier,
+ schema=new_schema,
+ partition_spec=new_spec,
+ sort_order=new_sort,
+ location=new_location,
+ properties={"override": "new", "added": "v"},
+ ) as txn:
+ txn.append(new_data)
+
+ replaced = catalog.load_table(test_table_identifier)
+
+ # Identity invariants.
+ assert replaced.metadata.table_uuid == original.metadata.table_uuid
+ assert replaced.metadata.location == new_location
+
+ # New schema / spec / sort applied; old entries retained in history.
+ assert {f.name for f in replaced.schema().fields} == {"id", "data",
"extra"}
+ assert sorted(s.schema_id for s in replaced.metadata.schemas) == [0, 1]
+ assert replaced.spec().fields[0].source_id == 1
+ assert isinstance(replaced.spec().fields[0].transform, IdentityTransform)
+ assert {s.spec_id for s in replaced.metadata.partition_specs} == {0, 1}
+ assert replaced.sort_order().fields == new_sort.fields
+ assert {s.order_id for s in replaced.metadata.sort_orders} == {0,
replaced.metadata.default_sort_order_id}
+
+ # Property merge: kept, overridden, added — and `format-version` does not
leak.
+ assert replaced.properties["keep"] == "yes"
+ assert replaced.properties["override"] == "new"
+ assert replaced.properties["added"] == "v"
+ assert "format-version" not in replaced.properties
+
+ # RTAS atomicity: new snapshot exists, has no parent (fresh start), old
snapshot is still
+ # in the snapshot list, and time-travel reads return the original rows.
+ new_snapshot = replaced.current_snapshot()
+ assert new_snapshot is not None
+ assert new_snapshot.snapshot_id != old_snapshot_id
+ assert new_snapshot.parent_snapshot_id is None
+ assert any(s.snapshot_id == old_snapshot_id for s in
replaced.metadata.snapshots)
+ assert replaced.scan().to_arrow().num_rows == 2
+ # Time-travel back to before the replace returns the original rows from
the old schema.
+ time_travel = replaced.scan(snapshot_id=old_snapshot_id).to_arrow()
+ assert time_travel.num_rows == original_data.num_rows
+ assert time_travel.column("id").to_pylist() ==
original_data.column("id").to_pylist()
+
+
+def test_replace_transaction_requires_table_exists(catalog: Catalog,
test_table_identifier: Identifier) -> None:
+ schema = Schema(NestedField(field_id=1, name="id", field_type=LongType(),
required=False))
+ with pytest.raises(NoSuchTableError):
+ catalog.replace_table(test_table_identifier, schema=schema)
+
+
+def test_replace_table_reuses_schema_id_when_identical(catalog: Catalog,
test_table_identifier: Identifier) -> None:
+ _, base_schema = _create_simple_table(catalog, test_table_identifier)
+ replaced = catalog.replace_table(test_table_identifier, schema=base_schema)
+ # Identical shape -> no new schema appended, current points back at id 0.
+ assert [s.schema_id for s in replaced.metadata.schemas] == [0]
+ assert replaced.metadata.current_schema_id == 0
+ assert replaced.metadata.last_column_id == 2
+
+
+def test_replace_table_reuses_partition_spec_and_sort_order_when_identical(
+ catalog: Catalog, test_table_identifier: Identifier
+) -> None:
+ spec = PartitionSpec(PartitionField(source_id=1, field_id=1000,
name="id_part", transform=IdentityTransform()))
+ sort = SortOrder(SortField(source_id=1, transform=IdentityTransform(),
direction=SortDirection.ASC))
+ _, schema = _create_simple_table(catalog, test_table_identifier,
partition_spec=spec)
+ # Introduce a sort order then replay both spec and sort — neither should
append a new entry.
+ sorted_first = catalog.replace_table(test_table_identifier, schema=schema,
partition_spec=spec, sort_order=sort)
+ sorted_order_id = sorted_first.metadata.default_sort_order_id
+ assert sorted_order_id != 0
+
+ replayed = catalog.replace_table(test_table_identifier, schema=schema,
partition_spec=spec, sort_order=sort)
+ assert [s.spec_id for s in replayed.metadata.partition_specs] == [0]
+ assert replayed.metadata.default_spec_id == 0
+ assert replayed.metadata.default_sort_order_id == sorted_order_id
+
+ # Dropping the sort order falls back to the unsorted order_id 0 (also
reused, not appended).
+ unsorted = catalog.replace_table(test_table_identifier, schema=schema,
partition_spec=spec)
+ assert unsorted.sort_order().is_unsorted
+ assert unsorted.metadata.default_sort_order_id == 0
+
+
[email protected]("keep_identifier", [True, False], ids=["preserves",
"drops"])
+def test_replace_table_identifier_field_ids(catalog: Catalog,
test_table_identifier: Identifier, keep_identifier: bool) -> None:
+ schema = Schema(
+ NestedField(field_id=1, name="id", field_type=LongType(),
required=True),
+ NestedField(field_id=2, name="data", field_type=StringType(),
required=False),
+ identifier_field_ids=[1],
+ )
+ _create_simple_table(catalog, test_table_identifier, schema=schema)
+ new_schema = (
+ Schema(
+ NestedField(field_id=1, name="id", field_type=LongType(),
required=True),
+ NestedField(field_id=2, name="data", field_type=StringType(),
required=False),
+ NestedField(field_id=3, name="extra", field_type=BooleanType(),
required=False),
+ identifier_field_ids=[1],
+ )
+ if keep_identifier
+ else Schema(
+ NestedField(field_id=1, name="id", field_type=LongType(),
required=False),
+ NestedField(field_id=2, name="data", field_type=StringType(),
required=False),
+ )
+ )
+ replaced = catalog.replace_table(test_table_identifier, schema=new_schema)
+ expected = [1] if keep_identifier else []
+ assert list(replaced.schema().identifier_field_ids) == expected
+
+
[email protected](
+ "format_version, expect_void_carry_forward",
+ [(1, True), (2, False)],
+ ids=["v1-carries-forward", "v2-drops"],
+)
+def test_replace_table_partition_field_carry_forward(
+ catalog: Catalog,
+ test_table_identifier: Identifier,
+ format_version: int,
+ expect_void_carry_forward: bool,
+) -> None:
+ spec = PartitionSpec(PartitionField(source_id=1, field_id=1000,
name="id_part", transform=IdentityTransform()))
+ _, schema = _create_simple_table(catalog, test_table_identifier,
partition_spec=spec, format_version=format_version)
+ replaced = catalog.replace_table(test_table_identifier, schema=schema)
+ new_spec = replaced.spec()
+ if expect_void_carry_forward:
+ void_field = next(f for f in new_spec.fields if f.field_id == 1000)
+ assert isinstance(void_field.transform, VoidTransform)
+ assert void_field.source_id == 1
+ assert void_field.name == "id_part"
+ else:
+ assert new_spec.is_unpartitioned()
+
+
+def test_replace_table_upgrades_format_version(catalog: Catalog,
test_table_identifier: Identifier) -> None:
+ _, schema = _create_simple_table(catalog, test_table_identifier,
format_version=1)
+ assert catalog.load_table(test_table_identifier).format_version == 1
+
+ upgraded = catalog.replace_table(test_table_identifier, schema=schema,
properties={"format-version": "2"})
+ assert upgraded.format_version == 2
+ # `format-version` is a control input, not a persisted property.
+ assert "format-version" not in upgraded.properties
+
+ # A follow-up replace stays at the upgraded version.
+ new_schema = Schema(*schema.fields, NestedField(field_id=3, name="extra",
field_type=BooleanType(), required=False))
+ replayed = catalog.replace_table(test_table_identifier, schema=new_schema)
+ assert replayed.format_version == 2
+ assert {f.name for f in replayed.schema().fields} == {"id", "data",
"extra"}
+
+
+def test_replace_table_rejects_format_version_downgrade(catalog: Catalog,
test_table_identifier: Identifier) -> None:
+ _, schema = _create_simple_table(catalog, test_table_identifier,
format_version=2)
+ with pytest.raises(ValueError, match="Cannot downgrade format-version"):
+ catalog.replace_table(test_table_identifier, schema=schema,
properties={"format-version": "1"})
+
+
[email protected]("location_kind", ["inherit", "explicit",
"trailing-slash"])
+def test_replace_table_location(catalog: Catalog, test_table_identifier:
Identifier, tmp_path: Path, location_kind: str) -> None:
+ _, schema = _create_simple_table(catalog, test_table_identifier)
+ existing = catalog.load_table(test_table_identifier).metadata.location
+
+ if location_kind == "inherit":
+ replaced = catalog.replace_table(test_table_identifier, schema=schema)
+ assert replaced.metadata.location == existing
+ else:
+ bare = f"file://{tmp_path}/relocated"
+ arg = bare + "/" if location_kind == "trailing-slash" else bare
+ replaced = catalog.replace_table(test_table_identifier, schema=schema,
location=arg)
+ assert replaced.metadata.location == bare
+
+
+def test_replace_table_transaction_rolls_back_on_failure(catalog: Catalog,
test_table_identifier: Identifier) -> None:
+ _create_simple_table(catalog, test_table_identifier)
+ catalog.load_table(test_table_identifier).append(_simple_data())
+ before = catalog.load_table(test_table_identifier).metadata
+
+ def run_failing_replace() -> None:
+ with catalog.replace_table_transaction(test_table_identifier,
schema=_REPLACE_SCHEMA):
+ raise RuntimeError("simulated failure inside replace transaction")
+
+ with pytest.raises(RuntimeError, match="simulated failure inside replace
transaction"):
+ run_failing_replace()
+
+ after = catalog.load_table(test_table_identifier).metadata
+ assert after.table_uuid == before.table_uuid
+ assert after.current_snapshot_id == before.current_snapshot_id
+ assert after.current_schema_id == before.current_schema_id
+ assert len(after.schemas) == len(before.schemas)
+
+
+def test_concurrent_replace_transaction_schema_conflict(catalog: Catalog,
test_table_identifier: Identifier) -> None:
Review Comment:
Mirrors Java's
[`testConcurrentReplaceTransactionSchemaConflict`](https://github.com/apache/iceberg/blob/2f6606a247e2b16be46ca6c02fc4cfc2e17691e6/core/src/test/java/org/apache/iceberg/catalog/CatalogTests.java#L2866-L2903).
The non-conflict variants (`testConcurrentReplaceTransactions`,
`testConcurrentReplaceTransactionSchema`) are deliberately not ported —
PyIceberg emits `AssertLastAssignedFieldId` unconditionally (see the divergence
note on `commit_transaction`), so those happy-path concurrent flows would
fail-fast here rather than succeed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
