Repository: apache/arrow-rs
Pull request #6208 (2024-08-07 16:58:59)

Add time dictionary coercions

tustvold: I wonder if we want to reduce the codegen by coercing to the corresponding primitive array type and then coercing this to a dictionary
adriangb:  Sorry I'm not sure I follow. Are you worried about the verbosity of handling all of these cases?
tustvold: Rust generics are monomorphised so if you call a generic function with lots of different parameters of it leads to lots of additional codegen hurting compile times and bloating your binary

Patch:
diff --git a/arrow-arith/src/aggregate.rs b/arrow-arith/src/aggregate.rs
index a81e7fc1a..a4915f589 100644
--- a/arrow-arith/src/aggregate.rs
+++ b/arrow-arith/src/aggregate.rs
@@ -350,11 +350,46 @@ pub fn min_boolean(array: &BooleanArray) -> Option<bool> {
     }
 
     // Note the min bool is false (0), so short circuit as soon as we see it
-    array
-        .iter()
-        .find(|&b| b == Some(false))
-        .flatten()
-        .or(Some(true))
+    match array.nulls() {
+        None => {
+            let bit_chunks = array.values().bit_chunks();
+            if bit_chunks.iter().any(|x| {
+                // u64::MAX has all bits set, so if the value is not that, then there is a false
+                x != u64::MAX
+            }) {
+                return Some(false);
+            }
+            // If the remainder bits are not all set, then there is a false
+            if bit_chunks.remainder_bits().count_ones() as usize != bit_chunks.remainder_len() {
+                Some(false)
+            } else {
+                Some(true)
+            }
+        }
+        Some(nulls) => {
+            let validity_chunks = nulls.inner().bit_chunks();
+            let value_chunks = array.values().bit_chunks();
+
+            if value_chunks
+                .iter()
+                .zip(validity_chunks.iter())
+                .any(|(value, validity)| {
+                    // We are looking for a false value, but because applying the validity mask
+                    // can create a false for a true value (e.g. value: true, validity: false), we instead invert the value, so that we have to look for a true.
+                    (!value & validity) != 0
+                })
+            {
+                return Some(false);
+            }
+
+            // Same trick as above: Instead of looking for a false, we invert the value bits and look for a true
+            if (!value_chunks.remainder_bits() & validity_chunks.remainder_bits()) != 0 {
+                Some(false)
+            } else {
+                Some(true)
+            }
+        }
+    }
 }
 
 /// Returns the maximum value in the boolean array
@@ -707,10 +742,7 @@ bit_operation!(
 ///
 /// Returns `None` if the array is empty or only contains null values.
 pub fn bool_and(array: &BooleanArray) -> Option<bool> {
-    if array.null_count() == array.len() {
-        return None;
-    }
-    Some(array.false_count() == 0)
+    min_boolean(array)
 }
 
 /// Returns true if any non-null input value is true, otherwise false.
@@ -1382,6 +1414,14 @@ mod tests {
         let a = BooleanArray::from(vec![Some(false), None]);
         assert_eq!(Some(false), min_boolean(&a));
         assert_eq!(Some(false), max_boolean(&a));
+
+        let a = BooleanArray::from(vec![Some(true)]);
+        assert_eq!(Some(true), min_boolean(&a));
+        assert_eq!(Some(true), max_boolean(&a));
+
+        let a = BooleanArray::from(vec![Some(false)]);
+        assert_eq!(Some(false), min_boolean(&a));
+        assert_eq!(Some(false), max_boolean(&a));
     }
 
     #[test]
diff --git a/arrow-cast/src/cast/dictionary.rs b/arrow-cast/src/cast/dictionary.rs
index ee2021d15..daaddc491 100644
--- a/arrow-cast/src/cast/dictionary.rs
+++ b/arrow-cast/src/cast/dictionary.rs
@@ -162,6 +162,26 @@ where
     take(cast_dict_values.as_ref(), dict_array.keys(), None)
 }
 
+/// Pack a data type into a dictionary array passing the values through a primitive array
+pub(crate) fn pack_array_to_dictionary_via_primitive<K: ArrowDictionaryKeyType>(
+    array: &dyn Array,
+    primitive_type: DataType,
+    dict_value_type: &DataType,
+    cast_options: &CastOptions,
+) -> Result<ArrayRef, ArrowError> {
+    let primitive = cast_with_options(array, &primitive_type, cast_options)?;
+    let dict = cast_with_options(
+        primitive.as_ref(),
+        &DataType::Dictionary(Box::new(K::DATA_TYPE), Box::new(primitive_type)),
+        cast_options,
+    )?;
+    cast_with_options(
+        dict.as_ref(),
+        &DataType::Dictionary(Box::new(K::DATA_TYPE), Box::new(dict_value_type.clone())),
+        cast_options,
+    )
+}
+
 /// Attempts to encode an array into an `ArrayDictionary` with index
 /// type K and value (dictionary) type value_type
 ///
