changes_as_per_pr1799_comments

Author: redpheonixx

pyiceberg/io/pyarrow.py

@@ -177,6 +177,7 @@
 from pyiceberg.utils.properties import get_first_property_value, property_as_bool, property_as_int
 from pyiceberg.utils.singleton import Singleton
 from pyiceberg.utils.truncate import truncate_upper_bound_binary_string, truncate_upper_bound_text_string
+from decimal import Decimal
 
 if TYPE_CHECKING:
     from pyiceberg.table import FileScanTask, WriteTask
@@ -1876,7 +1877,11 @@ def visit_fixed(self, fixed_type: FixedType) -> str:
         return "FIXED_LEN_BYTE_ARRAY"
 
     def visit_decimal(self, decimal_type: DecimalType) -> str:
-        return "FIXED_LEN_BYTE_ARRAY"
+        return (
+            "INT32" if decimal_type.precision <= 9 
+            else "INT64" if decimal_type.precision <= 18 
+            else "FIXED_LEN_BYTE_ARRAY"
+        )
 
     def visit_boolean(self, boolean_type: BooleanType) -> str:
         return "BOOLEAN"
@@ -2350,8 +2355,15 @@ def data_file_statistics_from_parquet_metadata(
                             stats_col.iceberg_type, statistics.physical_type, stats_col.mode.length
                         )
 
-                    col_aggs[field_id].update_min(statistics.min)
-                    col_aggs[field_id].update_max(statistics.max)
+                    if isinstance(stats_col.iceberg_type, DecimalType) and statistics.physical_type != "FIXED_LEN_BYTE_ARRAY":
+                        precision= stats_col.iceberg_type.precision
+                        scale = stats_col.iceberg_type.scale
+                        decimal_type = pa.decimal128(precision, scale)
+                        col_aggs[field_id].update_min(pa.array([Decimal(statistics.min_raw)/ (10 ** scale)], decimal_type)[0].as_py())
+                        col_aggs[field_id].update_max(pa.array([Decimal(statistics.max_raw)/ (10 ** scale)], decimal_type)[0].as_py())
+                    else:
+                        col_aggs[field_id].update_min(statistics.min)
+                        col_aggs[field_id].update_max(statistics.max)
 
                 except pyarrow.lib.ArrowNotImplementedError as e:
                     invalidate_col.add(field_id)

tests/io/test_pyarrow_stats.py

@@ -72,7 +72,7 @@
     StringType,
 )
 from pyiceberg.utils.datetime import date_to_days, datetime_to_micros, time_to_micros
-
+from decimal import Decimal
 
 @dataclass(frozen=True)
 class TestStruct:
@@ -446,6 +446,9 @@ def construct_test_table_primitive_types() -> Tuple[pq.FileMetaData, Union[Table
                     {"id": 10, "name": "strings", "required": False, "type": "string"},
                     {"id": 11, "name": "uuids", "required": False, "type": "uuid"},
                     {"id": 12, "name": "binaries", "required": False, "type": "binary"},
+                    {"id": 13, "name": "decimal8", "required": False, "type": "decimal(5, 2)"},
+                    {"id": 14, "name": "decimal16", "required": False, "type": "decimal(16, 6)"},
+                    {"id": 15, "name": "decimal32", "required": False, "type": "decimal(19, 6)"},
                 ],
             },
         ],
