implement table_metadata.sort_order() and better validation error message

Author: gabeiglio

pyiceberg/table/metadata.py

@@ -318,6 +318,10 @@ def current_snapshot(self) -> Snapshot | None:
     def next_sequence_number(self) -> int:
         return self.last_sequence_number + 1 if self.format_version > 1 else INITIAL_SEQUENCE_NUMBER
 
+    def sort_order(self) -> SortOrder:
+        """Get the current sort order for this table, or UNSORTED_SORT_ORDER if there is no sort order."""
+        return self.sort_order_by_id(self.default_sort_order_id) or UNSORTED_SORT_ORDER
+
     def sort_order_by_id(self, sort_order_id: int) -> SortOrder | None:
         """Get the sort order by sort_order_id."""
         return next((sort_order for sort_order in self.sort_orders if sort_order.order_id == sort_order_id), None)

pyiceberg/table/sorting.py

@@ -178,7 +178,9 @@ def check_compatible(self, schema: Schema) -> None:
             if not source_field.field_type.is_primitive:
                 raise ValidationError(f"Cannot sort by non-primitive source field: {source_field}")
             if not field.transform.can_transform(source_field.field_type):
-                raise ValidationError(f"Invalid source type {source_field.field_type} for transform: {field.transform}")
+                raise ValidationError(
+                    f"Invalid source type {source_field.field_type} for transform: {field.transform} on source field {source_field}"
+                )
 
 
 UNSORTED_SORT_ORDER_ID = 0

pyiceberg/table/update/__init__.py

@@ -708,9 +708,7 @@ def update_table_metadata(
 
     # Check correctness of partition spec, and sort order
     new_metadata.spec().check_compatible(new_metadata.schema())
-
-    if sort_order := new_metadata.sort_order_by_id(new_metadata.default_sort_order_id):
-        sort_order.check_compatible(new_metadata.schema())
+    new_metadata.sort_order().check_compatible(new_metadata.schema())
 
     if enforce_validation:
         return TableMetadataUtil.parse_obj(new_metadata.model_dump())