@@ -188,6 +208,45 @@ pub(crate) fn cast_to_dictionary<K: ArrowDictionaryKeyType>(
         Decimal256(_, _) => {
             pack_numeric_to_dictionary::<K, Decimal256Type>(array, dict_value_type, cast_options)
         }
+        Float16 => {
+            pack_numeric_to_dictionary::<K, Float16Type>(array, dict_value_type, cast_options)
+        }
+        Float32 => {
+            pack_numeric_to_dictionary::<K, Float32Type>(array, dict_value_type, cast_options)
+        }
+        Float64 => {
+            pack_numeric_to_dictionary::<K, Float64Type>(array, dict_value_type, cast_options)
+        }
+        Date32 => pack_array_to_dictionary_via_primitive::<K>(
+            array,
+            DataType::Int32,
+            dict_value_type,
+            cast_options,
+        ),
+        Date64 => pack_array_to_dictionary_via_primitive::<K>(
+            array,
+            DataType::Int64,
+            dict_value_type,
+            cast_options,
+        ),
+        Time32(_) => pack_array_to_dictionary_via_primitive::<K>(
+            array,
+            DataType::Int32,
+            dict_value_type,
+            cast_options,
+        ),
+        Time64(_) => pack_array_to_dictionary_via_primitive::<K>(
+            array,
+            DataType::Int64,
+            dict_value_type,
+            cast_options,
+        ),
+        Timestamp(_, _) => pack_array_to_dictionary_via_primitive::<K>(
+            array,
+            DataType::Int64,
+            dict_value_type,
+            cast_options,
+        ),
         Utf8 => {
             // If the input is a view type, we can avoid casting (thus copying) the data
             if array.data_type() == &DataType::Utf8View {
diff --git a/arrow-cast/src/cast/mod.rs b/arrow-cast/src/cast/mod.rs
index f6103cb84..93f8a06ea 100644
--- a/arrow-cast/src/cast/mod.rs
+++ b/arrow-cast/src/cast/mod.rs
@@ -225,10 +225,11 @@ pub fn can_cast_types(from_type: &DataType, to_type: &DataType) -> bool {
             | Timestamp(Millisecond, _)
             | Timestamp(Microsecond, _)
             | Timestamp(Nanosecond, _)
-            | Interval(_),
+            | Interval(_)
+            | BinaryView,
         ) => true,
         (Utf8 | LargeUtf8, Utf8View) => true,
-        (BinaryView, Binary | LargeBinary) => true,
+        (BinaryView, Binary | LargeBinary | Utf8 | LargeUtf8 | Utf8View ) => true,
         (Utf8 | LargeUtf8, _) => to_type.is_numeric() && to_type != &Float16,
         (_, Utf8 | LargeUtf8) => from_type.is_primitive(),
 
@@ -1229,6 +1230,9 @@ pub fn cast_with_options(
                 cast_byte_container::<BinaryType, LargeBinaryType>(&binary)
             }
             Utf8View => Ok(Arc::new(StringViewArray::from(array.as_string::<i32>()))),
+            BinaryView => Ok(Arc::new(
+                StringViewArray::from(array.as_string::<i32>()).to_binary_view(),
+            )),
             LargeUtf8 => cast_byte_container::<Utf8Type, LargeUtf8Type>(array),
             Time32(TimeUnit::Second) => parse_string::<Time32SecondType, i32>(array, cast_options),
             Time32(TimeUnit::Millisecond) => {
@@ -1282,6 +1286,7 @@ pub fn cast_with_options(
             Date64 => parse_string_view::<Date64Type>(array, cast_options),
             Binary => cast_view_to_byte::<StringViewType, GenericBinaryType<i32>>(array),
             LargeBinary => cast_view_to_byte::<StringViewType, GenericBinaryType<i64>>(array),
+            BinaryView => Ok(Arc::new(array.as_string_view().clone().to_binary_view())),
             Utf8 => cast_view_to_byte::<StringViewType, GenericStringType<i32>>(array),
             LargeUtf8 => cast_view_to_byte::<StringViewType, GenericStringType<i64>>(array),
             Time32(TimeUnit::Second) => parse_string_view::<Time32SecondType>(array, cast_options),
@@ -1339,6 +1344,13 @@ pub fn cast_with_options(
                 array.as_string::<i64>().clone(),
             ))),
             Utf8View => Ok(Arc::new(StringViewArray::from(array.as_string::<i64>()))),
+            BinaryView => Ok(Arc::new(BinaryViewArray::from(
+                array
+                    .as_string::<i64>()
+                    .into_iter()
+                    .map(|x| x.map(|x| x.as_bytes()))
+                    .collect::<Vec<_>>(),
+            ))),
             Time32(TimeUnit::Second) => parse_string::<Time32SecondType, i64>(array, cast_options),
             Time32(TimeUnit::Millisecond) => {
                 parse_string::<Time32MillisecondType, i64>(array, cast_options)
@@ -1417,6 +1429,20 @@ pub fn cast_with_options(
         (BinaryView, LargeBinary) => {
             cast_view_to_byte::<BinaryViewType, GenericBinaryType<i64>>(array)
         }
+        (BinaryView, Utf8) => {
+            let binary_arr = cast_view_to_byte::<BinaryViewType, GenericBinaryType<i32>>(array)?;
+            cast_binary_to_string::<i32>(&binary_arr, cast_options)
+        }
+        (BinaryView, LargeUtf8) => {
+            let binary_arr = cast_view_to_byte::<BinaryViewType, GenericBinaryType<i64>>(array)?;
+            cast_binary_to_string::<i64>(&binary_arr, cast_options)
+        }
+        (BinaryView, Utf8View) => {
+            Ok(Arc::new(array.as_binary_view().clone().to_string_view()?) as ArrayRef)
+        }
+        (BinaryView, _) => Err(ArrowError::CastError(format!(
+            "Casting from {from_type:?} to {to_type:?} not supported",
+        ))),
         (from_type, LargeUtf8) if from_type.is_primitive() => {
             value_to_string::<i64>(array, cast_options)
         }
@@ -2008,7 +2034,6 @@ pub fn cast_with_options(
                     })?,
             ))
         }
-
         (Date64, Timestamp(TimeUnit::Second, None)) => Ok(Arc::new(
             array
                 .as_primitive::<Date64Type>()
@@ -5256,12 +5281,6 @@ mod tests {
         }
     }
 
-    #[test]
-    fn test_string_to_view() {
-        _test_string_to_view::<i32>();
-        _test_string_to_view::<i64>();
-    }
-
     const VIEW_TEST_DATA: [Option<&str>; 5] = [
         Some("hello"),
         Some("repeated"),
@@ -5270,6 +5289,44 @@ mod tests {
         Some("repeated"),
     ];
 
+    #[test]
+    fn test_string_view_to_binary_view() {
+        let string_view_array = StringViewArray::from_iter(VIEW_TEST_DATA);
+
+        assert!(can_cast_types(
+            string_view_array.data_type(),
+            &DataType::BinaryView
+        ));
+
+        let binary_view_array = cast(&string_view_array, &DataType::BinaryView).unwrap();
+        assert_eq!(binary_view_array.data_type(), &DataType::BinaryView);
+
+        let expect_binary_view_array = BinaryViewArray::from_iter(VIEW_TEST_DATA);
+        assert_eq!(binary_view_array.as_ref(), &expect_binary_view_array);
+    }
+
+    #[test]
+    fn test_binary_view_to_string_view() {
+        let binary_view_array = BinaryViewArray::from_iter(VIEW_TEST_DATA);
+
+        assert!(can_cast_types(
+            binary_view_array.data_type(),
+            &DataType::Utf8View
+        ));
+
+        let string_view_array = cast(&binary_view_array, &DataType::Utf8View).unwrap();
+        assert_eq!(string_view_array.data_type(), &DataType::Utf8View);
+
+        let expect_string_view_array = StringViewArray::from_iter(VIEW_TEST_DATA);
+        assert_eq!(string_view_array.as_ref(), &expect_string_view_array);
+    }
+
+    #[test]
+    fn test_string_to_view() {
+        _test_string_to_view::<i32>();
+        _test_string_to_view::<i64>();
+    }
+
     fn _test_string_to_view<O>()
     where
         O: OffsetSizeTrait,
@@ -5281,11 +5338,22 @@ mod tests {
             &DataType::Utf8View
         ));
 
+        assert!(can_cast_types(
+            string_array.data_type(),
+            &DataType::BinaryView
+        ));
+
         let string_view_array = cast(&string_array, &DataType::Utf8View).unwrap();
         assert_eq!(string_view_array.data_type(), &DataType::Utf8View);
 
+        let binary_view_array = cast(&string_array, &DataType::BinaryView).unwrap();
+        assert_eq!(binary_view_array.data_type(), &DataType::BinaryView);
+
         let expect_string_view_array = StringViewArray::from_iter(VIEW_TEST_DATA);
         assert_eq!(string_view_array.as_ref(), &expect_string_view_array);
+
+        let expect_binary_view_array = BinaryViewArray::from_iter(VIEW_TEST_DATA);
+        assert_eq!(binary_view_array.as_ref(), &expect_binary_view_array);
     }
 
     #[test]
@@ -5380,7 +5448,7 @@ mod tests {
     where
         O: OffsetSizeTrait,
     {
-        let view_array = {
+        let string_view_array = {
             let mut builder = StringViewBuilder::new().with_fixed_block_size(8); // multiple buffers.
             for s in VIEW_TEST_DATA.iter() {
                 builder.append_option(*s);
@@ -5388,15 +5456,21 @@ mod tests {
             builder.finish()
         };
 
+        let binary_view_array = BinaryViewArray::from_iter(VIEW_TEST_DATA);
+
         let expected_string_array = GenericStringArray::<O>::from_iter(VIEW_TEST_DATA);
         let expected_type = expected_string_array.data_type();
 
-        assert!(can_cast_types(view_array.data_type(), expected_type));
+        assert!(can_cast_types(string_view_array.data_type(), expected_type));
+        assert!(can_cast_types(binary_view_array.data_type(), expected_type));
 
-        let string_array = cast(&view_array, expected_type).unwrap();
-        assert_eq!(string_array.data_type(), expected_type);
+        let string_view_casted_array = cast(&string_view_array, expected_type).unwrap();
+        assert_eq!(string_view_casted_array.data_type(), expected_type);
+        assert_eq!(string_view_casted_array.as_ref(), &expected_string_array);
 
-        assert_eq!(string_array.as_ref(), &expected_string_array);
+        let binary_view_casted_array = cast(&binary_view_array, expected_type).unwrap();
+        assert_eq!(binary_view_casted_array.data_type(), expected_type);
+        assert_eq!(binary_view_casted_array.as_ref(), &expected_string_array);
     }
 
     #[test]
@@ -6694,6 +6768,36 @@ mod tests {
         assert_eq!(array_to_strings(&cast_array), expected);
     }
 
+    #[test]
+    fn test_cast_time_array_to_dict() {
+        use DataType::*;
+
+        let array = Arc::new(Date32Array::from(vec![Some(1000), None, Some(2000)])) as ArrayRef;
+
+        let expected = vec!["1972-09-27", "null", "1975-06-24"];
+
+        let cast_type = Dictionary(Box::new(UInt8), Box::new(Date32));
+        let cast_array = cast(&array, &cast_type).expect("cast failed");
+        assert_eq!(cast_array.data_type(), &cast_type);
+        assert_eq!(array_to_strings(&cast_array), expected);
+    }
+
+    #[test]
+    fn test_cast_timestamp_array_to_dict() {
+        use DataType::*;
+
+        let array = Arc::new(
+            TimestampSecondArray::from(vec![Some(1000), None, Some(2000)]).with_timezone_utc(),
+        ) as ArrayRef;
+
+        let expected = vec!["1970-01-01T00:16:40", "null", "1970-01-01T00:33:20"];
+
+        let cast_type = Dictionary(Box::new(UInt8), Box::new(Timestamp(TimeUnit::Second, None)));
+        let cast_array = cast(&array, &cast_type).expect("cast failed");
+        assert_eq!(cast_array.data_type(), &cast_type);
+        assert_eq!(array_to_strings(&cast_array), expected);
+    }
+
     #[test]
     fn test_cast_string_array_to_dict() {
         use DataType::*;
diff --git a/arrow-select/src/filter.rs b/arrow-select/src/filter.rs
index 65ccbe1e0..8e06b07f5 100644
--- a/arrow-select/src/filter.rs
+++ b/arrow-select/src/filter.rs
@@ -552,10 +552,7 @@ fn filter_native<T: ArrowNativeType>(values: &[T], predicate: &FilterPredicate)
 }
 
 /// `filter` implementation for primitive arrays
-pub(crate) fn filter_primitive<T>(
-    array: &PrimitiveArray<T>,
-    predicate: &FilterPredicate,
-) -> PrimitiveArray<T>
+fn filter_primitive<T>(array: &PrimitiveArray<T>, predicate: &FilterPredicate) -> PrimitiveArray<T>
 where
     T: ArrowPrimitiveType,
 {
diff --git a/arrow-select/src/interleave.rs b/arrow-select/src/interleave.rs
index fccc02ac9..a2ab7ba63 100644
--- a/arrow-select/src/interleave.rs
+++ b/arrow-select/src/interleave.rs
@@ -63,7 +63,7 @@ macro_rules! dict_helper {
 ///   values array 1
 /// ```
 ///
-/// For selecting values by index from a single array see [`crate::interleave`]
+/// For selecting values by index from a single array see [`crate::take`]
 pub fn interleave(
     values: &[&dyn Array],
     indices: &[(usize, usize)],
diff --git a/arrow-select/src/take.rs b/arrow-select/src/take.rs
index b66133ac7..ed7179fd3 100644
--- a/arrow-select/src/take.rs
+++ b/arrow-select/src/take.rs
@@ -31,8 +31,6 @@ use arrow_schema::{ArrowError, DataType, FieldRef, UnionMode};
 
 use num::{One, Zero};
 
-use crate::filter::{filter_primitive, FilterBuilder};
-
 /// Take elements by index from [Array], creating a new [Array] from those indexes.
 ///
 /// ```text
@@ -251,13 +249,12 @@ fn take_impl<IndexType: ArrowPrimitiveType>(
             let children = fields.iter()
                 .map(|(field_type_id, _)| {
                     let mask = BooleanArray::from_unary(&type_ids, |value_type_id| value_type_id == field_type_id);
-                    let predicate = FilterBuilder::new(&mask).build();
 
-                    let indices = filter_primitive(&offsets, &predicate);
+                    let indices = crate::filter::filter(&offsets, &mask)?;
 
                     let values = values.child(field_type_id);
 
-                    take_impl(values, &indices)
+                    take_impl(values, indices.as_primitive::<Int32Type>())
                 })
                 .collect::<Result<_, _>>()?;
 
@@ -885,7 +882,7 @@ mod tests {
     use super::*;
     use arrow_array::builder::*;
     use arrow_buffer::{IntervalDayTime, IntervalMonthDayNano};
-    use arrow_schema::{Field, Fields, TimeUnit};
+    use arrow_schema::{Field, Fields, TimeUnit, UnionFields};
 
     fn test_take_decimal_arrays(
         data: Vec<Option<i128>>,
@@ -2308,4 +2305,22 @@ mod tests {
             take(&union, &indices, None).unwrap().to_data()
         );
     }
+
+    #[test]
+    fn test_take_union_dense_all_match_issue_6206() {
+        let fields = UnionFields::new(vec![0], vec![Field::new("a", DataType::Int64, false)]);
+        let ints = Arc::new(Int64Array::from(vec![1, 2, 3, 4, 5]));
+
+        let array = UnionArray::try_new(
+            fields,
+            ScalarBuffer::from(vec![0_i8, 0, 0, 0, 0]),
+            Some(ScalarBuffer::from_iter(0_i32..5)),
+            vec![ints],
+        )
+        .unwrap();
+
+        let indicies = Int64Array::from(vec![0, 2, 4]);
+        let array = take(&array, &indicies, None).unwrap();
+        assert_eq!(array.len(), 3);
+    }
 }
diff --git a/arrow/Cargo.toml b/arrow/Cargo.toml
index 12b6ddd6a..a0fd96415 100644
--- a/arrow/Cargo.toml
+++ b/arrow/Cargo.toml
@@ -56,6 +56,8 @@ arrow-string = { workspace = true }
 rand = { version = "0.8", default-features = false, features = ["std", "std_rng"], optional = true }
 pyo3 = { version = "0.22.2", default-features = false, optional = true }
 
+chrono = { workspace = true, optional = true }
+
 [package.metadata.docs.rs]
 features = ["prettyprint", "ipc_compression", "ffi", "pyarrow"]
 
@@ -70,7 +72,7 @@ prettyprint = ["arrow-cast/prettyprint"]
 # not the core arrow code itself. Be aware that `rand` must be kept as
 # an optional dependency for supporting compile to wasm32-unknown-unknown
 # target without assuming an environment containing JavaScript.
-test_utils = ["rand"]
+test_utils = ["rand", "dep:chrono"]
 pyarrow = ["pyo3", "ffi"]
 # force_validate runs full data validation for all arrays that are created
 # this is not enabled by default as it is too computationally expensive
diff --git a/arrow/benches/cast_kernels.rs b/arrow/benches/cast_kernels.rs
index 8803e8eea..ec7990d3d 100644
--- a/arrow/benches/cast_kernels.rs
+++ b/arrow/benches/cast_kernels.rs
@@ -114,6 +114,18 @@ fn build_decimal256_array(size: usize, precision: u8, scale: i8) -> ArrayRef {
     )
 }
 
+fn build_string_array(size: usize) -> ArrayRef {
+    let mut builder = StringBuilder::new();
+    for v in 0..size {
+        match v % 3 {
+            0 => builder.append_value("small"),
+            1 => builder.append_value("larger string more than 12 bytes"),
+            _ => builder.append_null(),
+        }
+    }
+    Arc::new(builder.finish())
+}
+
 fn build_dict_array(size: usize) -> ArrayRef {
     let values = StringArray::from_iter([
         Some("small"),
@@ -148,9 +160,12 @@ fn add_benchmark(c: &mut Criterion) {
 
     let decimal128_array = build_decimal128_array(512, 10, 3);
     let decimal256_array = build_decimal256_array(512, 50, 3);
+    let string_array = build_string_array(512);
+    let wide_string_array = cast(&string_array, &DataType::LargeUtf8).unwrap();
 
     let dict_array = build_dict_array(10_000);
     let string_view_array = cast(&dict_array, &DataType::Utf8View).unwrap();
+    let binary_view_array = cast(&string_view_array, &DataType::BinaryView).unwrap();
 
     c.bench_function("cast int32 to int32 512", |b| {
         b.iter(|| cast_array(&i32_array, DataType::Int32))
@@ -262,6 +277,30 @@ fn add_benchmark(c: &mut Criterion) {
             )
         })
     });
+    c.bench_function("cast string view to string", |b| {
+        b.iter(|| cast_array(&string_view_array, DataType::Utf8))
+    });
+    c.bench_function("cast string view to wide string", |b| {
+        b.iter(|| cast_array(&string_view_array, DataType::LargeUtf8))
+    });
+    c.bench_function("cast binary view to string", |b| {
+        b.iter(|| cast_array(&binary_view_array, DataType::Utf8))
+    });
+    c.bench_function("cast binary view to wide string", |b| {
+        b.iter(|| cast_array(&binary_view_array, DataType::LargeUtf8))
+    });
+    c.bench_function("cast string to binary view 512", |b| {
+        b.iter(|| cast_array(&string_array, DataType::BinaryView))
+    });
+    c.bench_function("cast wide string to binary view 512", |b| {
+        b.iter(|| cast_array(&wide_string_array, DataType::BinaryView))
+    });
+    c.bench_function("cast string view to binary view", |b| {
+        b.iter(|| cast_array(&string_view_array, DataType::BinaryView))
+    });
+    c.bench_function("cast binary view to string view", |b| {
+        b.iter(|| cast_array(&binary_view_array, DataType::Utf8View))
+    });
 }
 
 criterion_group!(benches, add_benchmark);
diff --git a/arrow/src/util/data_gen.rs b/arrow/src/util/data_gen.rs
index d563fa362..edb406c0f 100644
--- a/arrow/src/util/data_gen.rs
+++ b/arrow/src/util/data_gen.rs
@@ -19,6 +19,7 @@
 
 use std::sync::Arc;
 
+use rand::distributions::uniform::SampleRange;
 use rand::{distributions::uniform::SampleUniform, Rng};
 
 use crate::array::*;
@@ -117,30 +118,39 @@ pub fn create_random_array(
             size,
             primitive_null_density,
         )),
-        Timestamp(_, _) => {
-            let int64_array = Arc::new(create_primitive_array::<Int64Type>(
-                size,
-                primitive_null_density,
-            )) as ArrayRef;
-            return crate::compute::cast(&int64_array, field.data_type());
+        Timestamp(unit, _) => {
+            match unit {
+                TimeUnit::Second => Arc::new(create_random_temporal_array::<TimestampSecondType>(
+                    size,
+                    primitive_null_density,
+                )),
+                TimeUnit::Millisecond => Arc::new(create_random_temporal_array::<
+                    TimestampMillisecondType,
+                >(size, primitive_null_density)),
+                TimeUnit::Microsecond => Arc::new(create_random_temporal_array::<
+                    TimestampMicrosecondType,
+                >(size, primitive_null_density)),
+                TimeUnit::Nanosecond => Arc::new(create_random_temporal_array::<
+                    TimestampNanosecondType,
+                >(size, primitive_null_density)),
+            }
         }
-        Date32 => Arc::new(create_primitive_array::<Date32Type>(
+        Date32 => Arc::new(create_random_temporal_array::<Date32Type>(
             size,
             primitive_null_density,
         )),
-        Date64 => Arc::new(create_primitive_array::<Date64Type>(
+        Date64 => Arc::new(create_random_temporal_array::<Date64Type>(
             size,
             primitive_null_density,
         )),
         Time32(unit) => match unit {
-            TimeUnit::Second => Arc::new(create_primitive_array::<Time32SecondType>(
+            TimeUnit::Second => Arc::new(create_random_temporal_array::<Time32SecondType>(
                 size,
                 primitive_null_density,
             )) as ArrayRef,
-            TimeUnit::Millisecond => Arc::new(create_primitive_array::<Time32MillisecondType>(
-                size,
-                primitive_null_density,
-            )),
+            TimeUnit::Millisecond => Arc::new(
+                create_random_temporal_array::<Time32MillisecondType>(size, primitive_null_density),
+            ),
             _ => {
                 return Err(ArrowError::InvalidArgumentError(format!(
                     "Unsupported unit {unit:?} for Time32"
@@ -148,11 +158,10 @@ pub fn create_random_array(
             }
         },
         Time64(unit) => match unit {
-            TimeUnit::Microsecond => Arc::new(create_primitive_array::<Time64MicrosecondType>(
-                size,
-                primitive_null_density,
-            )) as ArrayRef,
-            TimeUnit::Nanosecond => Arc::new(create_primitive_array::<Time64NanosecondType>(
+            TimeUnit::Microsecond => Arc::new(
+                create_random_temporal_array::<Time64MicrosecondType>(size, primitive_null_density),
+            ) as ArrayRef,
+            TimeUnit::Nanosecond => Arc::new(create_random_temporal_array::<Time64NanosecondType>(
                 size,
                 primitive_null_density,
             )),
@@ -382,6 +391,124 @@ fn create_random_null_buffer(size: usize, null_density: f32) -> Buffer {
     mut_buf.into()
 }
 
+/// Useful for testing. The range of values are not likely to be representative of the
+/// actual bounds.
+pub trait RandomTemporalValue: ArrowTemporalType {
+    fn value_range() -> impl SampleRange<Self::Native>;
+
+    fn gen_range<R: Rng>(rng: &mut R) -> Self::Native
+    where
+        Self::Native: SampleUniform,
+    {
+        rng.gen_range(Self::value_range())
+    }
+
+    fn random<R: Rng>(rng: &mut R) -> Self::Native
+    where
+        Self::Native: SampleUniform,
+    {
+        Self::gen_range(rng)
+    }
+}
+
+impl RandomTemporalValue for TimestampSecondType {
+    /// Range of values for a timestamp in seconds. The range begins at the start
+    /// of the unix epoch and continues for 100 years.
+    fn value_range() -> impl SampleRange<Self::Native> {
+        0..60 * 60 * 24 * 365 * 100
+    }
+}
+
+impl RandomTemporalValue for TimestampMillisecondType {
+    /// Range of values for a timestamp in milliseconds. The range begins at the start
+    /// of the unix epoch and continues for 100 years.
+    fn value_range() -> impl SampleRange<Self::Native> {
+        0..1_000 * 60 * 60 * 24 * 365 * 100
+    }
+}
+
+impl RandomTemporalValue for TimestampMicrosecondType {
+    /// Range of values for a timestamp in microseconds. The range begins at the start
+    /// of the unix epoch and continues for 100 years.
+    fn value_range() -> impl SampleRange<Self::Native> {
+        0..1_000 * 1_000 * 60 * 60 * 24 * 365 * 100
+    }
+}
+
+impl RandomTemporalValue for TimestampNanosecondType {
+    /// Range of values for a timestamp in nanoseconds. The range begins at the start
+    /// of the unix epoch and continues for 100 years.
+    fn value_range() -> impl SampleRange<Self::Native> {
+        0..1_000 * 1_000 * 1_000 * 60 * 60 * 24 * 365 * 100
+    }
+}
+
+impl RandomTemporalValue for Date32Type {
+    /// Range of values representing the elapsed time since UNIX epoch in days. The
+    /// range begins at the start of the unix epoch and continues for 100 years.
+    fn value_range() -> impl SampleRange<Self::Native> {
+        0..365 * 100
+    }
+}
+
+impl RandomTemporalValue for Date64Type {
+    /// Range of values  representing the elapsed time since UNIX epoch in milliseconds.
+    /// The range begins at the start of the unix epoch and continues for 100 years.
+    fn value_range() -> impl SampleRange<Self::Native> {
+        0..1_000 * 60 * 60 * 24 * 365 * 100
+    }
+}
+
+impl RandomTemporalValue for Time32SecondType {
+    /// Range of values representing the elapsed time since midnight in seconds. The
+    /// range is from 0 to 24 hours.
+    fn value_range() -> impl SampleRange<Self::Native> {
+        0..60 * 60 * 24
+    }
+}
+
+impl RandomTemporalValue for Time32MillisecondType {
+    /// Range of values representing the elapsed time since midnight in milliseconds. The
+    /// range is from 0 to 24 hours.
+    fn value_range() -> impl SampleRange<Self::Native> {
+        0..1_000 * 60 * 60 * 24
+    }
+}
+
+impl RandomTemporalValue for Time64MicrosecondType {
+    /// Range of values representing the elapsed time since midnight in microseconds. The
+    /// range is from 0 to 24 hours.
+    fn value_range() -> impl SampleRange<Self::Native> {
+        0..1_000 * 1_000 * 60 * 60 * 24
+    }
+}
+
+impl RandomTemporalValue for Time64NanosecondType {
+    /// Range of values representing the elapsed time since midnight in nanoseconds. The
+    /// range is from 0 to 24 hours.
+    fn value_range() -> impl SampleRange<Self::Native> {
+        0..1_000 * 1_000 * 1_000 * 60 * 60 * 24
+    }
+}
+
+fn create_random_temporal_array<T>(size: usize, null_density: f32) -> PrimitiveArray<T>
+where
+    T: RandomTemporalValue,
+    <T as ArrowPrimitiveType>::Native: SampleUniform,
+{
+    let mut rng = seedable_rng();
+
+    (0..size)
+        .map(|_| {
+            if rng.gen::<f32>() < null_density {
+                None
+            } else {
+                Some(T::random(&mut rng))
+            }
+        })
+        .collect()
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/object_store/src/client/retry.rs b/object_store/src/client/retry.rs
index 5df4ce059..1fc689cdf 100644
--- a/object_store/src/client/retry.rs
+++ b/object_store/src/client/retry.rs
@@ -86,6 +86,14 @@ impl Error {
                 path,
                 source: Box::new(self),
             },
+            Some(StatusCode::FORBIDDEN) => crate::Error::PermissionDenied {
+                path,
+                source: Box::new(self),
+            },
+            Some(StatusCode::UNAUTHORIZED) => crate::Error::Unauthenticated {
+                path,
+                source: Box::new(self),
+            },
             _ => crate::Error::Generic {
                 store,
                 source: Box::new(self),
@@ -106,6 +114,10 @@ impl From<Error> for std::io::Error {
                 status: StatusCode::BAD_REQUEST,
                 ..
             } => Self::new(ErrorKind::InvalidInput, err),
+            Error::Client {
+                status: StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN,
+                ..
+            } => Self::new(ErrorKind::PermissionDenied, err),
             Error::Reqwest { source, .. } if source.is_timeout() => {
                 Self::new(ErrorKind::TimedOut, err)
             }
diff --git a/object_store/src/lib.rs b/object_store/src/lib.rs
index 7699477b9..4184d58a0 100644
--- a/object_store/src/lib.rs
+++ b/object_store/src/lib.rs
@@ -1274,6 +1274,26 @@ pub enum Error {
     #[snafu(display("Operation not yet implemented."))]
     NotImplemented,
 
+    #[snafu(display(
+        "The operation lacked the necessary privileges to complete for path {}: {}",
+        path,
+        source
+    ))]
+    PermissionDenied {
+        path: String,
+        source: Box<dyn std::error::Error + Send + Sync + 'static>,
+    },
+
+    #[snafu(display(
+        "The operation lacked valid authentication credentials for path {}: {}",
+        path,
+        source
+    ))]
+    Unauthenticated {
+        path: String,
+        source: Box<dyn std::error::Error + Send + Sync + 'static>,
+    },
+
     #[snafu(display("Configuration key: '{}' is not valid for store '{}'.", key, store))]
     UnknownConfigurationKey { store: &'static str, key: String },
 }
diff --git a/parquet/benches/arrow_reader.rs b/parquet/benches/arrow_reader.rs
index 814e75c24..18e16f0a4 100644
--- a/parquet/benches/arrow_reader.rs
+++ b/parquet/benches/arrow_reader.rs
@@ -20,6 +20,7 @@ use arrow::datatypes::DataType;
 use arrow_schema::Field;
 use criterion::measurement::WallTime;
 use criterion::{criterion_group, criterion_main, BenchmarkGroup, Criterion};
+use half::f16;
 use num::FromPrimitive;
 use num_bigint::BigInt;
 use parquet::arrow::array_reader::{
@@ -65,6 +66,8 @@ fn build_test_schema() -> SchemaDescPtr {
             }
             REQUIRED BYTE_ARRAY mandatory_binary_leaf;
             OPTIONAL BYTE_ARRAY optional_binary_leaf;
+            REQUIRED FIXED_LEN_BYTE_ARRAY (2) mandatory_f16_leaf (Float16);
+            OPTIONAL FIXED_LEN_BYTE_ARRAY (2) optional_f16_leaf (Float16);
         }
         ";
     parse_message_type(message_type)
@@ -84,6 +87,64 @@ pub fn seedable_rng() -> StdRng {
     StdRng::seed_from_u64(42)
 }
 
+// support byte array for float16
+fn build_encoded_f16_bytes_page_iterator<T>(
+    column_desc: ColumnDescPtr,
+    null_density: f32,
+    encoding: Encoding,
+    min: f32,
+    max: f32,
+) -> impl PageIterator + Clone
+where
+    T: parquet::data_type::DataType,
+    T::T: From<Vec<u8>>,
+{
+    let max_def_level = column_desc.max_def_level();
+    let max_rep_level = column_desc.max_rep_level();
+    let rep_levels = vec![0; VALUES_PER_PAGE];
+    let mut rng = seedable_rng();
+    let mut pages: Vec<Vec<parquet::column::page::Page>> = Vec::new();
+    for _i in 0..NUM_ROW_GROUPS {
+        let mut column_chunk_pages = Vec::new();
+        for _j in 0..PAGES_PER_GROUP {
+            // generate page
+            let mut values = Vec::with_capacity(VALUES_PER_PAGE);
+            let mut def_levels = Vec::with_capacity(VALUES_PER_PAGE);
+            for _k in 0..VALUES_PER_PAGE {
+                let def_level = if rng.gen::<f32>() < null_density {
+                    max_def_level - 1
+                } else {
+                    max_def_level
+                };
+                if def_level == max_def_level {
+                    // create the Float16 value
+                    let value = f16::from_f32(rng.gen_range(min..max));
+                    // Float16 in parquet is stored little-endian
+                    let bytes = match column_desc.physical_type() {
+                        Type::FIXED_LEN_BYTE_ARRAY => {
+                            // Float16 annotates FIXED_LEN_BYTE_ARRAY(2)
+                            assert_eq!(column_desc.type_length(), 2);
+                            value.to_le_bytes().to_vec()
+                        }
+                        _ => unimplemented!(),
+                    };
+                    let value = T::T::from(bytes);
+                    values.push(value);
+                }
+                def_levels.push(def_level);
+            }
+            let mut page_builder =
+                DataPageBuilderImpl::new(column_desc.clone(), values.len() as u32, true);
+            page_builder.add_rep_levels(max_rep_level, &rep_levels);
+            page_builder.add_def_levels(max_def_level, &def_levels);
+            page_builder.add_values::<T>(encoding, &values);
+            column_chunk_pages.push(page_builder.consume());
+        }
+        pages.push(column_chunk_pages);
+    }
+    InMemoryPageIterator::new(pages)
+}
+
 // support byte array for decimal
 fn build_encoded_decimal_bytes_page_iterator<T>(
     column_desc: ColumnDescPtr,
@@ -494,6 +555,19 @@ fn create_primitive_array_reader(
     }
 }
 
+fn create_f16_by_bytes_reader(
+    page_iterator: impl PageIterator + 'static,
+    column_desc: ColumnDescPtr,
+) -> Box<dyn ArrayReader> {
+    let physical_type = column_desc.physical_type();
+    match physical_type {
+        Type::FIXED_LEN_BYTE_ARRAY => {
+            make_fixed_len_byte_array_reader(Box::new(page_iterator), column_desc, None).unwrap()
+        }
+        _ => unimplemented!(),
+    }
+}
+
 fn create_decimal_by_bytes_reader(
     page_iterator: impl PageIterator + 'static,
     column_desc: ColumnDescPtr,
@@ -616,6 +690,131 @@ fn bench_byte_decimal<T>(
     });
 }
 
+fn bench_byte_stream_split_f16<T>(
+    group: &mut BenchmarkGroup<WallTime>,
+    mandatory_column_desc: &ColumnDescPtr,
+    optional_column_desc: &ColumnDescPtr,
+    min: f32,
+    max: f32,
+) where
+    T: parquet::data_type::DataType,
+    T::T: From<Vec<u8>>,
+{
+    let mut count: usize = 0;
+
+    // byte_stream_split encoded, no NULLs
+    let data = build_encoded_f16_bytes_page_iterator::<T>(
+        mandatory_column_desc.clone(),
+        0.0,
+        Encoding::BYTE_STREAM_SPLIT,
+        min,
+        max,
+    );
+    group.bench_function("byte_stream_split encoded, mandatory, no NULLs", |b| {
+        b.iter(|| {
+            let array_reader =
+                create_f16_by_bytes_reader(data.clone(), mandatory_column_desc.clone());
+            count = bench_array_reader(array_reader);
+        });
+        assert_eq!(count, EXPECTED_VALUE_COUNT);
+    });
+
+    let data = build_encoded_f16_bytes_page_iterator::<T>(
+        optional_column_desc.clone(),
+        0.0,
+        Encoding::BYTE_STREAM_SPLIT,
+        min,
+        max,
+    );
+    group.bench_function("byte_stream_split encoded, optional, no NULLs", |b| {
+        b.iter(|| {
+            let array_reader =
+                create_f16_by_bytes_reader(data.clone(), optional_column_desc.clone());
+            count = bench_array_reader(array_reader);
+        });
+        assert_eq!(count, EXPECTED_VALUE_COUNT);
+    });
+
+    let data = build_encoded_f16_bytes_page_iterator::<T>(
+        optional_column_desc.clone(),
+        0.5,
+        Encoding::BYTE_STREAM_SPLIT,
+        min,
+        max,
+    );
+    group.bench_function("byte_stream_split encoded, optional, half NULLs", |b| {
+        b.iter(|| {
+            let array_reader =
+                create_f16_by_bytes_reader(data.clone(), optional_column_desc.clone());
+            count = bench_array_reader(array_reader);
+        });
+        assert_eq!(count, EXPECTED_VALUE_COUNT);
+    });
+}
+
+fn bench_byte_stream_split_decimal<T>(
+    group: &mut BenchmarkGroup<WallTime>,
+    mandatory_column_desc: &ColumnDescPtr,
+    optional_column_desc: &ColumnDescPtr,
+    min: i128,
+    max: i128,
+) where
+    T: parquet::data_type::DataType,
+    T::T: From<Vec<u8>>,
+{
+    let mut count: usize = 0;
+
+    // byte_stream_split encoded, no NULLs
+    let data = build_encoded_decimal_bytes_page_iterator::<T>(
+        mandatory_column_desc.clone(),
+        0.0,
+        Encoding::BYTE_STREAM_SPLIT,
+        min,
+        max,
+    );
+    group.bench_function("byte_stream_split encoded, mandatory, no NULLs", |b| {
+        b.iter(|| {
+            let array_reader =
+                create_decimal_by_bytes_reader(data.clone(), mandatory_column_desc.clone());
+            count = bench_array_reader(array_reader);
+        });
+        assert_eq!(count, EXPECTED_VALUE_COUNT);
+    });
+
+    let data = build_encoded_decimal_bytes_page_iterator::<T>(
+        optional_column_desc.clone(),
+        0.0,
+        Encoding::BYTE_STREAM_SPLIT,
+        min,
+        max,
+    );
+    group.bench_function("byte_stream_split encoded, optional, no NULLs", |b| {
+        b.iter(|| {
+            let array_reader =
+                create_decimal_by_bytes_reader(data.clone(), optional_column_desc.clone());
+            count = bench_array_reader(array_reader);
+        });
+        assert_eq!(count, EXPECTED_VALUE_COUNT);
+    });
+
+    // half null
+    let data = build_encoded_decimal_bytes_page_iterator::<T>(
+        optional_column_desc.clone(),
+        0.5,
+        Encoding::BYTE_STREAM_SPLIT,
+        min,
+        max,
+    );
+    group.bench_function("byte_stream_split encoded, optional, half NULLs", |b| {
+        b.iter(|| {
+            let array_reader =
+                create_decimal_by_bytes_reader(data.clone(), optional_column_desc.clone());
+            count = bench_array_reader(array_reader);
+        });
+        assert_eq!(count, EXPECTED_VALUE_COUNT);
+    });
+}
+
 fn bench_primitive<T>(
     group: &mut BenchmarkGroup<WallTime>,
     mandatory_column_desc: &ColumnDescPtr,
@@ -797,6 +996,35 @@ fn bench_primitive<T>(
     });
 }
 
+fn byte_stream_split_benches(c: &mut Criterion) {
+    let schema = build_test_schema();
+
+    let mut group = c.benchmark_group("arrow_array_reader/BYTE_STREAM_SPLIT/Decimal128Array");
+    let mandatory_decimal4_leaf_desc = schema.column(12);
+    let optional_decimal4_leaf_desc = schema.column(13);
+    bench_byte_stream_split_decimal::<FixedLenByteArrayType>(
+        &mut group,
+        &mandatory_decimal4_leaf_desc,
+        &optional_decimal4_leaf_desc,
+        // precision is 16: the max is 9999999999999999
+        9999999999999000,
+        9999999999999999,
+    );
+    group.finish();
+
+    let mut group = c.benchmark_group("arrow_array_reader/BYTE_STREAM_SPLIT/Float16Array");
+    let mandatory_f16_leaf_desc = schema.column(17);
+    let optional_f16_leaf_desc = schema.column(18);
+    bench_byte_stream_split_f16::<FixedLenByteArrayType>(
+        &mut group,
+        &mandatory_f16_leaf_desc,
+        &optional_f16_leaf_desc,
+        -1.0,
+        1.0,
+    );
+    group.finish();
+}
+
 fn decimal_benches(c: &mut Criterion) {
     let schema = build_test_schema();
     // parquet int32, logical type decimal(8,2)
@@ -1334,5 +1562,10 @@ fn add_benches(c: &mut Criterion) {
     });
 }
 
-criterion_group!(benches, add_benches, decimal_benches,);
+criterion_group!(
+    benches,
+    add_benches,
+    decimal_benches,
+    byte_stream_split_benches,
+);
 criterion_main!(benches);
diff --git a/parquet/benches/encoding.rs b/parquet/benches/encoding.rs
index 80befe8da..bc18a49da 100644
--- a/parquet/benches/encoding.rs
+++ b/parquet/benches/encoding.rs
@@ -16,15 +16,23 @@
 // under the License.
 
 use criterion::*;
+use half::f16;
 use parquet::basic::Encoding;
-use parquet::data_type::{DataType, DoubleType, FloatType};
+use parquet::data_type::{
+    DataType, DoubleType, FixedLenByteArray, FixedLenByteArrayType, FloatType,
+};
 use parquet::decoding::{get_decoder, Decoder};
 use parquet::encoding::get_encoder;
 use parquet::schema::types::{ColumnDescPtr, ColumnDescriptor, ColumnPath, Type};
 use rand::prelude::*;
 use std::sync::Arc;
 
-fn bench_typed<T: DataType>(c: &mut Criterion, values: &[T::T], encoding: Encoding) {
+fn bench_typed<T: DataType>(
+    c: &mut Criterion,
+    values: &[T::T],
+    encoding: Encoding,
+    type_length: i32,
+) {
     let name = format!(
         "dtype={}, encoding={:?}",
         std::any::type_name::<T::T>(),
@@ -33,6 +41,7 @@ fn bench_typed<T: DataType>(c: &mut Criterion, values: &[T::T], encoding: Encodi
     let column_desc_ptr = ColumnDescPtr::new(ColumnDescriptor::new(
         Arc::new(
             Type::primitive_type_builder("", T::get_physical_type())
+                .with_length(type_length)
                 .build()
                 .unwrap(),
         ),
@@ -68,15 +77,25 @@ fn criterion_benchmark(c: &mut Criterion) {
     let mut rng = StdRng::seed_from_u64(0);
     let n = 16 * 1024;
 
+    let mut f16s = Vec::new();
     let mut f32s = Vec::new();
     let mut f64s = Vec::new();
+    let mut d128s = Vec::new();
     for _ in 0..n {
+        f16s.push(FixedLenByteArray::from(
+            f16::from_f32(rng.gen::<f32>()).to_le_bytes().to_vec(),
+        ));
         f32s.push(rng.gen::<f32>());
         f64s.push(rng.gen::<f64>());
+        d128s.push(FixedLenByteArray::from(
+            rng.gen::<i128>().to_be_bytes().to_vec(),
+        ));
     }
 
-    bench_typed::<FloatType>(c, &f32s, Encoding::BYTE_STREAM_SPLIT);
-    bench_typed::<DoubleType>(c, &f64s, Encoding::BYTE_STREAM_SPLIT);
+    bench_typed::<FloatType>(c, &f32s, Encoding::BYTE_STREAM_SPLIT, 0);
+    bench_typed::<DoubleType>(c, &f64s, Encoding::BYTE_STREAM_SPLIT, 0);
+    bench_typed::<FixedLenByteArrayType>(c, &f16s, Encoding::BYTE_STREAM_SPLIT, 2);
+    bench_typed::<FixedLenByteArrayType>(c, &d128s, Encoding::BYTE_STREAM_SPLIT, 16);
 }
 
 criterion_group!(benches, criterion_benchmark);
diff --git a/parquet/src/arrow/arrow_reader/statistics.rs b/parquet/src/arrow/arrow_reader/statistics.rs
index 369ea4a47..d487967c2 100644
--- a/parquet/src/arrow/arrow_reader/statistics.rs
+++ b/parquet/src/arrow/arrow_reader/statistics.rs
@@ -758,7 +758,7 @@ macro_rules! get_data_page_statistics {
     ($stat_type_prefix: ident, $data_type: ident, $iterator: ident) => {
         paste! {
             match $data_type {
-                Some(DataType::Boolean) => {
+                DataType::Boolean => {
                     let iterator = [<$stat_type_prefix BooleanDataPageStatsIterator>]::new($iterator);
                     let mut builder = BooleanBuilder::new();
                     for x in iterator {
@@ -772,7 +772,7 @@ macro_rules! get_data_page_statistics {
                     }
                     Ok(Arc::new(builder.finish()))
                 },
-                Some(DataType::UInt8) => Ok(Arc::new(
+                DataType::UInt8 => Ok(Arc::new(
                     UInt8Array::from_iter(
                         [<$stat_type_prefix Int32DataPageStatsIterator>]::new($iterator)
                             .map(|x| {
@@ -783,7 +783,7 @@ macro_rules! get_data_page_statistics {
                             .flatten()
                     )
                 )),
-                Some(DataType::UInt16) => Ok(Arc::new(
+                DataType::UInt16 => Ok(Arc::new(
                     UInt16Array::from_iter(
                         [<$stat_type_prefix Int32DataPageStatsIterator>]::new($iterator)
                             .map(|x| {
@@ -794,7 +794,7 @@ macro_rules! get_data_page_statistics {
                             .flatten()
                     )
                 )),
-                Some(DataType::UInt32) => Ok(Arc::new(
+                DataType::UInt32 => Ok(Arc::new(
                     UInt32Array::from_iter(
                         [<$stat_type_prefix Int32DataPageStatsIterator>]::new($iterator)
                             .map(|x| {
@@ -804,7 +804,7 @@ macro_rules! get_data_page_statistics {
                             })
                             .flatten()
                 ))),
-                Some(DataType::UInt64) => Ok(Arc::new(
+                DataType::UInt64 => Ok(Arc::new(
                     UInt64Array::from_iter(
                         [<$stat_type_prefix Int64DataPageStatsIterator>]::new($iterator)
                             .map(|x| {
@@ -814,7 +814,7 @@ macro_rules! get_data_page_statistics {
                             })
                             .flatten()
                 ))),
-                Some(DataType::Int8) => Ok(Arc::new(
+                DataType::Int8 => Ok(Arc::new(
                     Int8Array::from_iter(
                         [<$stat_type_prefix Int32DataPageStatsIterator>]::new($iterator)
                             .map(|x| {
@@ -825,7 +825,7 @@ macro_rules! get_data_page_statistics {
                             .flatten()
                     )
                 )),
-                Some(DataType::Int16) => Ok(Arc::new(
+                DataType::Int16 => Ok(Arc::new(
                     Int16Array::from_iter(
                         [<$stat_type_prefix Int32DataPageStatsIterator>]::new($iterator)
                             .map(|x| {
@@ -836,9 +836,9 @@ macro_rules! get_data_page_statistics {
                             .flatten()
                     )
                 )),
-                Some(DataType::Int32) => Ok(Arc::new(Int32Array::from_iter([<$stat_type_prefix Int32DataPageStatsIterator>]::new($iterator).flatten()))),
-                Some(DataType::Int64) => Ok(Arc::new(Int64Array::from_iter([<$stat_type_prefix Int64DataPageStatsIterator>]::new($iterator).flatten()))),
-                Some(DataType::Float16) => Ok(Arc::new(
+                DataType::Int32 => Ok(Arc::new(Int32Array::from_iter([<$stat_type_prefix Int32DataPageStatsIterator>]::new($iterator).flatten()))),
+                DataType::Int64 => Ok(Arc::new(Int64Array::from_iter([<$stat_type_prefix Int64DataPageStatsIterator>]::new($iterator).flatten()))),
+                DataType::Float16 => Ok(Arc::new(
                     Float16Array::from_iter(
                         [<$stat_type_prefix Float16DataPageStatsIterator>]::new($iterator)
                             .map(|x| {
@@ -849,11 +849,11 @@ macro_rules! get_data_page_statistics {
                             .flatten()
                     )
                 )),
-                Some(DataType::Float32) => Ok(Arc::new(Float32Array::from_iter([<$stat_type_prefix Float32DataPageStatsIterator>]::new($iterator).flatten()))),
-                Some(DataType::Float64) => Ok(Arc::new(Float64Array::from_iter([<$stat_type_prefix Float64DataPageStatsIterator>]::new($iterator).flatten()))),
-                Some(DataType::Binary) => Ok(Arc::new(BinaryArray::from_iter([<$stat_type_prefix ByteArrayDataPageStatsIterator>]::new($iterator).flatten()))),
-                Some(DataType::LargeBinary) => Ok(Arc::new(LargeBinaryArray::from_iter([<$stat_type_prefix ByteArrayDataPageStatsIterator>]::new($iterator).flatten()))),
-                Some(DataType::Utf8) => {
+                DataType::Float32 => Ok(Arc::new(Float32Array::from_iter([<$stat_type_prefix Float32DataPageStatsIterator>]::new($iterator).flatten()))),
+                DataType::Float64 => Ok(Arc::new(Float64Array::from_iter([<$stat_type_prefix Float64DataPageStatsIterator>]::new($iterator).flatten()))),
+                DataType::Binary => Ok(Arc::new(BinaryArray::from_iter([<$stat_type_prefix ByteArrayDataPageStatsIterator>]::new($iterator).flatten()))),
+                DataType::LargeBinary => Ok(Arc::new(LargeBinaryArray::from_iter([<$stat_type_prefix ByteArrayDataPageStatsIterator>]::new($iterator).flatten()))),
+                DataType::Utf8 => {
                     let mut builder = StringBuilder::new();
                     let iterator = [<$stat_type_prefix ByteArrayDataPageStatsIterator>]::new($iterator);
                     for x in iterator {
@@ -873,7 +873,7 @@ macro_rules! get_data_page_statistics {
                     }
                     Ok(Arc::new(builder.finish()))
                 },
-                Some(DataType::LargeUtf8) => {
+                DataType::LargeUtf8 => {
                     let mut builder = LargeStringBuilder::new();
                     let iterator = [<$stat_type_prefix ByteArrayDataPageStatsIterator>]::new($iterator);
                     for x in iterator {
@@ -893,10 +893,10 @@ macro_rules! get_data_page_statistics {
                     }
                     Ok(Arc::new(builder.finish()))
                 },
-                Some(DataType::Dictionary(_, value_type)) => {
-                    [<$stat_type_prefix:lower _ page_statistics>](Some(value_type), $iterator)
+                DataType::Dictionary(_, value_type) => {
+                    [<$stat_type_prefix:lower _ page_statistics>](value_type, $iterator)
                 },
-                Some(DataType::Timestamp(unit, timezone)) => {
+                DataType::Timestamp(unit, timezone) => {
                     let iter = [<$stat_type_prefix Int64DataPageStatsIterator>]::new($iterator).flatten();
                     Ok(match unit {
                         TimeUnit::Second => Arc::new(TimestampSecondArray::from_iter(iter).with_timezone_opt(timezone.clone())),
@@ -905,8 +905,8 @@ macro_rules! get_data_page_statistics {
                         TimeUnit::Nanosecond => Arc::new(TimestampNanosecondArray::from_iter(iter).with_timezone_opt(timezone.clone())),
                     })
                 },
-                Some(DataType::Date32) => Ok(Arc::new(Date32Array::from_iter([<$stat_type_prefix Int32DataPageStatsIterator>]::new($iterator).flatten()))),
-                Some(DataType::Date64) => Ok(
+                DataType::Date32 => Ok(Arc::new(Date32Array::from_iter([<$stat_type_prefix Int32DataPageStatsIterator>]::new($iterator).flatten()))),
+                DataType::Date64 => Ok(
                     Arc::new(
                         Date64Array::from_iter([<$stat_type_prefix Int32DataPageStatsIterator>]::new($iterator)
                             .map(|x| {
@@ -919,11 +919,11 @@ macro_rules! get_data_page_statistics {
                         )
                     )
                 ),
-                Some(DataType::Decimal128(precision, scale)) => Ok(Arc::new(
+                DataType::Decimal128(precision, scale) => Ok(Arc::new(
                     Decimal128Array::from_iter([<$stat_type_prefix Decimal128DataPageStatsIterator>]::new($iterator).flatten()).with_precision_and_scale(*precision, *scale)?)),
-                Some(DataType::Decimal256(precision, scale)) => Ok(Arc::new(
+                DataType::Decimal256(precision, scale) => Ok(Arc::new(
                     Decimal256Array::from_iter([<$stat_type_prefix Decimal256DataPageStatsIterator>]::new($iterator).flatten()).with_precision_and_scale(*precision, *scale)?)),
-                Some(DataType::Time32(unit)) => {
+                DataType::Time32(unit) => {
                     Ok(match unit {
                         TimeUnit::Second =>  Arc::new(Time32SecondArray::from_iter(
                             [<$stat_type_prefix Int32DataPageStatsIterator>]::new($iterator).flatten(),
@@ -937,7 +937,7 @@ macro_rules! get_data_page_statistics {
                         }
                     })
                 }
-                Some(DataType::Time64(unit)) => {
+                DataType::Time64(unit) => {
                     Ok(match unit {
                         TimeUnit::Microsecond =>  Arc::new(Time64MicrosecondArray::from_iter(
                             [<$stat_type_prefix Int64DataPageStatsIterator>]::new($iterator).flatten(),
@@ -951,7 +951,7 @@ macro_rules! get_data_page_statistics {
                         }
                     })
                 },
-                Some(DataType::FixedSizeBinary(size)) => {
+                DataType::FixedSizeBinary(size) => {
                     let mut builder = FixedSizeBinaryBuilder::new(*size);
                     let iterator = [<$stat_type_prefix FixedLenByteArrayDataPageStatsIterator>]::new($iterator);
                     for x in iterator {
@@ -964,18 +964,13 @@ macro_rules! get_data_page_statistics {
                             if x.len() == *size as usize {
                                 let _ = builder.append_value(x.data());
                             } else {
-                                // log::debug!(
-                                //     "FixedSizeBinary({}) statistics is a binary of size {}, ignoring it.",
-                                //     size,
-                                //     x.len(),
-                                // );
                                 builder.append_null();
                             }
                         }
                     }
                     Ok(Arc::new(builder.finish()))
                 },
-                Some(DataType::Utf8View) => {
+                DataType::Utf8View => {
                     let mut builder = StringViewBuilder::new();
                     let iterator = [<$stat_type_prefix ByteArrayDataPageStatsIterator>]::new($iterator);
                     for x in iterator {
@@ -995,7 +990,7 @@ macro_rules! get_data_page_statistics {
                     }
                     Ok(Arc::new(builder.finish()))
                 },
-                Some(DataType::BinaryView) => {
+                DataType::BinaryView => {
                     let mut builder = BinaryViewBuilder::new();
                     let iterator = [<$stat_type_prefix ByteArrayDataPageStatsIterator>]::new($iterator);
                     for x in iterator {
@@ -1010,23 +1005,22 @@ macro_rules! get_data_page_statistics {
                     }
                     Ok(Arc::new(builder.finish()))
                 },
-                Some(DataType::Null) |
-                Some(DataType::Duration(_)) |
-                Some(DataType::Interval(_)) |
-                Some(DataType::List(_)) |
-                Some(DataType::ListView(_)) |
-                Some(DataType::FixedSizeList(_, _)) |
-                Some(DataType::LargeList(_)) |
-                Some(DataType::LargeListView(_)) |
-                Some(DataType::Struct(_)) |
-                Some(DataType::Union(_, _)) |
-                Some(DataType::Map(_, _)) |
-                Some(DataType::RunEndEncoded(_, _)) => {
+                DataType::Null |
+                DataType::Duration(_) |
+                DataType::Interval(_) |
+                DataType::List(_) |
+                DataType::ListView(_) |
+                DataType::FixedSizeList(_, _) |
+                DataType::LargeList(_) |
+                DataType::LargeListView(_) |
+                DataType::Struct(_) |
+                DataType::Union(_, _) |
+                DataType::Map(_, _) |
+                DataType::RunEndEncoded(_, _) => {
                     let len = $iterator.count();
                     // don't know how to extract statistics, so return a null array
-                    Ok(new_null_array($data_type.unwrap(), len))
+                    Ok(new_null_array($data_type, len))
                 },
-                None => unimplemented!()  // not sure how to handle this
             }
         }
     }
@@ -1054,10 +1048,7 @@ fn max_statistics<'a, I: Iterator<Item = Option<&'a ParquetStatistics>>>(
 
 /// Extracts the min statistics from an iterator
 /// of parquet page [`Index`]'es to an [`ArrayRef`]
-pub(crate) fn min_page_statistics<'a, I>(
-    data_type: Option<&DataType>,
-    iterator: I,
-) -> Result<ArrayRef>
+pub(crate) fn min_page_statistics<'a, I>(data_type: &DataType, iterator: I) -> Result<ArrayRef>
 where
     I: Iterator<Item = (usize, &'a Index)>,
 {
@@ -1066,10 +1057,7 @@ where
 
 /// Extracts the max statistics from an iterator
 /// of parquet page [`Index`]'es to an [`ArrayRef`]
-pub(crate) fn max_page_statistics<'a, I>(
-    data_type: Option<&DataType>,
-    iterator: I,
-) -> Result<ArrayRef>
+pub(crate) fn max_page_statistics<'a, I>(data_type: &DataType, iterator: I) -> Result<ArrayRef>
 where
     I: Iterator<Item = (usize, &'a Index)>,
 {
@@ -1439,7 +1427,7 @@ impl<'a> StatisticsConverter<'a> {
             (*num_data_pages, column_page_index_per_row_group_per_column)
         });
 
-        min_page_statistics(Some(data_type), iter)
+        min_page_statistics(data_type, iter)
     }
 
     /// Extract the maximum values from Data Page statistics.
@@ -1470,7 +1458,7 @@ impl<'a> StatisticsConverter<'a> {
             (*num_data_pages, column_page_index_per_row_group_per_column)
         });
 
-        max_page_statistics(Some(data_type), iter)
+        max_page_statistics(data_type, iter)
     }
 
     /// Returns a [`UInt64Array`] with null counts for each data page.
diff --git a/parquet/src/basic.rs b/parquet/src/basic.rs
index 880d804a4..a12c9d860 100644
--- a/parquet/src/basic.rs
+++ b/parquet/src/basic.rs
@@ -294,13 +294,14 @@ pub enum Encoding {
     /// The ids are encoded using the RLE encoding.
     RLE_DICTIONARY,
 
-    /// Encoding for floating-point data.
+    /// Encoding for fixed-width data.
     ///
     /// K byte-streams are created where K is the size in bytes of the data type.
-    /// The individual bytes of an FP value are scattered to the corresponding stream and
+    /// The individual bytes of a value are scattered to the corresponding stream and
     /// the streams are concatenated.
     /// This itself does not reduce the size of the data but can lead to better compression
-    /// afterwards.
+    /// afterwards. Note that the use of this encoding with FIXED_LEN_BYTE_ARRAY(N) data may
+    /// perform poorly for large values of N.
     BYTE_STREAM_SPLIT,
 }
 


Test patch:
diff --git a/parquet/src/util/test_common/page_util.rs b/parquet/src/util/test_common/page_util.rs
index 3db43aef0..a1709efa9 100644
--- a/parquet/src/util/test_common/page_util.rs
+++ b/parquet/src/util/test_common/page_util.rs
@@ -51,13 +51,14 @@ pub struct DataPageBuilderImpl {
     rep_levels_byte_len: u32,
     def_levels_byte_len: u32,
     datapage_v2: bool,
+    type_width: i32,
 }
 
 impl DataPageBuilderImpl {
     // `num_values` is the number of non-null values to put in the data page.
     // `datapage_v2` flag is used to indicate if the generated data page should use V2
     // format or not.
-    pub fn new(_desc: ColumnDescPtr, num_values: u32, datapage_v2: bool) -> Self {
+    pub fn new(desc: ColumnDescPtr, num_values: u32, datapage_v2: bool) -> Self {
         DataPageBuilderImpl {
             encoding: None,
             num_values,
@@ -65,6 +66,7 @@ impl DataPageBuilderImpl {
             rep_levels_byte_len: 0,
             def_levels_byte_len: 0,
             datapage_v2,
+            type_width: desc.type_length(),
         }
     }
 
@@ -111,7 +113,7 @@ impl DataPageBuilder for DataPageBuilderImpl {
         // Create test column descriptor.
         let desc = {
             let ty = SchemaType::primitive_type_builder("t", T::get_physical_type())
-                .with_length(0)
+                .with_length(self.type_width)
                 .build()
                 .unwrap();
             Arc::new(ColumnDescriptor::new(