fix: reject decimal promotion that changes the scale

pyiceberg/schema.py

@@ -1697,7 +1697,7 @@ def _(file_type: BinaryType, read_type: IcebergType) -> IcebergType:
 @promote.register(DecimalType)
 def _(file_type: DecimalType, read_type: IcebergType) -> IcebergType:
     if isinstance(read_type, DecimalType):
-        if file_type.precision <= read_type.precision and file_type.scale == file_type.scale:
+        if file_type.precision <= read_type.precision and file_type.scale == read_type.scale:
             return read_type
         else:
             raise ResolveError(f"Cannot reduce precision from {file_type} to {read_type}")

tests/avro/test_resolver.py

@@ -256,6 +256,15 @@ def test_resolve_decimal_to_decimal_reduce_precision() -> None:
     assert "Cannot reduce precision from decimal(19, 25) to decimal(10, 25)" in str(exc_info.value)
 
 
+def test_resolve_decimal_to_decimal_change_scale() -> None:
+    # Changing the scale is not a valid promotion, even when the precision widens.
+    # Allowing it would reinterpret the file's unscaled integers at the wrong scale.
+    with pytest.raises(ResolveError) as exc_info:
+        _ = resolve_reader(DecimalType(9, 2), DecimalType(18, 4))
+
+    assert "Cannot reduce precision from decimal(9, 2) to decimal(18, 4)" in str(exc_info.value)
+
+
 def test_column_assignment() -> None:
     int_schema = {
         "type": "record",

tests/test_schema.py

@@ -868,6 +868,10 @@ def test_schema_select_cant_be_found(table_schema_nested: Schema) -> None:
     assert "Could not find column: 'BAZ'" in str(exc_info.value)
 
 
+def can_promote_decimal(file_type: DecimalType, read_type: DecimalType) -> bool:
+    return file_type.precision <= read_type.precision and file_type.scale == read_type.scale
+
+
 def should_promote(file_type: IcebergType, read_type: IcebergType) -> bool:
     if isinstance(file_type, IntegerType) and isinstance(read_type, LongType):
         return True
@@ -878,7 +882,7 @@ def should_promote(file_type: IcebergType, read_type: IcebergType) -> bool:
     if isinstance(file_type, BinaryType) and isinstance(read_type, StringType):
         return True
     if isinstance(file_type, DecimalType) and isinstance(read_type, DecimalType):
-        return file_type.precision <= read_type.precision and file_type.scale == file_type.scale
+        return can_promote_decimal(file_type, read_type)
     if isinstance(file_type, FixedType) and isinstance(read_type, UUIDType) and len(file_type) == 16:
         return True
     return False
@@ -945,6 +949,19 @@ def test_promotion(file_type: IcebergType, read_type: IcebergType) -> None:
             promote(file_type, read_type)
 
 
+def test_decimal_promotion() -> None:
+    # Widening precision while keeping the scale fixed is allowed.
+    assert promote(DecimalType(9, 2), DecimalType(18, 2)) == DecimalType(18, 2)
+    # Equal precision and scale resolves to the read type.
+    assert promote(DecimalType(9, 2), DecimalType(9, 2)) == DecimalType(9, 2)
+    # Changing the scale is never allowed, even when the precision widens.
+    with pytest.raises(ResolveError):
+        promote(DecimalType(9, 2), DecimalType(18, 4))
+    # Reducing the precision is not allowed.
+    with pytest.raises(ResolveError):
+        promote(DecimalType(18, 2), DecimalType(9, 2))
+
+
 def test_unknown_type_promotion_to_primitive() -> None:
     """Test that UnknownType can be promoted to primitive types (V3+ behavior)"""
     unknown_type = UnknownType()