Reject v3-only types and initial defaults in schema updates on pre-v3 tables

pyiceberg/schema.py

@@ -377,6 +377,11 @@ def check_format_version_compatibility(self, format_version: int) -> None:
                     f"{field.field_type} is only supported in {field.field_type.minimum_format_version()} or higher. "
                     f"Current format version is: {format_version}"
                 )
+            if format_version < 3 and field.initial_default is not None:
+                raise ValueError(
+                    f"Non-null initial-default for {field.name} is only supported in 3 or higher. "
+                    f"Current format version is: {format_version}"
+                )
 
 
 class SchemaVisitor(Generic[T], ABC):

pyiceberg/table/update/__init__.py

@@ -372,6 +372,7 @@ def _(update: RemovePropertiesUpdate, base_metadata: TableMetadata, context: _Ta
 
 @_apply_table_update.register(AddSchemaUpdate)
 def _(update: AddSchemaUpdate, base_metadata: TableMetadata, context: _TableMetadataUpdateContext) -> TableMetadata:
+    update.schema_.check_format_version_compatibility(base_metadata.format_version)
     metadata_updates: dict[str, Any] = {
         "last_column_id": max(base_metadata.last_column_id, update.schema_.highest_field_id),
         "schemas": base_metadata.schemas + [update.schema_],

pyiceberg/table/update/schema.py

@@ -633,6 +633,7 @@ def _commit(self) -> UpdatesAndRequirements:
 
         # Check if it is different current schema ID
         if existing_schema_id != self._schema.schema_id:
+            new_schema.check_format_version_compatibility(self._transaction.table_metadata.format_version)
             requirements += (AssertCurrentSchemaId(current_schema_id=self._schema.schema_id),)
             if existing_schema_id is None:
                 updates += (

tests/catalog/test_catalog_behaviors.py

@@ -48,7 +48,16 @@
 from pyiceberg.table.update import AddSchemaUpdate, SetCurrentSchemaUpdate
 from pyiceberg.transforms import IdentityTransform
 from pyiceberg.typedef import Identifier
-from pyiceberg.types import BooleanType, DoubleType, IntegerType, LongType, NestedField, StringType
+from pyiceberg.types import (
+    BooleanType,
+    DoubleType,
+    IntegerType,
+    LongType,
+    NestedField,
+    StringType,
+    TimestampNanoType,
+    UnknownType,
+)
 
 
 # Name parsing tests
@@ -1322,6 +1331,54 @@ def test_append_invalid_input_type_raises(catalog: Catalog) -> None:
         tbl.append("not an arrow object")
 
 
+@pytest.mark.parametrize("format_version", [1, 2])
+def test_update_schema_rejects_v3_only_types(catalog: Catalog, format_version: int) -> None:
+    catalog.create_namespace("default")
+    identifier = f"default.reject_v3_types_v{format_version}_{catalog.name}"
+    tbl = catalog.create_table(
+        identifier=identifier,
+        schema=Schema(NestedField(1, "id", LongType(), required=False)),
+        properties={"format-version": str(format_version)},
+    )
+
+    with pytest.raises(ValueError, match="timestamp_ns is only supported in 3 or higher"):
+        tbl.update_schema().add_column("ts", TimestampNanoType()).commit()
+    with pytest.raises(ValueError, match="unknown is only supported in 3 or higher"):
+        tbl.update_schema().add_column("u", UnknownType()).commit()
+
+    assert catalog.load_table(identifier).schema() == Schema(NestedField(1, "id", LongType(), required=False))
+
+
+@pytest.mark.parametrize("format_version", [1, 2])
+def test_update_schema_commit_checks_format_version(catalog: Catalog, format_version: int) -> None:
+    catalog.create_namespace("default")
+    identifier = f"default.commit_checks_format_version_v{format_version}_{catalog.name}"
+    tbl = catalog.create_table(
+        identifier=identifier,
+        schema=Schema(NestedField(1, "id", LongType(), required=False)),
+        properties={"format-version": str(format_version)},
+    )
+
+    # The check runs while building the updates, before they reach the table metadata
+    update = tbl.update_schema().add_column("ts", TimestampNanoType())
+    with pytest.raises(ValueError, match="timestamp_ns is only supported in 3 or higher"):
+        update._commit()
+
+
+@pytest.mark.parametrize("format_version", [1, 2])
+def test_update_schema_rejects_initial_default_before_v3(catalog: Catalog, format_version: int) -> None:
+    catalog.create_namespace("default")
+    identifier = f"default.reject_initial_default_v{format_version}_{catalog.name}"
+    tbl = catalog.create_table(
+        identifier=identifier,
+        schema=Schema(NestedField(1, "id", LongType(), required=False)),
+        properties={"format-version": str(format_version)},
+    )
+
+    with pytest.raises(ValueError, match="Non-null initial-default for d is only supported in 3 or higher"):
+        tbl.update_schema().add_column("d", IntegerType(), default_value=5).commit()
+
+
 def test_append_nan_to_identity_partitioned_table(catalog: Catalog) -> None:
     catalog.create_namespace("default")
     identifier = f"default.append_nan_identity_partition_{catalog.name}"

tests/integration/test_reads.py

@@ -1245,6 +1245,7 @@ def test_initial_default(catalog: Catalog, spark: SparkSession) -> None:
 
     # Do the bump version through Spark, since PyIceberg does not support this (yet)
     spark.sql(f"ALTER TABLE {identifier} SET TBLPROPERTIES('format-version'='3')")
+    tbl.refresh()
 
     with tbl.update_schema() as upd:
         upd.add_column("so_true", BooleanType(), required=False, default_value=True)