Fokko commented on code in PR #6323:
URL: https://github.com/apache/iceberg/pull/6323#discussion_r1216675280
##########
python/pyiceberg/table/__init__.py:
##########
@@ -69,21 +72,288 @@
import ray
from duckdb import DuckDBPyConnection
+ from pyiceberg.catalog import Catalog
ALWAYS_TRUE = AlwaysTrue()
+class TableUpdates:
+ _table: Table
+ _updates: Tuple[TableUpdate, ...]
+
+ def __init__(self, table: Table, actions: Optional[Tuple[TableUpdate,
...]] = None):
+ self._table = table
+ self._updates = actions or ()
+
+ def _append_updates(self, *new_updates: TableUpdate) -> TableUpdates:
+ """Appends updates to the set of staged updates
+
+ Args:
+ *new_updates: Any new updates
+
+ Raises:
+ ValueError: When the type of update is not unique.
+
+ Returns:
+ A new AlterTable object with the new updates appended
+ """
+ for new_update in new_updates:
+ type_new_update = type(new_update)
+ if any(type(update) == type_new_update for update in
self._updates):
+ raise ValueError(f"Updates in a single commit need to be
unique, duplicate: {type_new_update}")
+ self._updates = self._updates + new_updates
+ return self
+
+ def set_table_version(self, format_version: int) -> TableUpdates:
+ """Sets the table to a certain version
+
+ Args:
+ format_version: The newly set version
+
+ Returns:
+ The alter table builder
+ """
+ raise NotImplementedError("Not yet implemented")
+
+ def set_schema(self, new_schema: Schema) -> TableUpdates:
+ """Set the schema, and updates the current-schema-id
+
+ Args:
+ new_schema: The new schema
+
+ Returns:
+ The alter table builder
+
+ Raises:
+ ValueError: When a schema with the same fields already exists
+ """
+ raise NotImplementedError("Not yet implemented")
+
+ def set_partition_spec(self, spec: PartitionSpec) -> TableUpdates:
+ """Sets the partition spec, and updates the default-spec-id
+
+ Args:
+ spec: The new partition spec
+
+ Returns:
+ The alter table builder
+ """
+ raise NotImplementedError("Not yet implemented")
+
+ def set_sort_order(self, sort_order: SortOrder) -> TableUpdates:
+ """Sets the sort order, and updates the default-sort-order-id
+
+ Args:
+ sort_order: The new sort order
+
+ Returns:
+ The alter table builder
+ """
+ raise NotImplementedError("Not yet implemented")
+
+ def set_properties(self, **updates: str) -> TableUpdates:
+ """Set properties
+
+ When a property is already set, it will be overwritten
+
+ Args:
+ updates: The properties set on the table
+
+ Returns:
+ The alter table builder
+ """
+ return self._append_updates(SetPropertiesUpdate(updates=updates))
+
+ def remove_properties(self, *removals: str) -> TableUpdates:
+ """Removes properties
+
+ Args:
+ removals: Properties to be removed
+
+ Returns:
+ The alter table builder
+ """
+ return self._append_updates(RemovePropertiesUpdate(removals=removals))
+
+ def update_location(self, location: str) -> TableUpdates:
+ """Sets the new table location
+
+ Args:
+ location: The new location of the table
+
+ Returns:
+ The alter table builder
+ """
+ raise NotImplementedError("Not yet implemented")
+
+ def commit(self) -> Table:
+ """Commits the changes to the catalog
+
+ Returns:
+ The table with the updates applied
+ """
+ # Strip the catalog name
+ if len(self._updates) > 0:
+ table_response =
self._table.catalog.update_table(self._table.identifier[1:], self._updates)
+ return Table(
+ self._table.identifier,
+ metadata=table_response.metadata,
+ metadata_location=table_response.metadata_location,
+ io=self._table.io,
+ catalog=self._table.catalog,
+ )
+ else:
+ return self._table
+
+
+class TableUpdateAction(Enum):
+ upgrade_format_version = "upgrade-format-version"
+ add_schema = "add-schema"
+ set_current_schema = "set-current-schema"
+ add_spec = "add-spec"
+ set_default_spec = "set-default-spec"
+ add_sort_order = "add-sort-order"
+ set_default_sort_order = "set-default-sort-order"
+ add_snapshot = "add-snapshot"
+ set_snapshot_ref = "set-snapshot-ref"
+ remove_snapshots = "remove-snapshots"
+ remove_snapshot_ref = "remove-snapshot-ref"
+ set_location = "set-location"
+ set_properties = "set-properties"
+ remove_properties = "remove-properties"
+
+
+class TableUpdate(IcebergBaseModel):
+ action: TableUpdateAction
+
+
+class UpgradeFormatVersionUpdate(TableUpdate):
+ action = TableUpdateAction.upgrade_format_version
+ format_version: int = Field(alias="format-version")
+
+
+class AddSchemaUpdate(TableUpdate):
+ action = TableUpdateAction.add_schema
+ schema_: Schema = Field(alias="schema")
+
+
+class SetCurrentSchemaUpdate(TableUpdate):
+ action = TableUpdateAction.set_current_schema
+ schema_id: int = Field(
+ alias="schema-id", description="Schema ID to set as current, or -1 to
set last added schema", default=-1
+ )
+
+
+class AddPartitionSpecUpdate(TableUpdate):
+ action = TableUpdateAction.add_spec
+ spec: PartitionSpec
+
+
+class SetDefaultSpecUpdate(TableUpdate):
+ action = TableUpdateAction.set_default_spec
+ spec_id: int = Field(
+ alias="spec-id", description="Partition spec ID to set as the default,
or -1 to set last added spec", default=-1
+ )
+
+
+class AddSortOrderUpdate(TableUpdate):
+ action = TableUpdateAction.add_sort_order
+ sort_order: SortOrder = Field(alias="sort-order")
+
+
+class SetDefaultSortOrderUpdate(TableUpdate):
+ action = TableUpdateAction.set_default_sort_order
+ sort_order_id: int = Field(
+ alias="sort-order-id", description="Sort order ID to set as the
default, or -1 to set last added sort order", default=-1
+ )
+
+
+class AddSnapshotUpdate(TableUpdate):
+ action = TableUpdateAction.add_snapshot
+ snapshot: Snapshot
+
+
+class SetSnapshotRefUpdate(TableUpdate):
+ action = TableUpdateAction.set_snapshot_ref
+ ref_name: str = Field(alias="ref-name")
+ type: Literal["tag", "branch"]
+ snapshot_id: int = Field(alias="snapshot-id")
+ max_age_ref_ms: int = Field(alias="max-ref-age-ms")
+ max_snapshot_age_ms: int = Field(alias="max-snapshot-age-ms")
+ min_snapshots_to_keep: int = Field(alias="min-snapshots-to-keep")
+
+
+class RemoveSnapshotsUpdate(TableUpdate):
+ action = TableUpdateAction.remove_snapshots
+ snapshot_ids: List[int] = Field(alias="snapshot-ids")
+
+
+class RemoveSnapshotRefUpdate(TableUpdate):
+ action = TableUpdateAction.remove_snapshot_ref
+ ref_name: str = Field(alias="ref-name")
+
+
+class SetLocationUpdate(TableUpdate):
+ action = TableUpdateAction.set_location
+ location: str
+
+
+class SetPropertiesUpdate(TableUpdate):
+ action = TableUpdateAction.set_properties
+ updates: Dict[str, str]
+
+
+class RemovePropertiesUpdate(TableUpdate):
+ action = TableUpdateAction.remove_properties
+ removals: List[str]
+
+
+class Requirement(Enum):
+ assert_create = "assert-create"
+ assert_table_uuid = "assert-table-uuid"
+ assert_ref_snapshot_id = "assert-ref-snapshot-id"
+ assert_last_assigned_field_id = "assert-last-assigned-field-id"
+ assert_current_schema_id = "assert-current-schema-id"
+ assert_last_assigned_partition_id = "assert-last-assigned-partition-id"
+ assert_default_spec_id = "assert-default-spec-id"
+ assert_default_sort_order_id = "assert-default-sort-order-id"
+
+
+class TableRequirement(IcebergBaseModel):
Review Comment:
I couldn't agree more, updated the spec here:
https://github.com/apache/iceberg/pull/7710#issuecomment-1575557040 Already
pulled the new version into this PR.
--
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]
