From 1d02a8744c153dbb251c6bed9980d5eeb75f5ade Mon Sep 17 00:00:00 2001
From: Ankit Raj <ankitraj.srivastava15@gmail.com>
Date: Mon, 26 Oct 2026 15:52:14 +0000
Subject: [PATCH] Reject v3-only types and initial defaults in schema updates
 on pre-v3 tables

---
 pyiceberg/schema.py                     |  5 +++
 pyiceberg/table/update/__init__.py      |  1 +
 tests/catalog/test_catalog_behaviors.py | 43 ++++++++++++++++++++++++-
 3 files changed, 48 insertions(+), 1 deletion(-)

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/tests/catalog/test_catalog_behaviors.py b/tests/catalog/test_catalog_behaviors.py
index 1148ca8527..62832e67f8 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,38 @@ 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_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}"