@@ -470,6 +473,9 @@ def construct_test_table_primitive_types() -> Tuple[pq.FileMetaData, Union[Table
     strings = ["hello", "world"]
     uuids = [uuid.uuid3(uuid.NAMESPACE_DNS, "foo").bytes, uuid.uuid3(uuid.NAMESPACE_DNS, "bar").bytes]
     binaries = [b"hello", b"world"]
+    decimal8 = pa.array([Decimal('123.45'), Decimal('678.91')], pa.decimal128(8, 2))
+    decimal16 = pa.array([Decimal("12345679.123456"), Decimal("67891234.678912")], pa.decimal128(16, 6))
+    decimal32 = pa.array([Decimal("1234567890123.123456"), Decimal("9876543210703.654321")], pa.decimal128(19, 6))
 
     table = pa.Table.from_pydict(
         {
@@ -485,14 +491,17 @@ def construct_test_table_primitive_types() -> Tuple[pq.FileMetaData, Union[Table
             "strings": strings,
             "uuids": uuids,
             "binaries": binaries,
+            "decimal8": decimal8,
+            "decimal16": decimal16,
+            "decimal32": decimal32,
         },
         schema=arrow_schema,
     )
 
     metadata_collector: List[Any] = []
 
     with pa.BufferOutputStream() as f:
-        with pq.ParquetWriter(f, table.schema, metadata_collector=metadata_collector) as writer:
+        with pq.ParquetWriter(f, table.schema, metadata_collector=metadata_collector, store_decimal_as_integer=True) as writer:
             writer.write_table(table)
 
     return metadata_collector[0], table_metadata
@@ -510,13 +519,13 @@ def test_metrics_primitive_types() -> None:
     )
     datafile = DataFile(**statistics.to_serialized_dict())
 
-    assert len(datafile.value_counts) == 12
-    assert len(datafile.null_value_counts) == 12
+    assert len(datafile.value_counts) == 15
+    assert len(datafile.null_value_counts) == 15
     assert len(datafile.nan_value_counts) == 0
 
     tz = timezone(timedelta(seconds=19800))
 
-    assert len(datafile.lower_bounds) == 12
+    assert len(datafile.lower_bounds) == 15
     assert datafile.lower_bounds[1] == STRUCT_BOOL.pack(False)
     assert datafile.lower_bounds[2] == STRUCT_INT32.pack(23)
     assert datafile.lower_bounds[3] == STRUCT_INT64.pack(2)
@@ -529,8 +538,11 @@ def test_metrics_primitive_types() -> None:
     assert datafile.lower_bounds[10] == b"he"
     assert datafile.lower_bounds[11] == uuid.uuid3(uuid.NAMESPACE_DNS, "foo").bytes
     assert datafile.lower_bounds[12] == b"he"
+    assert datafile.lower_bounds[13][::-1].ljust(4, b'\x00') == STRUCT_INT32.pack(12345)
+    assert datafile.lower_bounds[14][::-1].ljust(8, b'\x00') == STRUCT_INT64.pack(12345679123456)
+    assert str(int.from_bytes(datafile.lower_bounds[15], byteorder='big', signed=True)).encode('utf-8')== b"1234567890123123456"
 
-    assert len(datafile.upper_bounds) == 12
+    assert len(datafile.upper_bounds) == 15
     assert datafile.upper_bounds[1] == STRUCT_BOOL.pack(True)
     assert datafile.upper_bounds[2] == STRUCT_INT32.pack(89)
     assert datafile.upper_bounds[3] == STRUCT_INT64.pack(54)
@@ -543,6 +555,9 @@ def test_metrics_primitive_types() -> None:
     assert datafile.upper_bounds[10] == b"wp"
     assert datafile.upper_bounds[11] == uuid.uuid3(uuid.NAMESPACE_DNS, "bar").bytes
     assert datafile.upper_bounds[12] == b"wp"
+    assert datafile.upper_bounds[13][::-1].ljust(4, b'\x00')== STRUCT_INT32.pack(67891)
+    assert datafile.upper_bounds[14][::-1].ljust(8, b'\x00')== STRUCT_INT64.pack(67891234678912)
+    assert str(int.from_bytes(datafile.upper_bounds[15], byteorder='big', signed=True)).encode('utf-8')== b"9876543210703654321"
 
 
 def construct_test_table_invalid_upper_bound() -> Tuple[pq.FileMetaData, Union[TableMetadataV1, TableMetadataV2]]: