Add deserialization of Bytes -> Decimal

src/io/parquet/read/deserialize/simple.rs

@@ -8,7 +8,7 @@ use parquet2::{
 
 use crate::types::i256;
 use crate::{
-    array::{Array, DictionaryKey, MutablePrimitiveArray, PrimitiveArray},
+    array::{Array, DictionaryKey, MutablePrimitiveArray, PrimitiveArray, BinaryArray},
     datatypes::{DataType, IntervalUnit, TimeUnit},
     error::{Error, Result},
     types::{days_ms, NativeType},
@@ -230,6 +230,36 @@ pub fn page_iter_to_arrays<'a, I: Pages + 'a>(
 
             Box::new(arrays) as _
         }
+        (PhysicalType::ByteArray, Decimal(_, _)) => {
+            let pages = binary::Iter::<i32, _>::new(
+                pages,
+                DataType::Binary,
+                chunk_size,
+                num_rows,
+            );
+
+            let pages = pages.map(move |maybe_array| {
+                let array = maybe_array?;
+                let array_len = array.len();
+                let array = array.as_any().downcast_ref::<BinaryArray<i32>>().unwrap();
+                let values = (0..array_len).map(|i| {
+                    let value = array.value(i);
+                    let n = value.len();
+                    if n > 16 {
+                        return Err(Error::Overflow);
+                    }
+                    Ok(super::super::convert_i128(value, n))
+                })
+                .collect::<Result<Vec<_>>>();
+                let validity = array.validity().cloned();
+
+                PrimitiveArray::<i128>::try_new(data_type.clone(), values?.into(), validity)
+            });
+
+            let arrays = pages.map(|x| x.map(|x| x.boxed()));
+
+            Box::new(arrays) as _
+        }
         (PhysicalType::Int32, Decimal256(_, _)) => dyn_iter(iden(primitive::IntegerIter::new(
             pages,
             data_type,

src/io/parquet/read/schema/convert.rs

@@ -142,6 +142,8 @@ fn from_byte_array(
         (_, Some(PrimitiveConvertedType::Bson)) => DataType::Binary,
         (_, Some(PrimitiveConvertedType::Enum)) => DataType::Binary,
         (_, Some(PrimitiveConvertedType::Utf8)) => DataType::Utf8,
+        (Some(PrimitiveLogicalType::Decimal(precision, scale)), _) => DataType::Decimal(*precision, *scale),
+        (_, Some(PrimitiveConvertedType::Decimal(precision, scale))) => DataType::Decimal(*precision, *scale),
         (_, _) => DataType::Binary,
     }
 }