Support upgrading tables to format version 3

pyiceberg/table/__init__.py

@@ -26,7 +26,7 @@
 from functools import cached_property
 from itertools import chain
 from types import TracebackType
-from typing import TYPE_CHECKING, Any, TypeVar
+from typing import TYPE_CHECKING, Any, TypeVar, cast
 
 from pydantic import Field
 
@@ -48,7 +48,7 @@
 from pyiceberg.table.inspect import InspectTable
 from pyiceberg.table.locations import LocationProvider, load_location_provider
 from pyiceberg.table.maintenance import MaintenanceTable
-from pyiceberg.table.metadata import INITIAL_SEQUENCE_NUMBER, TableMetadata
+from pyiceberg.table.metadata import INITIAL_SEQUENCE_NUMBER, SUPPORTED_TABLE_FORMAT_VERSION, TableMetadata
 from pyiceberg.table.name_mapping import NameMapping
 from pyiceberg.table.refs import MAIN_BRANCH, SnapshotRef
 from pyiceberg.table.snapshots import (
@@ -300,7 +300,7 @@ def upgrade_table_version(self, format_version: TableVersion) -> Transaction:
         Returns:
             The alter table builder.
         """
-        if format_version not in {1, 2}:
+        if not 1 <= format_version <= SUPPORTED_TABLE_FORMAT_VERSION:
             raise ValueError(f"Unsupported table format version: {format_version}")
 
         if format_version < self.table_metadata.format_version:
@@ -316,6 +316,9 @@ def set_properties(self, properties: Properties = EMPTY_DICT, **kwargs: Any) ->
 
         When a property is already set, it will be overwritten.
 
+        Setting the `format-version` property upgrades the table to that format
+        version instead of storing it as a property.
+
         Args:
             properties: The properties set on the table.
             kwargs: properties can also be pass as kwargs.
@@ -325,8 +328,12 @@ def set_properties(self, properties: Properties = EMPTY_DICT, **kwargs: Any) ->
         """
         if properties and kwargs:
             raise ValueError("Cannot pass both properties and kwargs")
-        updates = properties or kwargs
-        return self._apply((SetPropertiesUpdate(updates=updates),))
+        updates = dict(properties or kwargs)
+        if (new_format_version := updates.pop(TableProperties.FORMAT_VERSION, None)) is not None:
+            self.upgrade_table_version(format_version=cast(TableVersion, int(new_format_version)))
+        if updates:
+            return self._apply((SetPropertiesUpdate(updates=updates),))
+        return self
 
     def _set_ref_snapshot(
         self,

pyiceberg/table/metadata.py

@@ -66,7 +66,10 @@
 INITIAL_SPEC_ID = 0
 DEFAULT_SCHEMA_ID = 0
 
-SUPPORTED_TABLE_FORMAT_VERSION = 2
+SUPPORTED_TABLE_FORMAT_VERSION = 3
+
+INITIAL_ROW_ID = 0
+"""The initial value of `next-row-id` for newly upgraded V3 tables."""
 
 
 def cleanup_snapshot_id(data: dict[str, Any]) -> dict[str, Any]:

pyiceberg/table/update/__init__.py

@@ -29,7 +29,7 @@
 from pyiceberg.exceptions import CommitFailedException
 from pyiceberg.partitioning import PARTITION_FIELD_ID_START, PartitionSpec
 from pyiceberg.schema import Schema
-from pyiceberg.table.metadata import SUPPORTED_TABLE_FORMAT_VERSION, TableMetadata, TableMetadataUtil
+from pyiceberg.table.metadata import INITIAL_ROW_ID, SUPPORTED_TABLE_FORMAT_VERSION, TableMetadata, TableMetadataUtil
 from pyiceberg.table.refs import MAIN_BRANCH, SnapshotRef, SnapshotRefType
 from pyiceberg.table.snapshots import (
     MetadataLogEntry,
@@ -321,7 +321,10 @@ def _(
     elif update.format_version == base_metadata.format_version:
         return base_metadata
 
-    updated_metadata = base_metadata.model_copy(update={"format_version": update.format_version})
+    updates: dict[str, Any] = {"format_version": update.format_version}
+    if update.format_version >= 3 and getattr(base_metadata, "next_row_id", None) is None:
+        updates["next_row_id"] = INITIAL_ROW_ID
+    updated_metadata = base_metadata.model_copy(update=updates)
 
     context.add_update(update)
     return TableMetadataUtil._construct_without_validation(updated_metadata)

tests/integration/test_reads.py

@@ -969,10 +969,16 @@ def test_upgrade_table_version(catalog: Catalog) -> None:
             transaction.upgrade_table_version(format_version=1)
     assert "Cannot downgrade v2 table to v1" in str(e.value)
 
+    with table_test_table_version.transaction() as transaction:
+        transaction.upgrade_table_version(format_version=3)
+
+    assert table_test_table_version.format_version == 3
+    assert table_test_table_version.metadata.next_row_id == 0
+
     with pytest.raises(ValueError) as e:
         with table_test_table_version.transaction() as transaction:
-            transaction.upgrade_table_version(format_version=3)
-    assert "Unsupported table format version: 3" in str(e.value)
+            transaction.upgrade_table_version(format_version=4)  # type: ignore[arg-type]
+    assert "Unsupported table format version: 4" in str(e.value)
 
 
 @pytest.mark.integration

tests/table/test_init.py

@@ -83,6 +83,7 @@
     SetPropertiesUpdate,
     SetSnapshotRefUpdate,
     SetStatisticsUpdate,
+    UpgradeFormatVersionUpdate,
     _apply_table_update,
     _TableMetadataUpdateContext,
     update_table_metadata,
@@ -1240,6 +1241,60 @@ def test_update_metadata_with_multiple_updates(table_v1: Table) -> None:
     assert new_metadata.properties == {"owner": "test", "test_a": "test_a1"}
 
 
+def test_update_metadata_upgrade_to_v3(table_v2: Table) -> None:
+    from pyiceberg.table.metadata import TableMetadataV3
+
+    new_metadata = update_table_metadata(table_v2.metadata, (UpgradeFormatVersionUpdate(format_version=3),))
+    assert isinstance(new_metadata, TableMetadataV3)
+    assert new_metadata.format_version == 3
+    assert new_metadata.next_row_id == 0
+
+    # rebuild the metadata to trigger validation
+    validated = TableMetadataUtil.parse_obj(copy(new_metadata.model_dump()))
+    assert validated.format_version == 3
+    assert validated.next_row_id == 0
+
+
+def test_update_metadata_upgrade_v1_to_v3(table_v1: Table) -> None:
+    from pyiceberg.table.metadata import TableMetadataV3
+
+    new_metadata = update_table_metadata(table_v1.metadata, (UpgradeFormatVersionUpdate(format_version=3),))
+    assert isinstance(new_metadata, TableMetadataV3)
+    assert new_metadata.format_version == 3
+    assert new_metadata.next_row_id == 0
+
+    # rebuild the metadata to trigger validation
+    validated = TableMetadataUtil.parse_obj(copy(new_metadata.model_dump()))
+    assert validated.format_version == 3
+    assert validated.next_row_id == 0
+
+
+def test_update_metadata_upgrade_same_version_is_noop(table_v2: Table) -> None:
+    new_metadata = update_table_metadata(table_v2.metadata, (UpgradeFormatVersionUpdate(format_version=2),))
+    assert new_metadata == table_v2.metadata
+    assert new_metadata.format_version == 2
+
+
+def test_set_properties_format_version_upgrades(table_v2: Table) -> None:
+    transaction = table_v2.transaction()
+    transaction.set_properties({"format-version": "3", "owner": "test"})
+
+    assert transaction.table_metadata.format_version == 3
+    assert transaction.table_metadata.next_row_id == 0
+    assert transaction.table_metadata.properties["owner"] == "test"
+    assert "format-version" not in transaction.table_metadata.properties
+    assert UpgradeFormatVersionUpdate(format_version=3) in transaction._updates
+
+
+def test_set_properties_same_format_version_is_noop(table_v2: Table) -> None:
+    transaction = table_v2.transaction()
+    transaction.set_properties({"format-version": "2"})
+
+    assert transaction.table_metadata.format_version == 2
+    assert "format-version" not in transaction.table_metadata.properties
+    assert transaction._updates == ()
+
+
 def test_update_metadata_schema_immutability(
     table_v2_with_fixed_and_decimal_types: TableMetadataV2,
 ) -> None: