Repository: apache/arrow-rs
Pull request #6205 (2024-08-07 00:47:23)

fix(arrow): restrict the range of temporal values produced via `data_gen`
# Which issue does this PR close?

n/a

# Rationale for this change

Prior to this change, randomly generated temporal values spanned the entire range of their native types. For instance, `TimestampSecondType` generated values in the range of i64::MIN to i64::MAX; however, chrono's `DateTime::<Utc>::MAX_UTC` timestamp is `8210266876799`. This restricts them to the start of unix epoch to +100 years (1970-2070), I chose this range arbitrarily.

At work, we making a translation layer between our internal data types and arrow's. `create_random_array` and `create_random_batch` have been really helpful in terms of generating test data.

# What changes are included in this PR?

Defines a new `RandomTemporalValue` trait within `arrow::util::data_gen` that restricts the range when generating a random value.

# Are there any user-facing changes?

Yes. Docs have been added + no breaking changes.



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-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/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/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 c42f92838..d487967c2 100644
--- a/parquet/src/arrow/arrow_reader/statistics.rs
+++ b/parquet/src/arrow/arrow_reader/statistics.rs
@@ -17,6 +17,8 @@
 
 //! [`StatisticsConverter`] to convert statistics in parquet format to arrow [`ArrayRef`].
 
+/// Notice that all the corresponding tests are in
+/// `arrow-rs/parquet/tests/arrow_reader/statistics.rs`.
 use crate::arrow::buffer::bit_util::sign_extend_be;
 use crate::arrow::parquet_column;
 use crate::data_type::{ByteArray, FixedLenByteArray};
@@ -756,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 {
@@ -770,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| {
@@ -781,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| {
@@ -792,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| {
@@ -802,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| {
@@ -812,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| {
@@ -823,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| {
@@ -834,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| {
@@ -847,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 {
@@ -871,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 {
@@ -891,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())),
@@ -903,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| {
@@ -917,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(),
@@ -935,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(),
@@ -949,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 {
@@ -962,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 {
@@ -993,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 {
@@ -1008,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
             }
         }
     }
@@ -1052,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)>,
 {
@@ -1064,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)>,
 {
@@ -1437,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.
@@ -1468,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.
@@ -1568,1130 +1558,3 @@ impl<'a> StatisticsConverter<'a> {
         new_null_array(data_type, num_row_groups)
     }
 }
-
-#[cfg(test)]
-mod test {
-    use super::*;
-    use crate::arrow::arrow_reader::ArrowReaderBuilder;
-    use crate::arrow::arrow_writer::ArrowWriter;
-    use crate::file::metadata::{ParquetMetaData, RowGroupMetaData};
-    use crate::file::properties::{EnabledStatistics, WriterProperties};
-    use arrow::compute::kernels::cast_utils::Parser;
-    use arrow::datatypes::{i256, Date32Type, Date64Type};
-    use arrow::util::test_util::parquet_test_data;
-    use arrow_array::{
-        new_empty_array, new_null_array, Array, ArrayRef, BinaryArray, BinaryViewArray,
-        BooleanArray, Date32Array, Date64Array, Decimal128Array, Decimal256Array, Float32Array,
-        Float64Array, Int16Array, Int32Array, Int64Array, Int8Array, LargeBinaryArray, RecordBatch,
-        StringArray, StringViewArray, StructArray, TimestampNanosecondArray,
-    };
-    use arrow_schema::{DataType, Field, SchemaRef};
-    use bytes::Bytes;
-    use std::path::PathBuf;
-    use std::sync::Arc;
-    // TODO error cases (with parquet statistics that are mismatched in expected type)
-
-    #[test]
-    fn roundtrip_empty() {
-        let empty_bool_array = new_empty_array(&DataType::Boolean);
-        Test {
-            input: empty_bool_array.clone(),
-            expected_min: empty_bool_array.clone(),
-            expected_max: empty_bool_array.clone(),
-        }
-        .run()
-    }
-
-    #[test]
-    fn roundtrip_bool() {
-        Test {
-            input: bool_array([
-                // row group 1
-                Some(true),
-                None,
-                Some(true),
-                // row group 2
-                Some(true),
-                Some(false),
-                None,
-                // row group 3
-                None,
-                None,
-                None,
-            ]),
-            expected_min: bool_array([Some(true), Some(false), None]),
-            expected_max: bool_array([Some(true), Some(true), None]),
-        }
-        .run()
-    }
-
-    #[test]
-    fn roundtrip_int32() {
-        Test {
-            input: i32_array([
-                // row group 1
-                Some(1),
-                None,
-                Some(3),
-                // row group 2
-                Some(0),
-                Some(5),
-                None,
-                // row group 3
-                None,
-                None,
-                None,
-            ]),
-            expected_min: i32_array([Some(1), Some(0), None]),
-            expected_max: i32_array([Some(3), Some(5), None]),
-        }
-        .run()
-    }
-
-    #[test]
-    fn roundtrip_int64() {
-        Test {
-            input: i64_array([
-                // row group 1
-                Some(1),
-                None,
-                Some(3),
-                // row group 2
-                Some(0),
-                Some(5),
-                None,
-                // row group 3
-                None,
-                None,
-                None,
-            ]),
-            expected_min: i64_array([Some(1), Some(0), None]),
-            expected_max: i64_array(vec![Some(3), Some(5), None]),
-        }
-        .run()
-    }
-
-    #[test]
-    fn roundtrip_f32() {
-        Test {
-            input: f32_array([
-                // row group 1
-                Some(1.0),
-                None,
-                Some(3.0),
-                // row group 2
-                Some(-1.0),
-                Some(5.0),
-                None,
-                // row group 3
-                None,
-                None,
-                None,
-            ]),
-            expected_min: f32_array([Some(1.0), Some(-1.0), None]),
-            expected_max: f32_array([Some(3.0), Some(5.0), None]),
-        }
-        .run()
-    }
-
-    #[test]
-    fn roundtrip_f64() {
-        Test {
-            input: f64_array([
-                // row group 1
-                Some(1.0),
-                None,
-                Some(3.0),
-                // row group 2
-                Some(-1.0),
-                Some(5.0),
-                None,
-                // row group 3
-                None,
-                None,
-                None,
-            ]),
-            expected_min: f64_array([Some(1.0), Some(-1.0), None]),
-            expected_max: f64_array([Some(3.0), Some(5.0), None]),
-        }
-        .run()
-    }
-
-    #[test]
-    fn roundtrip_timestamp() {
-        Test {
-            input: timestamp_seconds_array(
-                [
-                    // row group 1
-                    Some(1),
-                    None,
-                    Some(3),
-                    // row group 2
-                    Some(9),
-                    Some(5),
-                    None,
-                    // row group 3
-                    None,
-                    None,
-                    None,
-                ],
-                None,
-            ),
-            expected_min: timestamp_seconds_array([Some(1), Some(5), None], None),
-            expected_max: timestamp_seconds_array([Some(3), Some(9), None], None),
-        }
-        .run();
-
-        Test {
-            input: timestamp_milliseconds_array(
-                [
-                    // row group 1
-                    Some(1),
-                    None,
-                    Some(3),
-                    // row group 2
-                    Some(9),
-                    Some(5),
-                    None,
-                    // row group 3
-                    None,
-                    None,
-                    None,
-                ],
-                None,
-            ),
-            expected_min: timestamp_milliseconds_array([Some(1), Some(5), None], None),
-            expected_max: timestamp_milliseconds_array([Some(3), Some(9), None], None),
-        }
-        .run();
-
-        Test {
-            input: timestamp_microseconds_array(
-                [
-                    // row group 1
-                    Some(1),
-                    None,
-                    Some(3),
-                    // row group 2
-                    Some(9),
-                    Some(5),
-                    None,
-                    // row group 3
-                    None,
-                    None,
-                    None,
-                ],
-                None,
-            ),
-            expected_min: timestamp_microseconds_array([Some(1), Some(5), None], None),
-            expected_max: timestamp_microseconds_array([Some(3), Some(9), None], None),
-        }
-        .run();
-
-        Test {
-            input: timestamp_nanoseconds_array(
-                [
-                    // row group 1
-                    Some(1),
-                    None,
-                    Some(3),
-                    // row group 2
-                    Some(9),
-                    Some(5),
-                    None,
-                    // row group 3
-                    None,
-                    None,
-                    None,
-                ],
-                None,
-            ),
-            expected_min: timestamp_nanoseconds_array([Some(1), Some(5), None], None),
-            expected_max: timestamp_nanoseconds_array([Some(3), Some(9), None], None),
-        }
-        .run()
-    }
-
-    #[test]
-    fn roundtrip_timestamp_timezoned() {
-        Test {
-            input: timestamp_seconds_array(
-                [
-                    // row group 1
-                    Some(1),
-                    None,
-                    Some(3),
-                    // row group 2
-                    Some(9),
-                    Some(5),
-                    None,
-                    // row group 3
-                    None,
-                    None,
-                    None,
-                ],
-                Some("UTC"),
-            ),
-            expected_min: timestamp_seconds_array([Some(1), Some(5), None], Some("UTC")),
-            expected_max: timestamp_seconds_array([Some(3), Some(9), None], Some("UTC")),
-        }
-        .run();
-
-        Test {
-            input: timestamp_milliseconds_array(
-                [
-                    // row group 1
-                    Some(1),
-                    None,
-                    Some(3),
-                    // row group 2
-                    Some(9),
-                    Some(5),
-                    None,
-                    // row group 3
-                    None,
-                    None,
-                    None,
-                ],
-                Some("UTC"),
-            ),
-            expected_min: timestamp_milliseconds_array([Some(1), Some(5), None], Some("UTC")),
-            expected_max: timestamp_milliseconds_array([Some(3), Some(9), None], Some("UTC")),
-        }
-        .run();
-
-        Test {
-            input: timestamp_microseconds_array(
-                [
-                    // row group 1
-                    Some(1),
-                    None,
-                    Some(3),
-                    // row group 2
-                    Some(9),
-                    Some(5),
-                    None,
-                    // row group 3
-                    None,
-                    None,
-                    None,
-                ],
-                Some("UTC"),
-            ),
-            expected_min: timestamp_microseconds_array([Some(1), Some(5), None], Some("UTC")),
-            expected_max: timestamp_microseconds_array([Some(3), Some(9), None], Some("UTC")),
-        }
-        .run();
-
-        Test {
-            input: timestamp_nanoseconds_array(
-                [
-                    // row group 1
-                    Some(1),
-                    None,
-                    Some(3),
-                    // row group 2
-                    Some(9),
-                    Some(5),
-                    None,
-                    // row group 3
-                    None,
-                    None,
-                    None,
-                ],
-                Some("UTC"),
-            ),
-            expected_min: timestamp_nanoseconds_array([Some(1), Some(5), None], Some("UTC")),
-            expected_max: timestamp_nanoseconds_array([Some(3), Some(9), None], Some("UTC")),
-        }
-        .run()
-    }
-
-    #[test]
-    fn roundtrip_decimal() {
-        Test {
-            input: Arc::new(
-                Decimal128Array::from(vec![
-                    // row group 1
-                    Some(100),
-                    None,
-                    Some(22000),
-                    // row group 2
-                    Some(500000),
-                    Some(330000),
-                    None,
-                    // row group 3
-                    None,
-                    None,
-                    None,
-                ])
-                .with_precision_and_scale(9, 2)
-                .unwrap(),
-            ),
-            expected_min: Arc::new(
-                Decimal128Array::from(vec![Some(100), Some(330000), None])
-                    .with_precision_and_scale(9, 2)
-                    .unwrap(),
-            ),
-            expected_max: Arc::new(
-                Decimal128Array::from(vec![Some(22000), Some(500000), None])
-                    .with_precision_and_scale(9, 2)
-                    .unwrap(),
-            ),
-        }
-        .run();
-
-        Test {
-            input: Arc::new(
-                Decimal256Array::from(vec![
-                    // row group 1
-                    Some(i256::from(100)),
-                    None,
-                    Some(i256::from(22000)),
-                    // row group 2
-                    Some(i256::MAX),
-                    Some(i256::MIN),
-                    None,
-                    // row group 3
-                    None,
-                    None,
-                    None,
-                ])
-                .with_precision_and_scale(76, 76)
-                .unwrap(),
-            ),
-            expected_min: Arc::new(
-                Decimal256Array::from(vec![Some(i256::from(100)), Some(i256::MIN), None])
-                    .with_precision_and_scale(76, 76)
-                    .unwrap(),
-            ),
-            expected_max: Arc::new(
-                Decimal256Array::from(vec![Some(i256::from(22000)), Some(i256::MAX), None])
-                    .with_precision_and_scale(76, 76)
-                    .unwrap(),
-            ),
-        }
-        .run()
-    }
-
-    #[test]
-    fn roundtrip_utf8() {
-        Test {
-            input: utf8_array([
-                // row group 1
-                Some("A"),
-                None,
-                Some("Q"),
-                // row group 2
-                Some("ZZ"),
-                Some("AA"),
-                None,
-                // row group 3
-                None,
-                None,
-                None,
-            ]),
-            expected_min: utf8_array([Some("A"), Some("AA"), None]),
-            expected_max: utf8_array([Some("Q"), Some("ZZ"), None]),
-        }
-        .run()
-    }
-
-    #[test]
-    fn roundtrip_string_view() {
-        Test {
-            input: string_view_array([
-                // row group 1
-                Some("A"),
-                None,
-                Some("Q"),
-                // row group 2
-                Some("ZZ"),
-                Some("A_longerthan12"),
-                None,
-                // row group 3
-                Some("A_longerthan12"),
-                None,
-                None,
-            ]),
-            expected_min: string_view_array([
-                Some("A"),
-                Some("A_longerthan12"),
-                Some("A_longerthan12"),
-            ]),
-            expected_max: string_view_array([Some("Q"), Some("ZZ"), Some("A_longerthan12")]),
-        }
-        .run()
-    }
-
-    #[test]
-    fn roundtrip_binary_view() {
-        let input: Vec<Option<&[u8]>> = vec![
-            // row group 1
-            Some(b"A"),
-            None,
-            Some(b"Q"),
-            // row group 2
-            Some(b"ZZ"),
-            Some(b"A_longerthan12"),
-            None,
-            // row group 3
-            Some(b"A_longerthan12"),
-            None,
-            None,
-        ];
-
-        let expected_min: Vec<Option<&[u8]>> =
-            vec![Some(b"A"), Some(b"A_longerthan12"), Some(b"A_longerthan12")];
-        let expected_max: Vec<Option<&[u8]>> =
-            vec![Some(b"Q"), Some(b"ZZ"), Some(b"A_longerthan12")];
-
-        let array = binary_view_array(input);
-
-        Test {
-            input: array,
-            expected_min: binary_view_array(expected_min),
-            expected_max: binary_view_array(expected_max),
-        }
-        .run()
-    }
-
-    #[test]
-    fn roundtrip_struct() {
-        let mut test = Test {
-            input: struct_array(vec![
-                // row group 1
-                (Some(true), Some(1)),
-                (None, None),
-                (Some(true), Some(3)),
-                // row group 2
-                (Some(true), Some(0)),
-                (Some(false), Some(5)),
-                (None, None),
-                // row group 3
-                (None, None),
-                (None, None),
-                (None, None),
-            ]),
-            expected_min: struct_array(vec![
-                (Some(true), Some(1)),
-                (Some(true), Some(0)),
-                (None, None),
-            ]),
-
-            expected_max: struct_array(vec![
-                (Some(true), Some(3)),
-                (Some(true), Some(0)),
-                (None, None),
-            ]),
-        };
-        // Due to https://github.com/apache/datafusion/issues/8334,
-        // statistics for struct arrays are not supported
-        test.expected_min = new_null_array(test.input.data_type(), test.expected_min.len());
-        test.expected_max = new_null_array(test.input.data_type(), test.expected_min.len());
-        test.run()
-    }
-
-    #[test]
-    fn roundtrip_binary() {
-        Test {
-            input: Arc::new(BinaryArray::from_opt_vec(vec![
-                // row group 1
-                Some(b"A"),
-                None,
-                Some(b"Q"),
-                // row group 2
-                Some(b"ZZ"),
-                Some(b"AA"),
-                None,
-                // row group 3
-                None,
-                None,
-                None,
-            ])),
-            expected_min: Arc::new(BinaryArray::from_opt_vec(vec![
-                Some(b"A"),
-                Some(b"AA"),
-                None,
-            ])),
-            expected_max: Arc::new(BinaryArray::from_opt_vec(vec![
-                Some(b"Q"),
-                Some(b"ZZ"),
-                None,
-            ])),
-        }
-        .run()
-    }
-
-    #[test]
-    fn roundtrip_date32() {
-        Test {
-            input: date32_array(vec![
-                // row group 1
-                Some("2021-01-01"),
-                None,
-                Some("2021-01-03"),
-                // row group 2
-                Some("2021-01-01"),
-                Some("2021-01-05"),
-                None,
-                // row group 3
-                None,
-                None,
-                None,
-            ]),
-            expected_min: date32_array(vec![Some("2021-01-01"), Some("2021-01-01"), None]),
-            expected_max: date32_array(vec![Some("2021-01-03"), Some("2021-01-05"), None]),
-        }
-        .run()
-    }
-
-    #[test]
-    fn roundtrip_date64() {
-        Test {
-            input: date64_array(vec![
-                // row group 1
-                Some("2021-01-01"),
-                None,
-                Some("2021-01-03"),
-                // row group 2
-                Some("2021-01-01"),
-                Some("2021-01-05"),
-                None,
-                // row group 3
-                None,
-                None,
-                None,
-            ]),
-            expected_min: date64_array(vec![Some("2021-01-01"), Some("2021-01-01"), None]),
-            expected_max: date64_array(vec![Some("2021-01-03"), Some("2021-01-05"), None]),
-        }
-        .run()
-    }
-
-    #[test]
-    fn roundtrip_large_binary_array() {
-        let input: Vec<Option<&[u8]>> = vec![
-            // row group 1
-            Some(b"A"),
-            None,
-            Some(b"Q"),
-            // row group 2
-            Some(b"ZZ"),
-            Some(b"AA"),
-            None,
-            // row group 3
-            None,
-            None,
-            None,
-        ];
-
-        let expected_min: Vec<Option<&[u8]>> = vec![Some(b"A"), Some(b"AA"), None];
-        let expected_max: Vec<Option<&[u8]>> = vec![Some(b"Q"), Some(b"ZZ"), None];
-
-        Test {
-            input: large_binary_array(input),
-            expected_min: large_binary_array(expected_min),
-            expected_max: large_binary_array(expected_max),
-        }
-        .run();
-    }
-
-    #[test]
-    fn struct_and_non_struct() {
-        // Ensures that statistics for an array that appears *after* a struct
-        // array are not wrong
-        let struct_col = struct_array(vec![
-            // row group 1
-            (Some(true), Some(1)),
-            (None, None),
-            (Some(true), Some(3)),
-        ]);
-        let int_col = i32_array([Some(100), Some(200), Some(300)]);
-        let expected_min = i32_array([Some(100)]);
-        let expected_max = i32_array(vec![Some(300)]);
-
-        // use a name that shadows a name in the struct column
-        match struct_col.data_type() {
-            DataType::Struct(fields) => {
-                assert_eq!(fields.get(1).unwrap().name(), "int_col")
-            }
-            _ => panic!("unexpected data type for struct column"),
-        };
-
-        let input_batch =
-            RecordBatch::try_from_iter([("struct_col", struct_col), ("int_col", int_col)]).unwrap();
-
-        let schema = input_batch.schema();
-
-        let metadata = parquet_metadata(schema.clone(), input_batch);
-        let parquet_schema = metadata.file_metadata().schema_descr();
-
-        // read the int_col statistics
-        let (idx, _) = parquet_column(parquet_schema, &schema, "int_col").unwrap();
-        assert_eq!(idx, 2);
-
-        let row_groups = metadata.row_groups();
-        let converter = StatisticsConverter::try_new("int_col", &schema, parquet_schema).unwrap();
-
-        let min = converter.row_group_mins(row_groups.iter()).unwrap();
-        assert_eq!(
-            &min,
-            &expected_min,
-            "Min. Statistics\n\n{}\n\n",
-            DisplayStats(row_groups)
-        );
-
-        let max = converter.row_group_maxes(row_groups.iter()).unwrap();
-        assert_eq!(
-            &max,
-            &expected_max,
-            "Max. Statistics\n\n{}\n\n",
-            DisplayStats(row_groups)
-        );
-    }
-
-    #[test]
-    fn nan_in_stats() {
-        // /parquet-testing/data/nan_in_stats.parquet
-        // row_groups: 1
-        // "x": Double({min: Some(1.0), max: Some(NaN), distinct_count: None, null_count: 0, min_max_deprecated: false, min_max_backwards_compatible: false})
-
-        TestFile::new("nan_in_stats.parquet")
-            .with_column(ExpectedColumn {
-                name: "x",
-                expected_min: Arc::new(Float64Array::from(vec![Some(1.0)])),
-                expected_max: Arc::new(Float64Array::from(vec![Some(f64::NAN)])),
-            })
-            .run();
-    }
-
-    #[test]
-    fn alltypes_plain() {
-        // /parquet-testing/data/datapage_v1-snappy-compressed-checksum.parquet
-        // row_groups: 1
-        // (has no statistics)
-        TestFile::new("alltypes_plain.parquet")
-            // No column statistics should be read as NULL, but with the right type
-            .with_column(ExpectedColumn {
-                name: "id",
-                expected_min: i32_array([None]),
-                expected_max: i32_array([None]),
-            })
-            .with_column(ExpectedColumn {
-                name: "bool_col",
-                expected_min: bool_array([None]),
-                expected_max: bool_array([None]),
-            })
-            .run();
-    }
-
-    #[test]
-    fn alltypes_tiny_pages() {
-        // /parquet-testing/data/alltypes_tiny_pages.parquet
-        // row_groups: 1
-        // "id": Int32({min: Some(0), max: Some(7299), distinct_count: None, null_count: 0, min_max_deprecated: false, min_max_backwards_compatible: false})
-        // "bool_col": Boolean({min: Some(false), max: Some(true), distinct_count: None, null_count: 0, min_max_deprecated: false, min_max_backwards_compatible: false})
-        // "tinyint_col": Int32({min: Some(0), max: Some(9), distinct_count: None, null_count: 0, min_max_deprecated: false, min_max_backwards_compatible: false})
-        // "smallint_col": Int32({min: Some(0), max: Some(9), distinct_count: None, null_count: 0, min_max_deprecated: false, min_max_backwards_compatible: false})
-        // "int_col": Int32({min: Some(0), max: Some(9), distinct_count: None, null_count: 0, min_max_deprecated: false, min_max_backwards_compatible: false})
-        // "bigint_col": Int64({min: Some(0), max: Some(90), distinct_count: None, null_count: 0, min_max_deprecated: false, min_max_backwards_compatible: false})
-        // "float_col": Float({min: Some(0.0), max: Some(9.9), distinct_count: None, null_count: 0, min_max_deprecated: false, min_max_backwards_compatible: false})
-        // "double_col": Double({min: Some(0.0), max: Some(90.89999999999999), distinct_count: None, null_count: 0, min_max_deprecated: false, min_max_backwards_compatible: false})
-        // "date_string_col": ByteArray({min: Some(ByteArray { data: "01/01/09" }), max: Some(ByteArray { data: "12/31/10" }), distinct_count: None, null_count: 0, min_max_deprecated: false, min_max_backwards_compatible: false})
-        // "string_col": ByteArray({min: Some(ByteArray { data: "0" }), max: Some(ByteArray { data: "9" }), distinct_count: None, null_count: 0, min_max_deprecated: false, min_max_backwards_compatible: false})
-        // "timestamp_col": Int96({min: None, max: None, distinct_count: None, null_count: 0, min_max_deprecated: true, min_max_backwards_compatible: true})
-        // "year": Int32({min: Some(2009), max: Some(2010), distinct_count: None, null_count: 0, min_max_deprecated: false, min_max_backwards_compatible: false})
-        // "month": Int32({min: Some(1), max: Some(12), distinct_count: None, null_count: 0, min_max_deprecated: false, min_max_backwards_compatible: false})
-        TestFile::new("alltypes_tiny_pages.parquet")
-            .with_column(ExpectedColumn {
-                name: "id",
-                expected_min: i32_array([Some(0)]),
-                expected_max: i32_array([Some(7299)]),
-            })
-            .with_column(ExpectedColumn {
-                name: "bool_col",
-                expected_min: bool_array([Some(false)]),
-                expected_max: bool_array([Some(true)]),
-            })
-            .with_column(ExpectedColumn {
-                name: "tinyint_col",
-                expected_min: i8_array([Some(0)]),
-                expected_max: i8_array([Some(9)]),
-            })
-            .with_column(ExpectedColumn {
-                name: "smallint_col",
-                expected_min: i16_array([Some(0)]),
-                expected_max: i16_array([Some(9)]),
-            })
-            .with_column(ExpectedColumn {
-                name: "int_col",
-                expected_min: i32_array([Some(0)]),
-                expected_max: i32_array([Some(9)]),
-            })
-            .with_column(ExpectedColumn {
-                name: "bigint_col",
-                expected_min: i64_array([Some(0)]),
-                expected_max: i64_array([Some(90)]),
-            })
-            .with_column(ExpectedColumn {
-                name: "float_col",
-                expected_min: f32_array([Some(0.0)]),
-                expected_max: f32_array([Some(9.9)]),
-            })
-            .with_column(ExpectedColumn {
-                name: "double_col",
-                expected_min: f64_array([Some(0.0)]),
-                expected_max: f64_array([Some(90.89999999999999)]),
-            })
-            .with_column(ExpectedColumn {
-                name: "date_string_col",
-                expected_min: utf8_array([Some("01/01/09")]),
-                expected_max: utf8_array([Some("12/31/10")]),
-            })
-            .with_column(ExpectedColumn {
-                name: "string_col",
-                expected_min: utf8_array([Some("0")]),
-                expected_max: utf8_array([Some("9")]),
-            })
-            // File has no min/max for timestamp_col
-            .with_column(ExpectedColumn {
-                name: "timestamp_col",
-                expected_min: timestamp_nanoseconds_array([None], None),
-                expected_max: timestamp_nanoseconds_array([None], None),
-            })
-            .with_column(ExpectedColumn {
-                name: "year",
-                expected_min: i32_array([Some(2009)]),
-                expected_max: i32_array([Some(2010)]),
-            })
-            .with_column(ExpectedColumn {
-                name: "month",
-                expected_min: i32_array([Some(1)]),
-                expected_max: i32_array([Some(12)]),
-            })
-            .run();
-    }
-
-    #[test]
-    fn fixed_length_decimal_legacy() {
-        // /parquet-testing/data/fixed_length_decimal_legacy.parquet
-        // row_groups: 1
-        // "value": FixedLenByteArray({min: Some(FixedLenByteArray(ByteArray { data: Some(ByteBufferPtr { data: b"\0\0\0\0\0\xc8" }) })), max: Some(FixedLenByteArray(ByteArray { data: "\0\0\0\0\t`" })), distinct_count: None, null_count: 0, min_max_deprecated: true, min_max_backwards_compatible: true})
-
-        TestFile::new("fixed_length_decimal_legacy.parquet")
-            .with_column(ExpectedColumn {
-                name: "value",
-                expected_min: Arc::new(
-                    Decimal128Array::from(vec![Some(200)])
-                        .with_precision_and_scale(13, 2)
-                        .unwrap(),
-                ),
-                expected_max: Arc::new(
-                    Decimal128Array::from(vec![Some(2400)])
-                        .with_precision_and_scale(13, 2)
-                        .unwrap(),
-                ),
-            })
-            .run();
-    }
-
-    const ROWS_PER_ROW_GROUP: usize = 3;
-
-    /// Writes the input batch into a parquet file, with every every three rows as
-    /// their own row group, and compares the min/maxes to the expected values
-    struct Test {
-        input: ArrayRef,
-        expected_min: ArrayRef,
-        expected_max: ArrayRef,
-    }
-
-    impl Test {
-        fn run(self) {
-            let Self {
-                input,
-                expected_min,
-                expected_max,
-            } = self;
-
-            let input_batch = RecordBatch::try_from_iter([("c1", input)]).unwrap();
-
-            let schema = input_batch.schema();
-
-            let metadata = parquet_metadata(schema.clone(), input_batch);
-            let parquet_schema = metadata.file_metadata().schema_descr();
-
-            let row_groups = metadata.row_groups();
-
-            for field in schema.fields() {
-                if field.data_type().is_nested() {
-                    let lookup = parquet_column(parquet_schema, &schema, field.name());
-                    assert_eq!(lookup, None);
-                    continue;
-                }
-
-                let converter =
-                    StatisticsConverter::try_new(field.name(), &schema, parquet_schema).unwrap();
-
-                assert_eq!(converter.arrow_field, field.as_ref());
-
-                let mins = converter.row_group_mins(row_groups.iter()).unwrap();
-                assert_eq!(
-                    &mins,
-                    &expected_min,
-                    "Min. Statistics\n\n{}\n\n",
-                    DisplayStats(row_groups)
-                );
-
-                let maxes = converter.row_group_maxes(row_groups.iter()).unwrap();
-                assert_eq!(
-                    &maxes,
-                    &expected_max,
-                    "Max. Statistics\n\n{}\n\n",
-                    DisplayStats(row_groups)
-                );
-            }
-        }
-    }
-
-    /// Write the specified batches out as parquet and return the metadata
-    fn parquet_metadata(schema: SchemaRef, batch: RecordBatch) -> Arc<ParquetMetaData> {
-        let props = WriterProperties::builder()
-            .set_statistics_enabled(EnabledStatistics::Chunk)
-            .set_max_row_group_size(ROWS_PER_ROW_GROUP)
-            .build();
-
-        let mut buffer = Vec::new();
-        let mut writer = ArrowWriter::try_new(&mut buffer, schema, Some(props)).unwrap();
-        writer.write(&batch).unwrap();
-        writer.close().unwrap();
-
-        let reader = ArrowReaderBuilder::try_new(Bytes::from(buffer)).unwrap();
-        reader.metadata().clone()
-    }
-
-    /// Formats the statistics nicely for display
-    struct DisplayStats<'a>(&'a [RowGroupMetaData]);
-    impl<'a> std::fmt::Display for DisplayStats<'a> {
-        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-            let row_groups = self.0;
-            writeln!(f, "  row_groups: {}", row_groups.len())?;
-            for rg in row_groups {
-                for col in rg.columns() {
-                    if let Some(statistics) = col.statistics() {
-                        writeln!(f, "   {}: {:?}", col.column_path(), statistics)?;
-                    }
-                }
-            }
-            Ok(())
-        }
-    }
-
-    struct ExpectedColumn {
-        name: &'static str,
-        expected_min: ArrayRef,
-        expected_max: ArrayRef,
-    }
-
-    /// Reads statistics out of the specified, and compares them to the expected values
-    struct TestFile {
-        file_name: &'static str,
-        expected_columns: Vec<ExpectedColumn>,
-    }
-
-    impl TestFile {
-        fn new(file_name: &'static str) -> Self {
-            Self {
-                file_name,
-                expected_columns: Vec::new(),
-            }
-        }
-
-        fn with_column(mut self, column: ExpectedColumn) -> Self {
-            self.expected_columns.push(column);
-            self
-        }
-
-        /// Reads the specified parquet file and validates that the expected min/max
-        /// values for the specified columns are as expected.
-        fn run(self) {
-            let path = PathBuf::from(parquet_test_data()).join(self.file_name);
-            let file = std::fs::File::open(path).unwrap();
-            let reader = ArrowReaderBuilder::try_new(file).unwrap();
-            let arrow_schema = reader.schema();
-            let metadata = reader.metadata();
-            let row_groups = metadata.row_groups();
-            let parquet_schema = metadata.file_metadata().schema_descr();
-
-            for expected_column in self.expected_columns {
-                let ExpectedColumn {
-                    name,
-                    expected_min,
-                    expected_max,
-                } = expected_column;
-
-                let converter =
-                    StatisticsConverter::try_new(name, arrow_schema, parquet_schema).unwrap();
-
-                // test accessors on the converter
-                let parquet_column_index =
-                    parquet_column(parquet_schema, arrow_schema, name).map(|(idx, _field)| idx);
-                assert_eq!(converter.parquet_column_index(), parquet_column_index);
-                assert_eq!(converter.arrow_field().name(), name);
-
-                let actual_min = converter.row_group_mins(row_groups.iter()).unwrap();
-                assert_eq!(&expected_min, &actual_min, "column {name}");
-
-                let actual_max = converter.row_group_maxes(row_groups.iter()).unwrap();
-                assert_eq!(&expected_max, &actual_max, "column {name}");
-            }
-        }
-    }
-
-    fn bool_array(input: impl IntoIterator<Item = Option<bool>>) -> ArrayRef {
-        let array: BooleanArray = input.into_iter().collect();
-        Arc::new(array)
-    }
-
-    fn i8_array(input: impl IntoIterator<Item = Option<i8>>) -> ArrayRef {
-        let array: Int8Array = input.into_iter().collect();
-        Arc::new(array)
-    }
-
-    fn i16_array(input: impl IntoIterator<Item = Option<i16>>) -> ArrayRef {
-        let array: Int16Array = input.into_iter().collect();
-        Arc::new(array)
-    }
-
-    fn i32_array(input: impl IntoIterator<Item = Option<i32>>) -> ArrayRef {
-        let array: Int32Array = input.into_iter().collect();
-        Arc::new(array)
-    }
-
-    fn i64_array(input: impl IntoIterator<Item = Option<i64>>) -> ArrayRef {
-        let array: Int64Array = input.into_iter().collect();
-        Arc::new(array)
-    }
-
-    fn f32_array(input: impl IntoIterator<Item = Option<f32>>) -> ArrayRef {
-        let array: Float32Array = input.into_iter().collect();
-        Arc::new(array)
-    }
-
-    fn f64_array(input: impl IntoIterator<Item = Option<f64>>) -> ArrayRef {
-        let array: Float64Array = input.into_iter().collect();
-        Arc::new(array)
-    }
-
-    fn timestamp_seconds_array(
-        input: impl IntoIterator<Item = Option<i64>>,
-        timzezone: Option<&str>,
-    ) -> ArrayRef {
-        let array: TimestampSecondArray = input.into_iter().collect();
-        match timzezone {
-            Some(tz) => Arc::new(array.with_timezone(tz)),
-            None => Arc::new(array),
-        }
-    }
-
-    fn timestamp_milliseconds_array(
-        input: impl IntoIterator<Item = Option<i64>>,
-        timzezone: Option<&str>,
-    ) -> ArrayRef {
-        let array: TimestampMillisecondArray = input.into_iter().collect();
-        match timzezone {
-            Some(tz) => Arc::new(array.with_timezone(tz)),
-            None => Arc::new(array),
-        }
-    }
-
-    fn timestamp_microseconds_array(
-        input: impl IntoIterator<Item = Option<i64>>,
-        timzezone: Option<&str>,
-    ) -> ArrayRef {
-        let array: TimestampMicrosecondArray = input.into_iter().collect();
-        match timzezone {
-            Some(tz) => Arc::new(array.with_timezone(tz)),
-            None => Arc::new(array),
-        }
-    }
-
-    fn timestamp_nanoseconds_array(
-        input: impl IntoIterator<Item = Option<i64>>,
-        timzezone: Option<&str>,
-    ) -> ArrayRef {
-        let array: TimestampNanosecondArray = input.into_iter().collect();
-        match timzezone {
-            Some(tz) => Arc::new(array.with_timezone(tz)),
-            None => Arc::new(array),
-        }
-    }
-
-    fn utf8_array<'a>(input: impl IntoIterator<Item = Option<&'a str>>) -> ArrayRef {
-        let array: StringArray = input
-            .into_iter()
-            .map(|s| s.map(|s| s.to_string()))
-            .collect();
-        Arc::new(array)
-    }
-
-    // returns a struct array with columns "bool_col" and "int_col" with the specified values
-    fn struct_array(input: Vec<(Option<bool>, Option<i32>)>) -> ArrayRef {
-        let boolean: BooleanArray = input.iter().map(|(b, _i)| b).collect();
-        let int: Int32Array = input.iter().map(|(_b, i)| i).collect();
-
-        let nullable = true;
-        let struct_array = StructArray::from(vec![
-            (
-                Arc::new(Field::new("bool_col", DataType::Boolean, nullable)),
-                Arc::new(boolean) as ArrayRef,
-            ),
-            (
-                Arc::new(Field::new("int_col", DataType::Int32, nullable)),
-                Arc::new(int) as ArrayRef,
-            ),
-        ]);
-        Arc::new(struct_array)
-    }
-
-    fn date32_array<'a>(input: impl IntoIterator<Item = Option<&'a str>>) -> ArrayRef {
-        let array = Date32Array::from(
-            input
-                .into_iter()
-                .map(|s| Date32Type::parse(s.unwrap_or_default()))
-                .collect::<Vec<_>>(),
-        );
-        Arc::new(array)
-    }
-
-    fn date64_array<'a>(input: impl IntoIterator<Item = Option<&'a str>>) -> ArrayRef {
-        let array = Date64Array::from(
-            input
-                .into_iter()
-                .map(|s| Date64Type::parse(s.unwrap_or_default()))
-                .collect::<Vec<_>>(),
-        );
-        Arc::new(array)
-    }
-
-    fn large_binary_array<'a>(input: impl IntoIterator<Item = Option<&'a [u8]>>) -> ArrayRef {
-        let array = LargeBinaryArray::from(input.into_iter().collect::<Vec<Option<&[u8]>>>());
-
-        Arc::new(array)
-    }
-
-    fn string_view_array<'a>(input: impl IntoIterator<Item = Option<&'a str>>) -> ArrayRef {
-        let array: StringViewArray = input
-            .into_iter()
-            .map(|s| s.map(|s| s.to_string()))
-            .collect();
-
-        Arc::new(array)
-    }
-
-    fn binary_view_array(input: Vec<Option<&[u8]>>) -> ArrayRef {
-        let array = BinaryViewArray::from(input.into_iter().collect::<Vec<Option<&[u8]>>>());
-
-        Arc::new(array)
-    }
-}
diff --git a/parquet/src/schema/types.rs b/parquet/src/schema/types.rs
index 0eb08f557..190374fd8 100644
--- a/parquet/src/schema/types.rs
+++ b/parquet/src/schema/types.rs
@@ -188,7 +188,7 @@ impl Type {
     }
 
     /// Returns `true` if this type is repeated or optional.
-    /// If this type doesn't have repetition defined, we still treat it as optional.
+    /// If this type doesn't have repetition defined, we treat it as required.
     pub fn is_optional(&self) -> bool {
         self.get_basic_info().has_repetition()
             && self.get_basic_info().repetition() != Repetition::REQUIRED
@@ -1932,6 +1932,7 @@ mod tests {
         let f1 = Type::group_type_builder("f").build().unwrap();
         let f2 = Type::group_type_builder("f").build().unwrap();
         assert!(f1.check_contains(&f2));
+        assert!(!f1.is_optional());
 
         // OK: fields match
         let f1 = test_new_group_type(


Test patch:
diff --git a/parquet-testing b/parquet-testing
index 1ba34478f..9b48ff4f9 160000
--- a/parquet-testing
+++ b/parquet-testing
@@ -1,1 +1,1 @@
-Subproject commit 1ba34478f535c89382263c42c675a9af4f57f2dd
+Subproject commit 9b48ff4f94dc5e89592d46a119884dbb88100884
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(
diff --git a/parquet/tests/arrow_reader/statistics.rs b/parquet/tests/arrow_reader/statistics.rs
index 75a73ac13..384be83d3 100644
--- a/parquet/tests/arrow_reader/statistics.rs
+++ b/parquet/tests/arrow_reader/statistics.rs
@@ -2195,3 +2195,430 @@ async fn test_column_non_existent() {
     }
     .run_with_schema(&schema);
 }
+
+/// The following tests were initially in `arrow-rs/parquet/src/arrow/arrow_reader/statistics.rs`.
+/// Part of them was moved here to avoid confusion and duplication,
+/// including edge conditions and data file validations for only `row_group` statistics.
+#[cfg(test)]
+mod test {
+    use super::*;
+    use arrow::util::test_util::parquet_test_data;
+    use arrow_array::{
+        new_empty_array, ArrayRef, BooleanArray, Decimal128Array, Float32Array, Float64Array,
+        Int16Array, Int32Array, Int64Array, Int8Array, RecordBatch, StringArray,
+        TimestampNanosecondArray,
+    };
+    use arrow_schema::{DataType, SchemaRef, TimeUnit};
+    use bytes::Bytes;
+    use parquet::arrow::parquet_column;
+    use parquet::file::metadata::{ParquetMetaData, RowGroupMetaData};
+    use std::path::PathBuf;
+    use std::sync::Arc;
+    // TODO error cases (with parquet statistics that are mismatched in expected type)
+
+    #[test]
+    fn roundtrip_empty() {
+        let all_types = vec![
+            DataType::Null,
+            DataType::Boolean,
+            DataType::Int8,
+            DataType::Int16,
+            DataType::Int32,
+            DataType::Int64,
+            DataType::UInt8,
+            DataType::UInt16,
+            DataType::UInt32,
+            DataType::UInt64,
+            DataType::Float16,
+            DataType::Float32,
+            DataType::Float64,
+            DataType::Timestamp(TimeUnit::Second, None),
+            DataType::Date32,
+            DataType::Date64,
+            // Skip types that don't support statistics
+            // DataType::Time32(Second),
+            // DataType::Time64(Second),
+            // DataType::Duration(Second),
+            // DataType::Interval(IntervalUnit),
+            DataType::Binary,
+            DataType::FixedSizeBinary(0),
+            DataType::LargeBinary,
+            DataType::BinaryView,
+            DataType::Utf8,
+            DataType::LargeUtf8,
+            DataType::Utf8View,
+            // DataType::List(FieldRef),
+            // DataType::ListView(FieldRef),
+            // DataType::FixedSizeList(FieldRef, i32),
+            // DataType::LargeList(FieldRef),
+            // DataType::LargeListView(FieldRef),
+            // DataType::Struct(Fields),
+            // DataType::Union(UnionFields, UnionMode),
+            // DataType::Dictionary(Box<DataType>, Box<DataType>),
+            // DataType::Decimal128(u8, i8),
+            // DataType::Decimal256(u8, i8),
+            // DataType::Map(FieldRef, bool),
+            // DataType::RunEndEncoded(FieldRef, FieldRef),
+        ];
+        for data_type in all_types {
+            let empty_array = new_empty_array(&data_type);
+            Test {
+                input: empty_array.clone(),
+                expected_min: empty_array.clone(),
+                expected_max: empty_array,
+            }
+            .run();
+        }
+    }
+
+    #[test]
+    fn nan_in_stats() {
+        // /parquet-testing/data/nan_in_stats.parquet
+        // row_groups: 1
+        // "x": Double({min: Some(1.0), max: Some(NaN), distinct_count: None, null_count: 0, min_max_deprecated: false, min_max_backwards_compatible: false})
+
+        TestFile::new("nan_in_stats.parquet")
+            .with_column(ExpectedColumn {
+                name: "x",
+                expected_min: Arc::new(Float64Array::from(vec![Some(1.0)])),
+                expected_max: Arc::new(Float64Array::from(vec![Some(f64::NAN)])),
+            })
+            .run();
+    }
+
+    #[test]
+    fn alltypes_plain() {
+        // /parquet-testing/data/datapage_v1-snappy-compressed-checksum.parquet
+        // row_groups: 1
+        // (has no statistics)
+        TestFile::new("alltypes_plain.parquet")
+            // No column statistics should be read as NULL, but with the right type
+            .with_column(ExpectedColumn {
+                name: "id",
+                expected_min: i32_array([None]),
+                expected_max: i32_array([None]),
+            })
+            .with_column(ExpectedColumn {
+                name: "bool_col",
+                expected_min: bool_array([None]),
+                expected_max: bool_array([None]),
+            })
+            .run();
+    }
+
+    #[test]
+    fn alltypes_tiny_pages() {
+        // /parquet-testing/data/alltypes_tiny_pages.parquet
+        // row_groups: 1
+        // "id": Int32({min: Some(0), max: Some(7299), distinct_count: None, null_count: 0, min_max_deprecated: false, min_max_backwards_compatible: false})
+        // "bool_col": Boolean({min: Some(false), max: Some(true), distinct_count: None, null_count: 0, min_max_deprecated: false, min_max_backwards_compatible: false})
+        // "tinyint_col": Int32({min: Some(0), max: Some(9), distinct_count: None, null_count: 0, min_max_deprecated: false, min_max_backwards_compatible: false})
+        // "smallint_col": Int32({min: Some(0), max: Some(9), distinct_count: None, null_count: 0, min_max_deprecated: false, min_max_backwards_compatible: false})
+        // "int_col": Int32({min: Some(0), max: Some(9), distinct_count: None, null_count: 0, min_max_deprecated: false, min_max_backwards_compatible: false})
+        // "bigint_col": Int64({min: Some(0), max: Some(90), distinct_count: None, null_count: 0, min_max_deprecated: false, min_max_backwards_compatible: false})
+        // "float_col": Float({min: Some(0.0), max: Some(9.9), distinct_count: None, null_count: 0, min_max_deprecated: false, min_max_backwards_compatible: false})
+        // "double_col": Double({min: Some(0.0), max: Some(90.89999999999999), distinct_count: None, null_count: 0, min_max_deprecated: false, min_max_backwards_compatible: false})
+        // "date_string_col": ByteArray({min: Some(ByteArray { data: "01/01/09" }), max: Some(ByteArray { data: "12/31/10" }), distinct_count: None, null_count: 0, min_max_deprecated: false, min_max_backwards_compatible: false})
+        // "string_col": ByteArray({min: Some(ByteArray { data: "0" }), max: Some(ByteArray { data: "9" }), distinct_count: None, null_count: 0, min_max_deprecated: false, min_max_backwards_compatible: false})
+        // "timestamp_col": Int96({min: None, max: None, distinct_count: None, null_count: 0, min_max_deprecated: true, min_max_backwards_compatible: true})
+        // "year": Int32({min: Some(2009), max: Some(2010), distinct_count: None, null_count: 0, min_max_deprecated: false, min_max_backwards_compatible: false})
+        // "month": Int32({min: Some(1), max: Some(12), distinct_count: None, null_count: 0, min_max_deprecated: false, min_max_backwards_compatible: false})
+        TestFile::new("alltypes_tiny_pages.parquet")
+            .with_column(ExpectedColumn {
+                name: "id",
+                expected_min: i32_array([Some(0)]),
+                expected_max: i32_array([Some(7299)]),
+            })
+            .with_column(ExpectedColumn {
+                name: "bool_col",
+                expected_min: bool_array([Some(false)]),
+                expected_max: bool_array([Some(true)]),
+            })
+            .with_column(ExpectedColumn {
+                name: "tinyint_col",
+                expected_min: i8_array([Some(0)]),
+                expected_max: i8_array([Some(9)]),
+            })
+            .with_column(ExpectedColumn {
+                name: "smallint_col",
+                expected_min: i16_array([Some(0)]),
+                expected_max: i16_array([Some(9)]),
+            })
+            .with_column(ExpectedColumn {
+                name: "int_col",
+                expected_min: i32_array([Some(0)]),
+                expected_max: i32_array([Some(9)]),
+            })
+            .with_column(ExpectedColumn {
+                name: "bigint_col",
+                expected_min: i64_array([Some(0)]),
+                expected_max: i64_array([Some(90)]),
+            })
+            .with_column(ExpectedColumn {
+                name: "float_col",
+                expected_min: f32_array([Some(0.0)]),
+                expected_max: f32_array([Some(9.9)]),
+            })
+            .with_column(ExpectedColumn {
+                name: "double_col",
+                expected_min: f64_array([Some(0.0)]),
+                expected_max: f64_array([Some(90.89999999999999)]),
+            })
+            .with_column(ExpectedColumn {
+                name: "date_string_col",
+                expected_min: utf8_array([Some("01/01/09")]),
+                expected_max: utf8_array([Some("12/31/10")]),
+            })
+            .with_column(ExpectedColumn {
+                name: "string_col",
+                expected_min: utf8_array([Some("0")]),
+                expected_max: utf8_array([Some("9")]),
+            })
+            // File has no min/max for timestamp_col
+            .with_column(ExpectedColumn {
+                name: "timestamp_col",
+                expected_min: timestamp_nanoseconds_array([None], None),
+                expected_max: timestamp_nanoseconds_array([None], None),
+            })
+            .with_column(ExpectedColumn {
+                name: "year",
+                expected_min: i32_array([Some(2009)]),
+                expected_max: i32_array([Some(2010)]),
+            })
+            .with_column(ExpectedColumn {
+                name: "month",
+                expected_min: i32_array([Some(1)]),
+                expected_max: i32_array([Some(12)]),
+            })
+            .run();
+    }
+
+    #[test]
+    fn fixed_length_decimal_legacy() {
+        // /parquet-testing/data/fixed_length_decimal_legacy.parquet
+        // row_groups: 1
+        // "value": FixedLenByteArray({min: Some(FixedLenByteArray(ByteArray { data: Some(ByteBufferPtr { data: b"\0\0\0\0\0\xc8" }) })), max: Some(FixedLenByteArray(ByteArray { data: "\0\0\0\0\t`" })), distinct_count: None, null_count: 0, min_max_deprecated: true, min_max_backwards_compatible: true})
+
+        TestFile::new("fixed_length_decimal_legacy.parquet")
+            .with_column(ExpectedColumn {
+                name: "value",
+                expected_min: Arc::new(
+                    Decimal128Array::from(vec![Some(200)])
+                        .with_precision_and_scale(13, 2)
+                        .unwrap(),
+                ),
+                expected_max: Arc::new(
+                    Decimal128Array::from(vec![Some(2400)])
+                        .with_precision_and_scale(13, 2)
+                        .unwrap(),
+                ),
+            })
+            .run();
+    }
+
+    const ROWS_PER_ROW_GROUP: usize = 3;
+
+    /// Writes the input batch into a parquet file, with every every three rows as
+    /// their own row group, and compares the min/maxes to the expected values
+    struct Test {
+        input: ArrayRef,
+        expected_min: ArrayRef,
+        expected_max: ArrayRef,
+    }
+
+    impl Test {
+        fn run(self) {
+            let Self {
+                input,
+                expected_min,
+                expected_max,
+            } = self;
+
+            let input_batch = RecordBatch::try_from_iter([("c1", input)]).unwrap();
+
+            let schema = input_batch.schema();
+
+            let metadata = parquet_metadata(schema.clone(), input_batch);
+            let parquet_schema = metadata.file_metadata().schema_descr();
+
+            let row_groups = metadata.row_groups();
+
+            for field in schema.fields() {
+                if field.data_type().is_nested() {
+                    let lookup = parquet_column(parquet_schema, &schema, field.name());
+                    assert_eq!(lookup, None);
+                    continue;
+                }
+
+                let converter =
+                    StatisticsConverter::try_new(field.name(), &schema, parquet_schema).unwrap();
+
+                assert_eq!(converter.arrow_field(), field.as_ref());
+
+                let mins = converter.row_group_mins(row_groups.iter()).unwrap();
+                assert_eq!(
+                    &mins,
+                    &expected_min,
+                    "Min. Statistics\n\n{}\n\n",
+                    DisplayStats(row_groups)
+                );
+
+                let maxes = converter.row_group_maxes(row_groups.iter()).unwrap();
+                assert_eq!(
+                    &maxes,
+                    &expected_max,
+                    "Max. Statistics\n\n{}\n\n",
+                    DisplayStats(row_groups)
+                );
+            }
+        }
+    }
+
+    /// Write the specified batches out as parquet and return the metadata
+    fn parquet_metadata(schema: SchemaRef, batch: RecordBatch) -> Arc<ParquetMetaData> {
+        let props = WriterProperties::builder()
+            .set_statistics_enabled(EnabledStatistics::Chunk)
+            .set_max_row_group_size(ROWS_PER_ROW_GROUP)
+            .build();
+
+        let mut buffer = Vec::new();
+        let mut writer = ArrowWriter::try_new(&mut buffer, schema, Some(props)).unwrap();
+        writer.write(&batch).unwrap();
+        writer.close().unwrap();
+
+        let reader = ArrowReaderBuilder::try_new(Bytes::from(buffer)).unwrap();
+        reader.metadata().clone()
+    }
+
+    /// Formats the statistics nicely for display
+    struct DisplayStats<'a>(&'a [RowGroupMetaData]);
+    impl<'a> std::fmt::Display for DisplayStats<'a> {
+        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+            let row_groups = self.0;
+            writeln!(f, "  row_groups: {}", row_groups.len())?;
+            for rg in row_groups {
+                for col in rg.columns() {
+                    if let Some(statistics) = col.statistics() {
+                        writeln!(f, "   {}: {:?}", col.column_path(), statistics)?;
+                    }
+                }
+            }
+            Ok(())
+        }
+    }
+
+    struct ExpectedColumn {
+        name: &'static str,
+        expected_min: ArrayRef,
+        expected_max: ArrayRef,
+    }
+
+    /// Reads statistics out of the specified, and compares them to the expected values
+    struct TestFile {
+        file_name: &'static str,
+        expected_columns: Vec<ExpectedColumn>,
+    }
+
+    impl TestFile {
+        fn new(file_name: &'static str) -> Self {
+            Self {
+                file_name,
+                expected_columns: Vec::new(),
+            }
+        }
+
+        fn with_column(mut self, column: ExpectedColumn) -> Self {
+            self.expected_columns.push(column);
+            self
+        }
+
+        /// Reads the specified parquet file and validates that the expected min/max
+        /// values for the specified columns are as expected.
+        fn run(self) {
+            let path = PathBuf::from(parquet_test_data()).join(self.file_name);
+            let file = File::open(path).unwrap();
+            let reader = ArrowReaderBuilder::try_new(file).unwrap();
+            let arrow_schema = reader.schema();
+            let metadata = reader.metadata();
+            let row_groups = metadata.row_groups();
+            let parquet_schema = metadata.file_metadata().schema_descr();
+
+            for expected_column in self.expected_columns {
+                let ExpectedColumn {
+                    name,
+                    expected_min,
+                    expected_max,
+                } = expected_column;
+
+                let converter =
+                    StatisticsConverter::try_new(name, arrow_schema, parquet_schema).unwrap();
+
+                // test accessors on the converter
+                let parquet_column_index =
+                    parquet_column(parquet_schema, arrow_schema, name).map(|(idx, _field)| idx);
+                assert_eq!(converter.parquet_column_index(), parquet_column_index);
+                assert_eq!(converter.arrow_field().name(), name);
+
+                let actual_min = converter.row_group_mins(row_groups.iter()).unwrap();
+                assert_eq!(&expected_min, &actual_min, "column {name}");
+
+                let actual_max = converter.row_group_maxes(row_groups.iter()).unwrap();
+                assert_eq!(&expected_max, &actual_max, "column {name}");
+            }
+        }
+    }
+
+    fn bool_array(input: impl IntoIterator<Item = Option<bool>>) -> ArrayRef {
+        let array: BooleanArray = input.into_iter().collect();
+        Arc::new(array)
+    }
+
+    fn i8_array(input: impl IntoIterator<Item = Option<i8>>) -> ArrayRef {
+        let array: Int8Array = input.into_iter().collect();
+        Arc::new(array)
+    }
+
+    fn i16_array(input: impl IntoIterator<Item = Option<i16>>) -> ArrayRef {
+        let array: Int16Array = input.into_iter().collect();
+        Arc::new(array)
+    }
+
+    fn i32_array(input: impl IntoIterator<Item = Option<i32>>) -> ArrayRef {
+        let array: Int32Array = input.into_iter().collect();
+        Arc::new(array)
+    }
+
+    fn i64_array(input: impl IntoIterator<Item = Option<i64>>) -> ArrayRef {
+        let array: Int64Array = input.into_iter().collect();
+        Arc::new(array)
+    }
+
+    fn f32_array(input: impl IntoIterator<Item = Option<f32>>) -> ArrayRef {
+        let array: Float32Array = input.into_iter().collect();
+        Arc::new(array)
+    }
+
+    fn f64_array(input: impl IntoIterator<Item = Option<f64>>) -> ArrayRef {
+        let array: Float64Array = input.into_iter().collect();
+        Arc::new(array)
+    }
+
+    fn timestamp_nanoseconds_array(
+        input: impl IntoIterator<Item = Option<i64>>,
+        timzezone: Option<&str>,
+    ) -> ArrayRef {
+        let array: TimestampNanosecondArray = input.into_iter().collect();
+        match timzezone {
+            Some(tz) => Arc::new(array.with_timezone(tz)),
+            None => Arc::new(array),
+        }
+    }
+
+    fn utf8_array<'a>(input: impl IntoIterator<Item = Option<&'a str>>) -> ArrayRef {
+        let array: StringArray = input
+            .into_iter()
+            .map(|s| s.map(|s| s.to_string()))
+            .collect();
+        Arc::new(array)
+    }
+}
diff --git a/testing b/testing
index e270341fb..735ae7128 160000
--- a/testing
+++ b/testing
@@ -1,1 +1,1 @@
-Subproject commit e270341fb5f3ff785410e6286cc42898e9d6a99c
+Subproject commit 735ae7128d571398dd798d7ff004adebeb342883