diff --git a/pyiceberg/schema.py b/pyiceberg/schema.py index 99f983074b..44b6028cd6 100644 --- a/pyiceberg/schema.py +++ b/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): diff --git a/pyiceberg/table/update/__init__.py b/pyiceberg/table/update/__init__.py index aa1a09ebb2..aa8a0d493a 100644 --- a/pyiceberg/table/update/__init__.py +++ b/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_], diff --git a/pyiceberg/table/update/schema.py b/pyiceberg/table/update/schema.py index 828f1e877a..3e872f879b 100644 --- a/pyiceberg/table/update/schema.py +++ b/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 += ( diff --git a/tests/catalog/test_catalog_behaviors.py b/tests/catalog/test_catalog_behaviors.py index 1148ca8527..5cf0026ccc 100644 --- a/tests/catalog/test_catalog_behaviors.py +++ b/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}" diff --git a/tests/integration/test_reads.py b/tests/integration/test_reads.py index a151d62b82..3d7e0b49ed 100644 --- a/tests/integration/test_reads.py +++ b/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)