Avro: Skip decimal fields using their fixed size

pyiceberg/avro/reader.py

@@ -277,7 +277,7 @@ def read(self, decoder: BinaryDecoder) -> Decimal:
         return bytes_to_decimal(decoder.read(self._length), self.scale)
 
     def skip(self, decoder: BinaryDecoder) -> None:
-        decoder.skip_bytes()
+        decoder.skip(self._length)
 
     def __repr__(self) -> str:
         """Return the string representation of the DecimalReader class."""

tests/avro/test_file.py

@@ -455,6 +455,30 @@ def field_uuid(self) -> UUID:
         assert record[idx] == avro_entry_read_with_fastavro[idx], f"Invalid {field} read with fastavro"
 
 
+@pytest.mark.parametrize("is_required", [True, False])
+@pytest.mark.parametrize("sync_interval", [1, avro.DEFAULT_SYNC_INTERVAL])
+def test_project_decimal_field(is_required: bool, sync_interval: int) -> None:
+    file_schema = Schema(
+        NestedField(field_id=1, name="amount", field_type=DecimalType(6, 2), required=is_required),
+        NestedField(field_id=2, name="id", field_type=IntegerType(), required=True),
+    )
+    records = [Record(Decimal("0.00"), 10), Record(Decimal("123.45"), 42), Record(Decimal("-123.45"), -42)]
+    if not is_required:
+        records.insert(0, Record(None, 0))
+
+    with TemporaryDirectory() as tmpdir:
+        tmp_avro_file = tmpdir + "/decimal_projection.avro"
+        with avro.AvroOutputFile[Record](
+            PyArrowFileIO().new_output(tmp_avro_file), file_schema, "decimal_projection", sync_interval=sync_interval
+        ) as out:
+            out.write_block(records)
+
+        with avro.AvroFile[Record](
+            PyArrowFileIO().new_input(tmp_avro_file), read_schema=Schema(file_schema.find_field(2))
+        ) as avro_reader:
+            assert [record[0] for record in avro_reader] == [record[1] for record in records]
+
+
 def manifest_entry(index: int) -> ManifestEntry:
     return ManifestEntry.from_args(
         status=ManifestEntryStatus.ADDED,

tests/avro/test_reader.py

@@ -68,6 +68,7 @@
     UnknownType,
     UUIDType,
 )
+from pyiceberg.utils.decimal import decimal_required_bytes, decimal_to_bytes, unscaled_to_decimal
 
 AVAILABLE_DECODERS = [StreamingBinaryDecoder, CythonBinaryDecoder]
 
@@ -285,6 +286,20 @@ def test_decimal_reader() -> None:
     assert construct_reader(DecimalType(25, 19)) == DecimalReader(25, 19)
 
 
+@pytest.mark.parametrize("decoder_class", AVAILABLE_DECODERS)
+@pytest.mark.parametrize("precision", range(1, 39))
+@pytest.mark.parametrize("sign", [-1, 0, 1])
+def test_skip_decimal_reader(decoder_class: Callable[[bytes], BinaryDecoder], precision: int, sign: int) -> None:
+    value = unscaled_to_decimal(sign * (10**precision - 1), scale=precision)
+    encoded_value = decimal_to_bytes(value, byte_length=decimal_required_bytes(precision))
+    decoder = decoder_class(encoded_value + b"\x54")
+
+    DecimalReader(precision, precision).skip(decoder)
+
+    assert decoder.tell() == len(encoded_value)
+    assert IntegerReader().read(decoder) == 42
+
+
 def test_boolean_reader() -> None:
     assert construct_reader(BooleanType()) == BooleanReader()