[python] Guard nullable to not-null in update column type

UpdateColumnType carries its own target nullability, but with keep_nullability
false the handler applied it without the null-to-not-null guard that
UpdateColumnNullability enforces. A type change such as BIGINT NOT NULL on a
nullable column therefore succeeded under the default table options, while Java
SchemaManager#updateColumnType rejects it unless
alter-column-null-to-not-null.disabled=false.

Thread disable_null_to_not_null into _handle_update_column_type and call
_assert_nullability_change when keep_nullability is false.

Author: TheR1sing3un

paimon-python/pypaimon/schema/schema_manager.py

@@ -216,7 +216,8 @@ def update_func(field: DataField, depth: int) -> DataField:
 
 
 def _handle_update_column_type(
-    change: UpdateColumnType, new_fields: List[DataField]
+    change: UpdateColumnType, new_fields: List[DataField],
+    disable_null_to_not_null: bool
 ):
     from pypaimon.schema.data_types import DataTypeParser
     field_names = change.field_names
@@ -227,6 +228,13 @@ def update_func(field: DataField, depth: int) -> DataField:
         target_root = DataTypeParser.parse_data_type(change.new_data_type.to_dict())
         if change.keep_nullability:
             target_root.nullable = source_root.nullable
+        else:
+            # A type change carries its own nullability; guard nullable ->
+            # not null just like UpdateColumnNullability (mirrors Java
+            # SchemaManager#updateColumnType).
+            _assert_nullability_change(
+                source_root.nullable, target_root.nullable,
+                '.'.join(field_names), disable_null_to_not_null)
         if not supports_cast(source_root, target_root):
             raise ValueError(
                 "Column type {}[{}] cannot be converted to {} without losing information."
@@ -701,7 +709,8 @@ def _generate_table_schema(
                 _assert_not_updating_primary_keys(
                     old_table_schema, change.field_names, "update"
                 )
-                _handle_update_column_type(change, new_fields)
+                _handle_update_column_type(
+                    change, new_fields, disable_null_to_not_null)
             elif isinstance(change, UpdateColumnNullability):
                 if change.new_nullability:
                     _assert_not_updating_primary_keys(

paimon-python/pypaimon/tests/filesystem_catalog_test.py

@@ -219,6 +219,44 @@ def test_alter_table(self):
         table = catalog.get_table(identifier)
         self.assertEqual(len(table.fields), 2)
 
+    def test_update_column_type_guards_null_to_not_null(self):
+        catalog = CatalogFactory.create({"warehouse": self.warehouse})
+        catalog.create_database("test_db_guard", False)
+
+        def _make_table(name, options):
+            identifier = "test_db_guard.{}".format(name)
+            schema = Schema(
+                fields=[
+                    DataField.from_dict({"id": 0, "name": "k", "type": "INT"}),
+                    DataField.from_dict({"id": 1, "name": "v", "type": "BIGINT"}),
+                ],
+                partition_keys=[], primary_keys=[], options=options, comment="",
+            )
+            catalog.create_table(identifier, schema, False)
+            return identifier
+
+        # Default option (disabled=true) rejects nullable -> not null, mirroring
+        # Java SchemaManager#updateColumnType.
+        default_id = _make_table("default_opt", {})
+        with self.assertRaises(RuntimeError) as ctx:
+            catalog.alter_table(
+                default_id,
+                [SchemaChange.update_column_type(
+                    "v", AtomicType("BIGINT", nullable=False))],
+                False)
+        self.assertIn("nullable to non nullable", str(ctx.exception))
+
+        # Opting out via the table option allows the transition.
+        allowed_id = _make_table(
+            "allow_opt", {"alter-column-null-to-not-null.disabled": "false"})
+        catalog.alter_table(
+            allowed_id,
+            [SchemaChange.update_column_type(
+                "v", AtomicType("BIGINT", nullable=False))],
+            False)
+        table = catalog.get_table(allowed_id)
+        self.assertFalse(table.fields[1].type.nullable)
+
     def test_add_column_before_partition(self):
         catalog = CatalogFactory.create({
             "warehouse": self.warehouse