Reject conflicted source-id and source-ids fields

pyiceberg/partitioning.py

@@ -109,7 +109,9 @@ def __init__(
     @classmethod
     def map_source_ids_onto_source_id(cls, data: Any) -> Any:
         if isinstance(data, dict):
-            if "source-id" not in data and "source-ids" in data:
+            if "source-ids" in data:
+                if "source-id" in data:
+                    raise ValueError("source-id and source-ids are mutually exclusive")
                 source_ids = data["source-ids"]
                 if isinstance(source_ids, list):
                     if len(source_ids) == 0:

pyiceberg/table/sorting.py

@@ -101,7 +101,9 @@ def set_null_order(cls, values: dict[str, Any]) -> dict[str, Any]:
     @classmethod
     def map_source_ids_onto_source_id(cls, data: Any) -> Any:
         if isinstance(data, dict):
-            if "source-id" not in data and "source-ids" in data:
+            if "source-ids" in data:
+                if "source-id" in data:
+                    raise ValueError("source-id and source-ids are mutually exclusive")
                 source_ids = data["source-ids"]
                 if isinstance(source_ids, list):
                     if len(source_ids) == 0:

tests/table/test_partitioning.py

@@ -273,6 +273,12 @@ def test_deserialize_partition_field_empty_source_ids_rejected() -> None:
         PartitionField.model_validate_json(json_partition_spec)
 
 
+def test_deserialize_partition_field_source_id_and_source_ids_rejected() -> None:
+    json_partition_spec = """{"source-id": 5, "source-ids": [1, 2], "field-id": 1000, "transform": "bucket[4]", "name": "m"}"""
+    with pytest.raises(Exception, match="source-id and source-ids are mutually exclusive"):
+        PartitionField.model_validate_json(json_partition_spec)
+
+
 def test_incompatible_source_column_not_found() -> None:
     schema = Schema(NestedField(1, "foo", IntegerType()), NestedField(2, "bar", IntegerType()))
 

tests/table/test_sorting.py

@@ -144,6 +144,12 @@ def test_deserialize_sort_field_empty_source_ids_rejected() -> None:
         SortField.model_validate_json(payload)
 
 
+def test_deserialize_sort_field_source_id_and_source_ids_rejected() -> None:
+    payload = '{"source-id":5,"source-ids":[19,20],"transform":"bucket[4]","direction":"asc","null-order":"nulls-first"}'
+    with pytest.raises(Exception, match="source-id and source-ids are mutually exclusive"):
+        SortField.model_validate_json(payload)
+
+
 def test_incompatible_source_column_not_found(sort_order: SortOrder) -> None:
     schema = Schema(NestedField(1, "foo", IntegerType()), NestedField(2, "bar", IntegerType()))