fix: Allow update_column to change required for list elements and map values

Closes #2786

The `update_schema().update_column()` method was not updating the
`required` property for list elements or map values. The `_ApplyChanges`
visitor's `list()` and `map()` methods always used the original
`element_required` and `value_required` values without checking for
updates in `self._updates`.

This fix modifies both methods to check `self._updates` for the
element/value field ID and use the updated `required` value if present.

Author: Somasundaram Sekar

pyiceberg/table/update/schema.py

@@ -805,7 +805,11 @@ def list(self, list_type: ListType, element_result: IcebergType | None) -> Icebe
         if element_type is None:
             raise ValueError(f"Cannot delete element type from list: {element_result}")
 
-        return ListType(element_id=list_type.element_id, element=element_type, element_required=list_type.element_required)
+        element_required = list_type.element_required
+        if update := self._updates.get(list_type.element_id):
+            element_required = update.required
+
+        return ListType(element_id=list_type.element_id, element=element_type, element_required=element_required)
 
     def map(self, map_type: MapType, key_result: IcebergType | None, value_result: IcebergType | None) -> IcebergType | None:
         key_id: int = map_type.key_field.field_id
@@ -827,12 +831,16 @@ def map(self, map_type: MapType, key_result: IcebergType | None, value_result: I
         if value_type is None:
             raise ValueError(f"Cannot delete value type from map: {value_field}")
 
+        value_required = map_type.value_required
+        if update := self._updates.get(map_type.value_id):
+            value_required = update.required
+
         return MapType(
             key_id=map_type.key_id,
             key_type=map_type.key_type,
             value_id=map_type.value_id,
             value_type=value_type,
-            value_required=map_type.value_required,
+            value_required=value_required,
         )
 
     def primitive(self, primitive: PrimitiveType) -> IcebergType | None:

tests/table/test_init.py

@@ -627,6 +627,54 @@ def test_add_nested_list_type_column(table_v2: Table) -> None:
     assert new_schema.highest_field_id == 7
 
 
+def test_update_list_element_required(table_v2: Table) -> None:
+    """Test that update_column can change list element's required property."""
+    # Add a list column with optional elements
+    update = UpdateSchema(transaction=table_v2.transaction())
+    list_type = ListType(element_id=1, element_type=StringType(), element_required=False)
+    update.add_column(path="tags", field_type=list_type)
+    schema_with_list = update._apply()
+
+    # Verify initial state
+    field = schema_with_list.find_field("tags")
+    assert isinstance(field.field_type, ListType)
+    assert field.field_type.element_required is False
+
+    # Update element to required
+    update2 = UpdateSchema(transaction=table_v2.transaction(), schema=schema_with_list)
+    update2._allow_incompatible_changes = True  # Allow optional -> required
+    new_schema = update2.update_column(("tags", "element"), required=True)._apply()
+
+    # Verify the update
+    field = new_schema.find_field("tags")
+    assert isinstance(field.field_type, ListType)
+    assert field.field_type.element_required is True
+
+
+def test_update_map_value_required(table_v2: Table) -> None:
+    """Test that update_column can change map value's required property."""
+    # Add a map column with optional values
+    update = UpdateSchema(transaction=table_v2.transaction())
+    map_type = MapType(key_id=1, key_type=StringType(), value_id=2, value_type=IntegerType(), value_required=False)
+    update.add_column(path="metadata", field_type=map_type)
+    schema_with_map = update._apply()
+
+    # Verify initial state
+    field = schema_with_map.find_field("metadata")
+    assert isinstance(field.field_type, MapType)
+    assert field.field_type.value_required is False
+
+    # Update value to required
+    update2 = UpdateSchema(transaction=table_v2.transaction(), schema=schema_with_map)
+    update2._allow_incompatible_changes = True  # Allow optional -> required
+    new_schema = update2.update_column(("metadata", "value"), required=True)._apply()
+
+    # Verify the update
+    field = new_schema.find_field("metadata")
+    assert isinstance(field.field_type, MapType)
+    assert field.field_type.value_required is True
+
+
 def test_apply_set_properties_update(table_v2: Table) -> None:
     base_metadata = table_v2.metadata