Reject encryption properties on pre-v3 tables

pyiceberg/table/metadata.py

@@ -636,6 +636,28 @@ def model_dump_json(self, exclude_none: bool = True, exclude: Any | None = None,
 TableMetadata = Annotated[TableMetadataV1 | TableMetadataV2 | TableMetadataV3, Field(discriminator="format_version")]
 
 
+def _check_properties_format_version_compatibility(properties: Properties, format_version: int) -> None:
+    """Check that the table properties are compatible for the given table format version.
+
+    Args:
+      properties: The table properties to check.
+      format_version: The Iceberg table format version.
+
+    Raises:
+      ValueError: If the properties are not compatible for the format version.
+    """
+    from pyiceberg.table import TableProperties
+
+    if format_version >= 3:
+        return
+
+    encryption_properties = sorted(
+        {TableProperties.ENCRYPTION_KEY_ID, TableProperties.ENCRYPTION_DATA_KEY_LENGTH} & properties.keys()
+    )
+    if encryption_properties:
+        raise ValueError(f"Invalid properties for v{format_version}: {encryption_properties}")
+
+
 def new_table_metadata(
     schema: Schema,
     partition_spec: PartitionSpec,
@@ -650,6 +672,7 @@ def new_table_metadata(
     format_version = int(properties.pop(TableProperties.FORMAT_VERSION, TableProperties.DEFAULT_FORMAT_VERSION))
 
     schema.check_format_version_compatibility(format_version)
+    _check_properties_format_version_compatibility(properties, format_version)
 
     fresh_schema = assign_fresh_schema_ids(schema)
     fresh_partition_spec = assign_fresh_partition_spec_ids(partition_spec, schema, fresh_schema)

pyiceberg/table/update/__init__.py

@@ -35,6 +35,7 @@
     TableMetadata,
     TableMetadataUtil,
     TableMetadataV3,
+    _check_properties_format_version_compatibility,
 )
 from pyiceberg.table.refs import MAIN_BRANCH, SnapshotRef, SnapshotRefType
 from pyiceberg.table.snapshots import (
@@ -350,6 +351,8 @@ def _(update: SetPropertiesUpdate, base_metadata: TableMetadata, context: _Table
     if len(update.updates) == 0:
         return base_metadata
 
+    _check_properties_format_version_compatibility(update.updates, base_metadata.format_version)
+
     properties = dict(base_metadata.properties)
     properties.update(update.updates)
 

tests/catalog/test_catalog_behaviors.py

@@ -209,6 +209,22 @@ def test_create_v1_table(catalog: Catalog, table_schema_nested: Schema, test_tab
     catalog.drop_table(test_table_identifier)
 
 
+@pytest.mark.parametrize("format_version", [1, 2])
+def test_create_table_rejects_encryption_properties_before_v3(
+    catalog: Catalog, table_schema_nested: Schema, test_table_identifier: Identifier, format_version: int
+) -> None:
+    catalog.create_namespace(Catalog.namespace_from(test_table_identifier))
+
+    with pytest.raises(ValueError, match=rf"Invalid properties for v{format_version}: \['encryption.key-id'\]"):
+        catalog.create_table(
+            test_table_identifier,
+            table_schema_nested,
+            properties={"format-version": str(format_version), "encryption.key-id": "key"},
+        )
+
+    assert not catalog.table_exists(test_table_identifier)
+
+
 def test_create_table_custom_sort_order(catalog: Catalog, table_schema_nested: Schema, test_table_identifier: Identifier) -> None:
     namespace = Catalog.namespace_from(test_table_identifier)
     catalog.create_namespace(namespace)

tests/table/test_init.py

@@ -1025,6 +1025,38 @@ def test_apply_set_properties_update(table_v2: Table) -> None:
     assert new_metadata_add_only.last_updated_ms > base_metadata.last_updated_ms
 
 
+@pytest.mark.parametrize("table_fixture", ["table_v1", "table_v2"])
+@pytest.mark.parametrize("property_name", ["encryption.key-id", "encryption.data-key-length"])
+def test_apply_set_properties_update_rejects_encryption_before_v3(
+    table_fixture: str, property_name: str, request: pytest.FixtureRequest
+) -> None:
+    table: Table = request.getfixturevalue(table_fixture)
+
+    with pytest.raises(ValueError, match=rf"Invalid properties for v{table.format_version}: \['{property_name}'\]"):
+        update_table_metadata(table.metadata, (SetPropertiesUpdate(updates={property_name: "value"}),))
+
+
+def test_apply_set_properties_update_allows_encryption_on_v3(table_v3: Table) -> None:
+    new_metadata = update_table_metadata(
+        table_v3.metadata,
+        (SetPropertiesUpdate(updates={"encryption.key-id": "key", "encryption.data-key-length": "32"}),),
+    )
+
+    assert new_metadata.properties["encryption.key-id"] == "key"
+    assert new_metadata.properties["encryption.data-key-length"] == "32"
+
+
+def test_apply_remove_encryption_property_before_v3(table_v2: Table) -> None:
+    # A pre-v3 table that already has encryption.key-id must be repairable
+    base_metadata = table_v2.metadata.model_copy(
+        update={"properties": {**table_v2.metadata.properties, "encryption.key-id": "key"}}
+    )
+
+    new_metadata = update_table_metadata(base_metadata, (RemovePropertiesUpdate(removals=["encryption.key-id"]),))
+
+    assert "encryption.key-id" not in new_metadata.properties
+
+
 def test_apply_remove_properties_update(table_v2: Table) -> None:
     base_metadata = update_table_metadata(
         table_v2.metadata,