Repository: apache/arrow-rs
Pull request #6187 (2024-08-03 11:22:56)

Make it clear that `StatisticsConverter` can not panic
# Which issue does this PR close?

Part of https://github.com/apache/arrow-rs/issues/6164

# Rationale for this change
 
As discussed with @Kev1n8 on https://github.com/apache/arrow-rs/pull/6181#discussion_r1702702575 the `DataType` is always known and provided, so having the code check for `None` is unnecessary and confusing.


# What changes are included in this PR?

Remove the `Option` handling in code that is always invoked with `Some`

# Are there any user-facing changes?
No

<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->

<!---
If there are any breaking changes to public APIs, please add the `breaking change` label.
-->



Patch:
diff --git a/arrow-array/src/array/primitive_array.rs b/arrow-array/src/array/primitive_array.rs
index 7ec9ff928..70a8ceaef 100644
--- a/arrow-array/src/array/primitive_array.rs
+++ b/arrow-array/src/array/primitive_array.rs
@@ -239,13 +239,13 @@ pub type Float32Array = PrimitiveArray<Float32Type>;
 /// Construction
 ///
 /// ```
-/// # use arrow_array::Float32Array;
+/// # use arrow_array::Float64Array;
 /// // Create from Vec<Option<f32>>
-/// let arr = Float32Array::from(vec![Some(1.0), None, Some(2.0)]);
+/// let arr = Float64Array::from(vec![Some(1.0), None, Some(2.0)]);
 /// // Create from Vec<f32>
-/// let arr = Float32Array::from(vec![1.0, 2.0, 3.0]);
+/// let arr = Float64Array::from(vec![1.0, 2.0, 3.0]);
 /// // Create iter/collect
-/// let arr: Float32Array = std::iter::repeat(42.0).take(10).collect();
+/// let arr: Float64Array = std::iter::repeat(42.0).take(10).collect();
 /// ```
 ///
 /// See [`PrimitiveArray`] for more information and examples
diff --git a/arrow-row/src/lib.rs b/arrow-row/src/lib.rs
index a6fd03b5b..2d9af7575 100644
--- a/arrow-row/src/lib.rs
+++ b/arrow-row/src/lib.rs
@@ -835,10 +835,20 @@ impl Rows {
 
     /// Returns the row at index `row`
     pub fn row(&self, row: usize) -> Row<'_> {
-        let end = self.offsets[row + 1];
-        let start = self.offsets[row];
+        assert!(row + 1 < self.offsets.len());
+        unsafe { self.row_unchecked(row) }
+    }
+
+    /// Returns the row at `index` without bounds checking
+    ///
+    /// # Safety
+    /// Caller must ensure that `index` is less than the number of offsets (#rows + 1)
+    pub unsafe fn row_unchecked(&self, index: usize) -> Row<'_> {
+        let end = unsafe { self.offsets.get_unchecked(index + 1) };
+        let start = unsafe { self.offsets.get_unchecked(index) };
+        let data = unsafe { self.buffer.get_unchecked(*start..*end) };
         Row {
-            data: &self.buffer[start..end],
+            data,
             config: &self.config,
         }
     }
@@ -898,7 +908,9 @@ impl<'a> Iterator for RowsIter<'a> {
         if self.end == self.start {
             return None;
         }
-        let row = self.rows.row(self.start);
+
+        // SAFETY: We have checked that `start` is less than `end`
+        let row = unsafe { self.rows.row_unchecked(self.start) };
         self.start += 1;
         Some(row)
     }
@@ -920,7 +932,8 @@ impl<'a> DoubleEndedIterator for RowsIter<'a> {
         if self.end == self.start {
             return None;
         }
-        let row = self.rows.row(self.end);
+        // Safety: We have checked that `start` is less than `end`
+        let row = unsafe { self.rows.row_unchecked(self.end) };
         self.end -= 1;
         Some(row)
     }
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/benches/aggregate_kernels.rs b/arrow/benches/aggregate_kernels.rs
index 434bb4778..6e224a48c 100644
--- a/arrow/benches/aggregate_kernels.rs
+++ b/arrow/benches/aggregate_kernels.rs
@@ -104,6 +104,9 @@ fn add_benchmark(c: &mut Criterion) {
             .bench_function("or nonnull mixed", |b| {
                 b.iter(|| bool_or(&nonnull_bools_mixed))
             })
+            .bench_function("and nonnull mixed", |b| {
+                b.iter(|| bool_and(&nonnull_bools_mixed))
+            })
             .bench_function("min nonnull false", |b| {
                 b.iter(|| min_boolean(&nonnull_bools_all_false))
             })
@@ -113,6 +116,9 @@ fn add_benchmark(c: &mut Criterion) {
             .bench_function("or nonnull false", |b| {
                 b.iter(|| bool_or(&nonnull_bools_all_false))
             })
+            .bench_function("and nonnull false", |b| {
+                b.iter(|| bool_and(&nonnull_bools_all_false))
+            })
             .bench_function("min nonnull true", |b| {
                 b.iter(|| min_boolean(&nonnull_bools_all_true))
             })
@@ -122,6 +128,9 @@ fn add_benchmark(c: &mut Criterion) {
             .bench_function("or nonnull true", |b| {
                 b.iter(|| bool_or(&nonnull_bools_all_true))
             })
+            .bench_function("and nonnull true", |b| {
+                b.iter(|| bool_and(&nonnull_bools_all_true))
+            })
             .bench_function("min nullable mixed", |b| {
                 b.iter(|| min_boolean(&nullable_bool_mixed))
             })
@@ -131,6 +140,9 @@ fn add_benchmark(c: &mut Criterion) {
             .bench_function("or nullable mixed", |b| {
                 b.iter(|| bool_or(&nullable_bool_mixed))
             })
+            .bench_function("and nullable mixed", |b| {
+                b.iter(|| bool_and(&nullable_bool_mixed))
+            })
             .bench_function("min nullable false", |b| {
                 b.iter(|| min_boolean(&nullable_bool_all_false))
             })
@@ -140,6 +152,9 @@ fn add_benchmark(c: &mut Criterion) {
             .bench_function("or nullable false", |b| {
                 b.iter(|| bool_or(&nullable_bool_all_false))
             })
+            .bench_function("and nullable false", |b| {
+                b.iter(|| bool_and(&nullable_bool_all_false))
+            })
             .bench_function("min nullable true", |b| {
                 b.iter(|| min_boolean(&nullable_bool_all_true))
             })
@@ -148,6 +163,9 @@ fn add_benchmark(c: &mut Criterion) {
             })
             .bench_function("or nullable true", |b| {
                 b.iter(|| bool_or(&nullable_bool_all_true))
+            })
+            .bench_function("and nullable true", |b| {
+                b.iter(|| bool_and(&nullable_bool_all_true))
             });
     }
 }
diff --git a/arrow/benches/row_format.rs b/arrow/benches/row_format.rs
index 0fb63b5b3..773cdc652 100644
--- a/arrow/benches/row_format.rs
+++ b/arrow/benches/row_format.rs
@@ -56,6 +56,22 @@ fn do_bench(c: &mut Criterion, name: &str, cols: Vec<ArrayRef>) {
     });
 }
 
+fn bench_iter(c: &mut Criterion) {
+    let col = create_string_view_array_with_len(40960, 0., 100, false);
+    let converter = RowConverter::new(vec![SortField::new(col.data_type().clone())]).unwrap();
+    let rows = converter
+        .convert_columns(&[Arc::new(col) as ArrayRef])
+        .unwrap();
+
+    c.bench_function("iterate rows", |b| {
+        b.iter(|| {
+            for r in rows.iter() {
+                std::hint::black_box(r.as_ref());
+            }
+        })
+    });
+}
+
 fn row_bench(c: &mut Criterion) {
     let cols = vec![Arc::new(create_primitive_array::<UInt64Type>(4096, 0.)) as ArrayRef];
     do_bench(c, "4096 u64(0)", cols);
@@ -145,6 +161,8 @@ fn row_bench(c: &mut Criterion) {
         Arc::new(create_primitive_array::<Int64Type>(4096, 0.)) as ArrayRef,
     ];
     do_bench(c, "4096 4096 string_dictionary(20, 0.5), string_dictionary(30, 0), string_dictionary(100, 0), i64(0)", cols);
+
+    bench_iter(c);
 }
 
 criterion_group!(benches, row_bench);
diff --git a/parquet/Cargo.toml b/parquet/Cargo.toml
index f95982759..ad740b0a2 100644
--- a/parquet/Cargo.toml
+++ b/parquet/Cargo.toml
@@ -44,7 +44,7 @@ arrow-schema = { workspace = true, optional = true }
 arrow-select = { workspace = true, optional = true }
 arrow-ipc = { workspace = true, optional = true }
 # Intentionally not a path dependency as object_store is released separately
-object_store = { version = "0.10.0", default-features = false, optional = true }
+object_store = { version = "0.10.2", default-features = false, optional = true }
 
 bytes = { version = "1.1", default-features = false, features = ["std"] }
 thrift = { version = "0.17", default-features = false }
@@ -82,15 +82,15 @@ serde_json = { version = "1.0", features = ["std"], default-features = false }
 arrow = { workspace = true, features = ["ipc", "test_utils", "prettyprint", "json"] }
 tokio = { version = "1.0", default-features = false, features = ["macros", "rt", "io-util", "fs"] }
 rand = { version = "0.8", default-features = false, features = ["std", "std_rng"] }
-object_store = { version = "0.10.0", default-features = false, features = ["azure"] }
+object_store = { version = "0.10.2", default-features = false, features = ["azure"] }
 
 # TODO: temporary to fix parquet wasm build
 # upstream issue: https://github.com/gyscos/zstd-rs/issues/269
 [target.'cfg(target_family = "wasm")'.dependencies]
-zstd-sys = { version = ">=2.0.0, <2.0.13", optional = true, default-features = false }
+zstd-sys = { version = ">=2.0.0, <2.0.14", optional = true, default-features = false }
 
 [target.'cfg(target_family = "wasm")'.dev-dependencies]
-zstd-sys = { version = ">=2.0.0, <2.0.13", default-features = false }
+zstd-sys = { version = ">=2.0.0, <2.0.14", default-features = false }
 
 [package.metadata.docs.rs]
 all-features = true
diff --git a/parquet/benches/encoding.rs b/parquet/benches/encoding.rs
index bdbca3567..80befe8da 100644
--- a/parquet/benches/encoding.rs
+++ b/parquet/benches/encoding.rs
@@ -30,30 +30,30 @@ fn bench_typed<T: DataType>(c: &mut Criterion, values: &[T::T], encoding: Encodi
         std::any::type_name::<T::T>(),
         encoding
     );
+    let column_desc_ptr = ColumnDescPtr::new(ColumnDescriptor::new(
+        Arc::new(
+            Type::primitive_type_builder("", T::get_physical_type())
+                .build()
+                .unwrap(),
+        ),
+        0,
+        0,
+        ColumnPath::new(vec![]),
+    ));
     c.bench_function(&format!("encoding: {}", name), |b| {
         b.iter(|| {
-            let mut encoder = get_encoder::<T>(encoding).unwrap();
+            let mut encoder = get_encoder::<T>(encoding, &column_desc_ptr).unwrap();
             encoder.put(values).unwrap();
             encoder.flush_buffer().unwrap();
         });
     });
 
-    let mut encoder = get_encoder::<T>(encoding).unwrap();
+    let mut encoder = get_encoder::<T>(encoding, &column_desc_ptr).unwrap();
     encoder.put(values).unwrap();
     let encoded = encoder.flush_buffer().unwrap();
     println!("{} encoded as {} bytes", name, encoded.len(),);
 
     let mut buffer = vec![T::T::default(); values.len()];
-    let column_desc_ptr = ColumnDescPtr::new(ColumnDescriptor::new(
-        Arc::new(
-            Type::primitive_type_builder("", T::get_physical_type())
-                .build()
-                .unwrap(),
-        ),
-        0,
-        0,
-        ColumnPath::new(vec![]),
-    ));
     c.bench_function(&format!("decoding: {}", name), |b| {
         b.iter(|| {
             let mut decoder: Box<dyn Decoder<T>> =
diff --git a/parquet/src/arrow/array_reader/fixed_len_byte_array.rs b/parquet/src/arrow/array_reader/fixed_len_byte_array.rs
index a9159bb47..8098f3240 100644
--- a/parquet/src/arrow/array_reader/fixed_len_byte_array.rs
+++ b/parquet/src/arrow/array_reader/fixed_len_byte_array.rs
@@ -341,6 +341,10 @@ impl ColumnValueDecoder for ValueDecoder {
             Encoding::DELTA_BYTE_ARRAY => Decoder::Delta {
                 decoder: DeltaByteArrayDecoder::new(data)?,
             },
+            Encoding::BYTE_STREAM_SPLIT => Decoder::ByteStreamSplit {
+                buf: data,
+                offset: 0,
+            },
             _ => {
                 return Err(general_err!(
                     "unsupported encoding for fixed length byte array: {}",
@@ -400,6 +404,20 @@ impl ColumnValueDecoder for ValueDecoder {
                     Ok(())
                 })
             }
+            Decoder::ByteStreamSplit { buf, offset } => {
+                // we have n=`byte_length` streams of length `buf.len/byte_length`
+                // to read value i, we need the i'th byte from each of the streams
+                // so `offset` should be the value offset, not the byte offset
+                let total_values = buf.len() / self.byte_length;
+                let to_read = num_values.min(total_values - *offset);
+                out.buffer.reserve(to_read * self.byte_length);
+
+                // now read the n streams and reassemble values into the output buffer
+                read_byte_stream_split(&mut out.buffer, buf, *offset, to_read, self.byte_length);
+
+                *offset += to_read;
+                Ok(to_read)
+            }
         }
     }
 
@@ -412,6 +430,32 @@ impl ColumnValueDecoder for ValueDecoder {
             }
             Decoder::Dict { decoder } => decoder.skip(num_values),
             Decoder::Delta { decoder } => decoder.skip(num_values),
+            Decoder::ByteStreamSplit { offset, buf } => {
+                let total_values = buf.len() / self.byte_length;
+                let to_read = num_values.min(total_values - *offset);
+                *offset += to_read;
+                Ok(to_read)
+            }
+        }
+    }
+}
+
+// `src` is an array laid out like a NxM matrix where N == `data_width` and
+// M == total_values_in_src. Each "row" of the matrix is a stream of bytes, with stream `i`
+// containing the `ith` byte for each value. Each "column" is a single value.
+// This will reassemble `num_values` values by reading columns of the matrix starting at
+// `offset`. Values will be appended to `dst`.
+fn read_byte_stream_split(
+    dst: &mut Vec<u8>,
+    src: &mut Bytes,
+    offset: usize,
+    num_values: usize,
+    data_width: usize,
+) {
+    let stride = src.len() / data_width;
+    for i in 0..num_values {
+        for j in 0..data_width {
+            dst.push(src[offset + j * stride + i]);
         }
     }
 }
@@ -420,6 +464,7 @@ enum Decoder {
     Plain { buf: Bytes, offset: usize },
     Dict { decoder: DictIndexDecoder },
     Delta { decoder: DeltaByteArrayDecoder },
+    ByteStreamSplit { buf: Bytes, offset: usize },
 }
 
 #[cfg(test)]
diff --git a/parquet/src/arrow/arrow_reader/mod.rs b/parquet/src/arrow/arrow_reader/mod.rs
index c696763d6..a0302fa86 100644
--- a/parquet/src/arrow/arrow_reader/mod.rs
+++ b/parquet/src/arrow/arrow_reader/mod.rs
@@ -1059,6 +1059,7 @@ mod tests {
                 Encoding::PLAIN,
                 Encoding::RLE_DICTIONARY,
                 Encoding::DELTA_BINARY_PACKED,
+                Encoding::BYTE_STREAM_SPLIT,
             ],
         );
         run_single_column_reader_tests::<Int64Type, _, Int64Type>(
@@ -1070,6 +1071,7 @@ mod tests {
                 Encoding::PLAIN,
                 Encoding::RLE_DICTIONARY,
                 Encoding::DELTA_BINARY_PACKED,
+                Encoding::BYTE_STREAM_SPLIT,
             ],
         );
         run_single_column_reader_tests::<FloatType, _, FloatType>(
@@ -1641,6 +1643,86 @@ mod tests {
         assert_eq!(row_count, 300);
     }
 
+    #[test]
+    fn test_read_extended_byte_stream_split() {
+        let path = format!(
+            "{}/byte_stream_split_extended.gzip.parquet",
+            arrow::util::test_util::parquet_test_data(),
+        );
+        let file = File::open(path).unwrap();
+        let record_reader = ParquetRecordBatchReader::try_new(file, 128).unwrap();
+
+        let mut row_count = 0;
+        for batch in record_reader {
+            let batch = batch.unwrap();
+            row_count += batch.num_rows();
+
+            // 0,1 are f16
+            let f16_col = batch.column(0).as_primitive::<Float16Type>();
+            let f16_bss = batch.column(1).as_primitive::<Float16Type>();
+            assert_eq!(f16_col.len(), f16_bss.len());
+            f16_col
+                .iter()
+                .zip(f16_bss.iter())
+                .for_each(|(l, r)| assert_eq!(l.unwrap(), r.unwrap()));
+
+            // 2,3 are f32
+            let f32_col = batch.column(2).as_primitive::<Float32Type>();
+            let f32_bss = batch.column(3).as_primitive::<Float32Type>();
+            assert_eq!(f32_col.len(), f32_bss.len());
+            f32_col
+                .iter()
+                .zip(f32_bss.iter())
+                .for_each(|(l, r)| assert_eq!(l.unwrap(), r.unwrap()));
+
+            // 4,5 are f64
+            let f64_col = batch.column(4).as_primitive::<Float64Type>();
+            let f64_bss = batch.column(5).as_primitive::<Float64Type>();
+            assert_eq!(f64_col.len(), f64_bss.len());
+            f64_col
+                .iter()
+                .zip(f64_bss.iter())
+                .for_each(|(l, r)| assert_eq!(l.unwrap(), r.unwrap()));
+
+            // 6,7 are i32
+            let i32_col = batch.column(6).as_primitive::<types::Int32Type>();
+            let i32_bss = batch.column(7).as_primitive::<types::Int32Type>();
+            assert_eq!(i32_col.len(), i32_bss.len());
+            i32_col
+                .iter()
+                .zip(i32_bss.iter())
+                .for_each(|(l, r)| assert_eq!(l.unwrap(), r.unwrap()));
+
+            // 8,9 are i64
+            let i64_col = batch.column(8).as_primitive::<types::Int64Type>();
+            let i64_bss = batch.column(9).as_primitive::<types::Int64Type>();
+            assert_eq!(i64_col.len(), i64_bss.len());
+            i64_col
+                .iter()
+                .zip(i64_bss.iter())
+                .for_each(|(l, r)| assert_eq!(l.unwrap(), r.unwrap()));
+
+            // 10,11 are FLBA(5)
+            let flba_col = batch.column(10).as_fixed_size_binary();
+            let flba_bss = batch.column(11).as_fixed_size_binary();
+            assert_eq!(flba_col.len(), flba_bss.len());
+            flba_col
+                .iter()
+                .zip(flba_bss.iter())
+                .for_each(|(l, r)| assert_eq!(l.unwrap(), r.unwrap()));
+
+            // 12,13 are FLBA(4) (decimal(7,3))
+            let dec_col = batch.column(12).as_primitive::<Decimal128Type>();
+            let dec_bss = batch.column(13).as_primitive::<Decimal128Type>();
+            assert_eq!(dec_col.len(), dec_bss.len());
+            dec_col
+                .iter()
+                .zip(dec_bss.iter())
+                .for_each(|(l, r)| assert_eq!(l.unwrap(), r.unwrap()));
+        }
+        assert_eq!(row_count, 200);
+    }
+
     #[test]
     fn test_read_incorrect_map_schema_file() {
         let testdata = arrow::util::test_util::parquet_test_data();
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/arrow/arrow_writer/mod.rs b/parquet/src/arrow/arrow_writer/mod.rs
index db5d4502c..0c07f541b 100644
--- a/parquet/src/arrow/arrow_writer/mod.rs
+++ b/parquet/src/arrow/arrow_writer/mod.rs
@@ -1791,7 +1791,11 @@ mod tests {
             | DataType::UInt64
             | DataType::UInt32
             | DataType::UInt16
-            | DataType::UInt8 => vec![Encoding::PLAIN, Encoding::DELTA_BINARY_PACKED],
+            | DataType::UInt8 => vec![
+                Encoding::PLAIN,
+                Encoding::DELTA_BINARY_PACKED,
+                Encoding::BYTE_STREAM_SPLIT,
+            ],
             DataType::Float32 | DataType::Float64 => {
                 vec![Encoding::PLAIN, Encoding::BYTE_STREAM_SPLIT]
             }
diff --git a/parquet/src/arrow/async_writer/mod.rs b/parquet/src/arrow/async_writer/mod.rs
index 274d8fef8..50bb5c046 100644
--- a/parquet/src/arrow/async_writer/mod.rs
+++ b/parquet/src/arrow/async_writer/mod.rs
@@ -22,6 +22,7 @@
 //! read the documentation there before using this API.
 //!
 //! Here is an example for using [`AsyncArrowWriter`]:
+//!
 //! ```
 //! # #[tokio::main(flavor="current_thread")]
 //! # async fn main() {
@@ -49,6 +50,13 @@
 //! assert_eq!(to_write, read);
 //! # }
 //! ```
+//!
+//! [`object_store`] provides it's native implementation of [`AsyncFileWriter`] by [`ParquetObjectWriter`].
+
+#[cfg(feature = "object_store")]
+mod store;
+#[cfg(feature = "object_store")]
+pub use store::*;
 
 use crate::{
     arrow::arrow_writer::ArrowWriterOptions,
@@ -65,7 +73,7 @@ use futures::FutureExt;
 use std::mem;
 use tokio::io::{AsyncWrite, AsyncWriteExt};
 
-/// The asynchronous interface used by [`AsyncArrowWriter`] to write parquet files
+/// The asynchronous interface used by [`AsyncArrowWriter`] to write parquet files.
 pub trait AsyncFileWriter: Send {
     /// Write the provided bytes to the underlying writer
     ///
diff --git a/parquet/src/arrow/async_writer/store.rs b/parquet/src/arrow/async_writer/store.rs
new file mode 100644
index 000000000..ad09eae49
--- /dev/null
+++ b/parquet/src/arrow/async_writer/store.rs
@@ -0,0 +1,157 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use bytes::Bytes;
+use futures::future::BoxFuture;
+use std::sync::Arc;
+
+use crate::arrow::async_writer::AsyncFileWriter;
+use crate::errors::{ParquetError, Result};
+use object_store::buffered::BufWriter;
+use object_store::path::Path;
+use object_store::ObjectStore;
+use tokio::io::AsyncWriteExt;
+
+/// [`ParquetObjectWriter`] for writing to parquet to [`ObjectStore`]
+///
+/// ```
+/// # use arrow_array::{ArrayRef, Int64Array, RecordBatch};
+/// # use object_store::memory::InMemory;
+/// # use object_store::path::Path;
+/// # use object_store::ObjectStore;
+/// # use std::sync::Arc;
+///
+/// # use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
+/// # use parquet::arrow::async_writer::ParquetObjectWriter;
+/// # use parquet::arrow::AsyncArrowWriter;
+///
+/// # #[tokio::main(flavor="current_thread")]
+/// # async fn main() {
+///     let store = Arc::new(InMemory::new());
+///
+///     let col = Arc::new(Int64Array::from_iter_values([1, 2, 3])) as ArrayRef;
+///     let to_write = RecordBatch::try_from_iter([("col", col)]).unwrap();
+///
+///     let object_store_writer = ParquetObjectWriter::new(store.clone(), Path::from("test"));
+///     let mut writer =
+///         AsyncArrowWriter::try_new(object_store_writer, to_write.schema(), None).unwrap();
+///     writer.write(&to_write).await.unwrap();
+///     writer.close().await.unwrap();
+///
+///     let buffer = store
+///         .get(&Path::from("test"))
+///         .await
+///         .unwrap()
+///         .bytes()
+///         .await
+///         .unwrap();
+///     let mut reader = ParquetRecordBatchReaderBuilder::try_new(buffer)
+///         .unwrap()
+///         .build()
+///         .unwrap();
+///     let read = reader.next().unwrap().unwrap();
+///
+///     assert_eq!(to_write, read);
+/// # }
+/// ```
+#[derive(Debug)]
+pub struct ParquetObjectWriter {
+    w: BufWriter,
+}
+
+impl ParquetObjectWriter {
+    /// Create a new [`ParquetObjectWriter`] that writes to the specified path in the given store.
+    ///
+    /// To configure the writer behavior, please build [`BufWriter`] and then use [`Self::from_buf_writer`]
+    pub fn new(store: Arc<dyn ObjectStore>, path: Path) -> Self {
+        Self::from_buf_writer(BufWriter::new(store, path))
+    }
+
+    /// Construct a new ParquetObjectWriter via a existing BufWriter.
+    pub fn from_buf_writer(w: BufWriter) -> Self {
+        Self { w }
+    }
+
+    /// Consume the writer and return the underlying BufWriter.
+    pub fn into_inner(self) -> BufWriter {
+        self.w
+    }
+}
+
+impl AsyncFileWriter for ParquetObjectWriter {
+    fn write(&mut self, bs: Bytes) -> BoxFuture<'_, Result<()>> {
+        Box::pin(async {
+            self.w
+                .put(bs)
+                .await
+                .map_err(|err| ParquetError::External(Box::new(err)))
+        })
+    }
+
+    fn complete(&mut self) -> BoxFuture<'_, Result<()>> {
+        Box::pin(async {
+            self.w
+                .shutdown()
+                .await
+                .map_err(|err| ParquetError::External(Box::new(err)))
+        })
+    }
+}
+impl From<BufWriter> for ParquetObjectWriter {
+    fn from(w: BufWriter) -> Self {
+        Self::from_buf_writer(w)
+    }
+}
+#[cfg(test)]
+mod tests {
+    use arrow_array::{ArrayRef, Int64Array, RecordBatch};
+    use object_store::memory::InMemory;
+    use std::sync::Arc;
+
+    use super::*;
+    use crate::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
+    use crate::arrow::AsyncArrowWriter;
+
+    #[tokio::test]
+    async fn test_async_writer() {
+        let store = Arc::new(InMemory::new());
+
+        let col = Arc::new(Int64Array::from_iter_values([1, 2, 3])) as ArrayRef;
+        let to_write = RecordBatch::try_from_iter([("col", col)]).unwrap();
+
+        let object_store_writer = ParquetObjectWriter::new(store.clone(), Path::from("test"));
+        let mut writer =
+            AsyncArrowWriter::try_new(object_store_writer, to_write.schema(), None).unwrap();
+        writer.write(&to_write).await.unwrap();
+        writer.close().await.unwrap();
+
+        let buffer = store
+            .get(&Path::from("test"))
+            .await
+            .unwrap()
+            .bytes()
+            .await
+            .unwrap();
+        let mut reader = ParquetRecordBatchReaderBuilder::try_new(buffer)
+            .unwrap()
+            .build()
+            .unwrap();
+        let read = reader.next().unwrap().unwrap();
+
+        assert_eq!(to_write, read);
+    }
+}
diff --git a/parquet/src/column/writer/encoder.rs b/parquet/src/column/writer/encoder.rs
index 9d01c0904..7371c72a5 100644
--- a/parquet/src/column/writer/encoder.rs
+++ b/parquet/src/column/writer/encoder.rs
@@ -191,6 +191,7 @@ impl<T: DataType> ColumnValueEncoder for ColumnValueEncoderImpl<T> {
             props
                 .encoding(descr.path())
                 .unwrap_or_else(|| fallback_encoding(T::get_physical_type(), props)),
+            descr,
         )?;
 
         let statistics_enabled = props.statistics_enabled(descr.path());
diff --git a/parquet/src/encodings/decoding.rs b/parquet/src/encodings/decoding.rs
index b3dd13523..16467b32d 100644
--- a/parquet/src/encodings/decoding.rs
+++ b/parquet/src/encodings/decoding.rs
@@ -27,6 +27,9 @@ use super::rle::RleDecoder;
 use crate::basic::*;
 use crate::data_type::private::ParquetValueType;
 use crate::data_type::*;
+use crate::encodings::decoding::byte_stream_split_decoder::{
+    ByteStreamSplitDecoder, VariableWidthByteStreamSplitDecoder,
+};
 use crate::errors::{ParquetError, Result};
 use crate::schema::types::ColumnDescPtr;
 use crate::util::bit_util::{self, BitReader};
@@ -87,6 +90,7 @@ pub(crate) mod private {
             encoding: Encoding,
         ) -> Result<Box<dyn Decoder<T>>> {
             match encoding {
+                Encoding::BYTE_STREAM_SPLIT => Ok(Box::new(ByteStreamSplitDecoder::new())),
                 Encoding::DELTA_BINARY_PACKED => Ok(Box::new(DeltaBitPackDecoder::new())),
                 _ => get_decoder_default(descr, encoding),
             }
@@ -99,6 +103,7 @@ pub(crate) mod private {
             encoding: Encoding,
         ) -> Result<Box<dyn Decoder<T>>> {
             match encoding {
+                Encoding::BYTE_STREAM_SPLIT => Ok(Box::new(ByteStreamSplitDecoder::new())),
                 Encoding::DELTA_BINARY_PACKED => Ok(Box::new(DeltaBitPackDecoder::new())),
                 _ => get_decoder_default(descr, encoding),
             }
@@ -111,9 +116,7 @@ pub(crate) mod private {
             encoding: Encoding,
         ) -> Result<Box<dyn Decoder<T>>> {
             match encoding {
-                Encoding::BYTE_STREAM_SPLIT => Ok(Box::new(
-                    byte_stream_split_decoder::ByteStreamSplitDecoder::new(),
-                )),
+                Encoding::BYTE_STREAM_SPLIT => Ok(Box::new(ByteStreamSplitDecoder::new())),
                 _ => get_decoder_default(descr, encoding),
             }
         }
@@ -124,9 +127,7 @@ pub(crate) mod private {
             encoding: Encoding,
         ) -> Result<Box<dyn Decoder<T>>> {
             match encoding {
-                Encoding::BYTE_STREAM_SPLIT => Ok(Box::new(
-                    byte_stream_split_decoder::ByteStreamSplitDecoder::new(),
-                )),
+                Encoding::BYTE_STREAM_SPLIT => Ok(Box::new(ByteStreamSplitDecoder::new())),
                 _ => get_decoder_default(descr, encoding),
             }
         }
@@ -153,6 +154,9 @@ pub(crate) mod private {
             encoding: Encoding,
         ) -> Result<Box<dyn Decoder<T>>> {
             match encoding {
+                Encoding::BYTE_STREAM_SPLIT => Ok(Box::new(
+                    VariableWidthByteStreamSplitDecoder::new(descr.type_length()),
+                )),
                 Encoding::DELTA_BYTE_ARRAY => Ok(Box::new(DeltaByteArrayDecoder::new())),
                 _ => get_decoder_default(descr, encoding),
             }
@@ -1785,7 +1789,7 @@ mod tests {
             ],
             vec![f32::from_le_bytes([0xA3, 0xB4, 0xC5, 0xD6])],
         ];
-        test_byte_stream_split_decode::<FloatType>(data);
+        test_byte_stream_split_decode::<FloatType>(data, -1);
     }
 
     #[test]
@@ -1794,7 +1798,76 @@ mod tests {
             f64::from_le_bytes([0, 1, 2, 3, 4, 5, 6, 7]),
             f64::from_le_bytes([8, 9, 10, 11, 12, 13, 14, 15]),
         ]];
-        test_byte_stream_split_decode::<DoubleType>(data);
+        test_byte_stream_split_decode::<DoubleType>(data, -1);
+    }
+
+    #[test]
+    fn test_byte_stream_split_multiple_i32() {
+        let data = vec![
+            vec![
+                i32::from_le_bytes([0xAA, 0xBB, 0xCC, 0xDD]),
+                i32::from_le_bytes([0x00, 0x11, 0x22, 0x33]),
+            ],
+            vec![i32::from_le_bytes([0xA3, 0xB4, 0xC5, 0xD6])],
+        ];
+        test_byte_stream_split_decode::<Int32Type>(data, -1);
+    }
+
+    #[test]
+    fn test_byte_stream_split_i64() {
+        let data = vec![vec![
+            i64::from_le_bytes([0, 1, 2, 3, 4, 5, 6, 7]),
+            i64::from_le_bytes([8, 9, 10, 11, 12, 13, 14, 15]),
+        ]];
+        test_byte_stream_split_decode::<Int64Type>(data, -1);
+    }
+
+    fn test_byte_stream_split_flba(type_width: usize) {
+        let data = vec![
+            vec![
+                FixedLenByteArrayType::gen(type_width as i32),
+                FixedLenByteArrayType::gen(type_width as i32),
+            ],
+            vec![FixedLenByteArrayType::gen(type_width as i32)],
+        ];
+        test_byte_stream_split_decode::<FixedLenByteArrayType>(data, type_width as i32);
+    }
+
+    #[test]
+    fn test_byte_stream_split_flba5() {
+        test_byte_stream_split_flba(5);
+    }
+
+    #[test]
+    fn test_byte_stream_split_flba16() {
+        test_byte_stream_split_flba(16);
+    }
+
+    #[test]
+    fn test_byte_stream_split_flba19() {
+        test_byte_stream_split_flba(19);
+    }
+
+    #[test]
+    #[should_panic(expected = "Mismatched FixedLenByteArray sizes: 4 != 5")]
+    fn test_byte_stream_split_flba_mismatch() {
+        let data = vec![
+            vec![
+                FixedLenByteArray::from(vec![0xAA, 0xAB, 0xAC, 0xAD, 0xAE]),
+                FixedLenByteArray::from(vec![0xBA, 0xBB, 0xBC, 0xBD, 0xBE]),
+            ],
+            vec![FixedLenByteArray::from(vec![0xCA, 0xCB, 0xCC, 0xCD])],
+        ];
+        test_byte_stream_split_decode::<FixedLenByteArrayType>(data, 5);
+    }
+
+    #[test]
+    #[should_panic(expected = "Input data length is not a multiple of type width 4")]
+    fn test_byte_stream_split_flba_bad_input() {
+        let mut decoder = VariableWidthByteStreamSplitDecoder::<FixedLenByteArrayType>::new(4);
+        decoder
+            .set_data(Bytes::from(vec![1, 2, 3, 4, 5]), 1)
+            .unwrap();
     }
 
     #[test]
@@ -1808,33 +1881,42 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_skip_byte_stream_split_ints() {
+        let block_data = vec![3, 4, 1, 5];
+        test_skip::<Int32Type>(block_data.clone(), Encoding::BYTE_STREAM_SPLIT, 2);
+        test_skip::<Int64Type>(
+            block_data.into_iter().map(|x| x as i64).collect(),
+            Encoding::BYTE_STREAM_SPLIT,
+            100,
+        );
+    }
+
     fn test_rle_value_decode<T: DataType>(data: Vec<Vec<T::T>>) {
-        test_encode_decode::<T>(data, Encoding::RLE);
+        test_encode_decode::<T>(data, Encoding::RLE, -1);
     }
 
     fn test_delta_bit_packed_decode<T: DataType>(data: Vec<Vec<T::T>>) {
-        test_encode_decode::<T>(data, Encoding::DELTA_BINARY_PACKED);
+        test_encode_decode::<T>(data, Encoding::DELTA_BINARY_PACKED, -1);
     }
 
-    fn test_byte_stream_split_decode<T: DataType>(data: Vec<Vec<T::T>>) {
-        test_encode_decode::<T>(data, Encoding::BYTE_STREAM_SPLIT);
+    fn test_byte_stream_split_decode<T: DataType>(data: Vec<Vec<T::T>>, type_width: i32) {
+        test_encode_decode::<T>(data, Encoding::BYTE_STREAM_SPLIT, type_width);
     }
 
     fn test_delta_byte_array_decode(data: Vec<Vec<ByteArray>>) {
-        test_encode_decode::<ByteArrayType>(data, Encoding::DELTA_BYTE_ARRAY);
+        test_encode_decode::<ByteArrayType>(data, Encoding::DELTA_BYTE_ARRAY, -1);
     }
 
     // Input data represents vector of data slices to write (test multiple `put()` calls)
     // For example,
     //   vec![vec![1, 2, 3]] invokes `put()` once and writes {1, 2, 3}
     //   vec![vec![1, 2], vec![3]] invokes `put()` twice and writes {1, 2, 3}
-    fn test_encode_decode<T: DataType>(data: Vec<Vec<T::T>>, encoding: Encoding) {
-        // Type length should not really matter for encode/decode test,
-        // otherwise change it based on type
-        let col_descr = create_test_col_desc_ptr(-1, T::get_physical_type());
+    fn test_encode_decode<T: DataType>(data: Vec<Vec<T::T>>, encoding: Encoding, type_width: i32) {
+        let col_descr = create_test_col_desc_ptr(type_width, T::get_physical_type());
 
         // Encode data
-        let mut encoder = get_encoder::<T>(encoding).expect("get encoder");
+        let mut encoder = get_encoder::<T>(encoding, &col_descr).expect("get encoder");
 
         for v in &data[..] {
             encoder.put(&v[..]).expect("ok to encode");
@@ -1867,7 +1949,7 @@ mod tests {
         let col_descr = create_test_col_desc_ptr(-1, T::get_physical_type());
 
         // Encode data
-        let mut encoder = get_encoder::<T>(encoding).expect("get encoder");
+        let mut encoder = get_encoder::<T>(encoding, &col_descr).expect("get encoder");
 
         encoder.put(&data).expect("ok to encode");
 
diff --git a/parquet/src/encodings/decoding/byte_stream_split_decoder.rs b/parquet/src/encodings/decoding/byte_stream_split_decoder.rs
index 98841d21e..9b2f43ace 100644
--- a/parquet/src/encodings/decoding/byte_stream_split_decoder.rs
+++ b/parquet/src/encodings/decoding/byte_stream_split_decoder.rs
@@ -19,8 +19,9 @@ use std::marker::PhantomData;
 
 use bytes::Bytes;
 
-use crate::basic::Encoding;
-use crate::data_type::{DataType, SliceAsBytes};
+use crate::basic::{Encoding, Type};
+use crate::data_type::private::ParquetValueType;
+use crate::data_type::{DataType, FixedLenByteArray, SliceAsBytes};
 use crate::errors::{ParquetError, Result};
 
 use super::Decoder;
@@ -62,6 +63,22 @@ fn join_streams_const<const TYPE_SIZE: usize>(
     }
 }
 
+// Like the above, but type_size is not known at compile time.
+fn join_streams_variable(
+    src: &[u8],
+    dst: &mut [u8],
+    stride: usize,
+    type_size: usize,
+    values_decoded: usize,
+) {
+    let sub_src = &src[values_decoded..];
+    for i in 0..dst.len() / type_size {
+        for j in 0..type_size {
+            dst[i * type_size + j] = sub_src[i + j * stride];
+        }
+    }
+}
+
 impl<T: DataType> Decoder<T> for ByteStreamSplitDecoder<T> {
     fn set_data(&mut self, data: Bytes, num_values: usize) -> Result<()> {
         self.encoded_bytes = data;
@@ -76,7 +93,7 @@ impl<T: DataType> Decoder<T> for ByteStreamSplitDecoder<T> {
         let num_values = buffer.len().min(total_remaining_values);
         let buffer = &mut buffer[..num_values];
 
-        // SAFETY: f32 and f64 has no constraints on their internal representation, so we can modify it as we want
+        // SAFETY: i/f32 and i/f64 has no constraints on their internal representation, so we can modify it as we want
         let raw_out_bytes = unsafe { <T as DataType>::T::slice_as_bytes_mut(buffer) };
         let type_size = T::get_type_size();
         let stride = self.encoded_bytes.len() / type_size;
@@ -119,3 +136,126 @@ impl<T: DataType> Decoder<T> for ByteStreamSplitDecoder<T> {
         Ok(to_skip)
     }
 }
+
+pub struct VariableWidthByteStreamSplitDecoder<T: DataType> {
+    _phantom: PhantomData<T>,
+    encoded_bytes: Bytes,
+    total_num_values: usize,
+    values_decoded: usize,
+    type_width: usize,
+}
+
+impl<T: DataType> VariableWidthByteStreamSplitDecoder<T> {
+    pub(crate) fn new(type_length: i32) -> Self {
+        Self {
+            _phantom: PhantomData,
+            encoded_bytes: Bytes::new(),
+            total_num_values: 0,
+            values_decoded: 0,
+            type_width: type_length as usize,
+        }
+    }
+}
+
+impl<T: DataType> Decoder<T> for VariableWidthByteStreamSplitDecoder<T> {
+    fn set_data(&mut self, data: Bytes, num_values: usize) -> Result<()> {
+        // Rough check that all data elements are the same length
+        if data.len() % self.type_width != 0 {
+            return Err(general_err!(
+                "Input data length is not a multiple of type width {}",
+                self.type_width
+            ));
+        }
+
+        match T::get_physical_type() {
+            Type::FIXED_LEN_BYTE_ARRAY => {
+                self.encoded_bytes = data;
+                self.total_num_values = num_values;
+                self.values_decoded = 0;
+                Ok(())
+            }
+            _ => Err(general_err!(
+                "VariableWidthByteStreamSplitDecoder only supports FixedLenByteArrayType"
+            )),
+        }
+    }
+
+    fn get(&mut self, buffer: &mut [<T as DataType>::T]) -> Result<usize> {
+        let total_remaining_values = self.values_left();
+        let num_values = buffer.len().min(total_remaining_values);
+        let buffer = &mut buffer[..num_values];
+        let type_size = self.type_width;
+
+        // Since this is FIXED_LEN_BYTE_ARRAY data, we can't use slice_as_bytes_mut. Instead we'll
+        // have to do some data copies.
+        let mut tmp_vec = vec![0_u8; num_values * type_size];
+        let raw_out_bytes = tmp_vec.as_mut_slice();
+
+        let stride = self.encoded_bytes.len() / type_size;
+        match type_size {
+            2 => join_streams_const::<2>(
+                &self.encoded_bytes,
+                raw_out_bytes,
+                stride,
+                self.values_decoded,
+            ),
+            4 => join_streams_const::<4>(
+                &self.encoded_bytes,
+                raw_out_bytes,
+                stride,
+                self.values_decoded,
+            ),
+            8 => join_streams_const::<8>(
+                &self.encoded_bytes,
+                raw_out_bytes,
+                stride,
+                self.values_decoded,
+            ),
+            16 => join_streams_const::<16>(
+                &self.encoded_bytes,
+                raw_out_bytes,
+                stride,
+                self.values_decoded,
+            ),
+            _ => join_streams_variable(
+                &self.encoded_bytes,
+                raw_out_bytes,
+                stride,
+                type_size,
+                self.values_decoded,
+            ),
+        }
+        self.values_decoded += num_values;
+
+        // create a buffer from the vec so far (and leave a new Vec in its place)
+        let vec_with_data = std::mem::take(&mut tmp_vec);
+        // convert Vec to Bytes (which is a ref counted wrapper)
+        let bytes_with_data = Bytes::from(vec_with_data);
+        for (i, bi) in buffer.iter_mut().enumerate().take(num_values) {
+            // Get a view into the data, without also copying the bytes
+            let data = bytes_with_data.slice(i * type_size..(i + 1) * type_size);
+            // TODO: perhaps add a `set_from_bytes` method to `DataType` to avoid downcasting
+            let bi = bi
+                .as_mut_any()
+                .downcast_mut::<FixedLenByteArray>()
+                .expect("Decoding fixed length byte array");
+            bi.set_data(data);
+        }
+
+        Ok(num_values)
+    }
+
+    fn values_left(&self) -> usize {
+        self.total_num_values - self.values_decoded
+    }
+
+    fn encoding(&self) -> Encoding {
+        Encoding::BYTE_STREAM_SPLIT
+    }
+
+    fn skip(&mut self, num_values: usize) -> Result<usize> {
+        let to_skip = usize::min(self.values_left(), num_values);
+        self.values_decoded += to_skip;
+        Ok(to_skip)
+    }
+}
diff --git a/parquet/src/encodings/encoding/byte_stream_split_encoder.rs b/parquet/src/encodings/encoding/byte_stream_split_encoder.rs
index 3d5ba4cc2..0726c6b1c 100644
--- a/parquet/src/encodings/encoding/byte_stream_split_encoder.rs
+++ b/parquet/src/encodings/encoding/byte_stream_split_encoder.rs
@@ -16,14 +16,14 @@
 // under the License.
 
 use crate::basic::{Encoding, Type};
-use crate::data_type::DataType;
-use crate::data_type::SliceAsBytes;
+use crate::data_type::{AsBytes, DataType, SliceAsBytes};
 
 use crate::errors::{ParquetError, Result};
 
 use super::Encoder;
 
-use bytes::Bytes;
+use bytes::{BufMut, Bytes};
+use std::cmp;
 use std::marker::PhantomData;
 
 pub struct ByteStreamSplitEncoder<T> {
@@ -40,7 +40,7 @@ impl<T: DataType> ByteStreamSplitEncoder<T> {
     }
 }
 
-// Here we assume src contains the full data (which it must, since we're
+// Here we assume src contains the full data (which it must, since we
 // can only know where to split the streams once all data is collected).
 // We iterate over the input bytes and write them to their strided output
 // byte locations.
@@ -53,13 +53,28 @@ fn split_streams_const<const TYPE_SIZE: usize>(src: &[u8], dst: &mut [u8]) {
     }
 }
 
+// Like above, but type_size is not known at compile time.
+fn split_streams_variable(src: &[u8], dst: &mut [u8], type_size: usize) {
+    const BLOCK_SIZE: usize = 4;
+    let stride = src.len() / type_size;
+    for j in (0..type_size).step_by(BLOCK_SIZE) {
+        let jrange = cmp::min(BLOCK_SIZE, type_size - j);
+        for i in 0..stride {
+            for jj in 0..jrange {
+                dst[i + (j + jj) * stride] = src[i * type_size + j + jj];
+            }
+        }
+    }
+}
+
 impl<T: DataType> Encoder<T> for ByteStreamSplitEncoder<T> {
     fn put(&mut self, values: &[T::T]) -> Result<()> {
         self.buffer
             .extend(<T as DataType>::T::slice_as_bytes(values));
+
         ensure_phys_ty!(
-            Type::FLOAT | Type::DOUBLE,
-            "ByteStreamSplitEncoder only supports FloatType or DoubleType"
+            Type::FLOAT | Type::DOUBLE | Type::INT32 | Type::INT64,
+            "ByteStreamSplitEncoder does not support Int96, Boolean, or ByteArray types"
         );
 
         Ok(())
@@ -96,3 +111,121 @@ impl<T: DataType> Encoder<T> for ByteStreamSplitEncoder<T> {
         self.buffer.capacity() * std::mem::size_of::<u8>()
     }
 }
+
+pub struct VariableWidthByteStreamSplitEncoder<T> {
+    buffer: Vec<u8>,
+    type_width: usize,
+    _p: PhantomData<T>,
+}
+
+impl<T: DataType> VariableWidthByteStreamSplitEncoder<T> {
+    pub(crate) fn new(type_length: i32) -> Self {
+        Self {
+            buffer: Vec::new(),
+            type_width: type_length as usize,
+            _p: PhantomData,
+        }
+    }
+}
+
+fn put_fixed<T: DataType, const TYPE_SIZE: usize>(dst: &mut [u8], values: &[T::T]) {
+    let mut idx = 0;
+    values.iter().for_each(|x| {
+        let bytes = x.as_bytes();
+        if bytes.len() != TYPE_SIZE {
+            panic!(
+                "Mismatched FixedLenByteArray sizes: {} != {}",
+                bytes.len(),
+                TYPE_SIZE
+            );
+        }
+        dst[idx..(TYPE_SIZE + idx)].copy_from_slice(&bytes[..TYPE_SIZE]);
+        idx += TYPE_SIZE;
+    });
+}
+
+fn put_variable<T: DataType>(dst: &mut [u8], values: &[T::T], type_width: usize) {
+    let mut idx = 0;
+    values.iter().for_each(|x| {
+        let bytes = x.as_bytes();
+        if bytes.len() != type_width {
+            panic!(
+                "Mismatched FixedLenByteArray sizes: {} != {}",
+                bytes.len(),
+                type_width
+            );
+        }
+        dst[idx..idx + type_width].copy_from_slice(bytes);
+        idx += type_width;
+    });
+}
+
+impl<T: DataType> Encoder<T> for VariableWidthByteStreamSplitEncoder<T> {
+    fn put(&mut self, values: &[T::T]) -> Result<()> {
+        ensure_phys_ty!(
+            Type::FIXED_LEN_BYTE_ARRAY,
+            "VariableWidthByteStreamSplitEncoder only supports FixedLenByteArray types"
+        );
+
+        // FixedLenByteArray is implemented as ByteArray, so there may be gaps making
+        // slice_as_bytes untenable
+        let idx = self.buffer.len();
+        let data_len = values.len() * self.type_width;
+        // Ensure enough capacity for the new data
+        self.buffer.reserve(values.len() * self.type_width);
+        // ...and extend the size of buffer to allow direct access
+        self.buffer.put_bytes(0_u8, data_len);
+        // Get a slice of the buffer corresponding to the location of the new data
+        let out_buf = &mut self.buffer[idx..idx + data_len];
+
+        // Now copy `values` into the buffer. For `type_width` <= 8 use a fixed size when
+        // performing the copy as it is significantly faster.
+        match self.type_width {
+            2 => put_fixed::<T, 2>(out_buf, values),
+            3 => put_fixed::<T, 3>(out_buf, values),
+            4 => put_fixed::<T, 4>(out_buf, values),
+            5 => put_fixed::<T, 5>(out_buf, values),
+            6 => put_fixed::<T, 6>(out_buf, values),
+            7 => put_fixed::<T, 7>(out_buf, values),
+            8 => put_fixed::<T, 8>(out_buf, values),
+            _ => put_variable::<T>(out_buf, values, self.type_width),
+        }
+
+        Ok(())
+    }
+
+    fn encoding(&self) -> Encoding {
+        Encoding::BYTE_STREAM_SPLIT
+    }
+
+    fn estimated_data_encoded_size(&self) -> usize {
+        self.buffer.len()
+    }
+
+    fn flush_buffer(&mut self) -> Result<Bytes> {
+        let mut encoded = vec![0; self.buffer.len()];
+        let type_size = match T::get_physical_type() {
+            Type::FIXED_LEN_BYTE_ARRAY => self.type_width,
+            _ => T::get_type_size(),
+        };
+        // split_streams_const() is faster up to type_width == 8
+        match type_size {
+            2 => split_streams_const::<2>(&self.buffer, &mut encoded),
+            3 => split_streams_const::<3>(&self.buffer, &mut encoded),
+            4 => split_streams_const::<4>(&self.buffer, &mut encoded),
+            5 => split_streams_const::<5>(&self.buffer, &mut encoded),
+            6 => split_streams_const::<6>(&self.buffer, &mut encoded),
+            7 => split_streams_const::<7>(&self.buffer, &mut encoded),
+            8 => split_streams_const::<8>(&self.buffer, &mut encoded),
+            _ => split_streams_variable(&self.buffer, &mut encoded, type_size),
+        }
+
+        self.buffer.clear();
+        Ok(encoded.into())
+    }
+
+    /// return the estimated memory size of this encoder.
+    fn estimated_memory_size(&self) -> usize {
+        self.buffer.capacity() * std::mem::size_of::<u8>()
+    }
+}
diff --git a/parquet/src/encodings/encoding/mod.rs b/parquet/src/encodings/encoding/mod.rs
index ecc376712..f6d94e033 100644
--- a/parquet/src/encodings/encoding/mod.rs
+++ b/parquet/src/encodings/encoding/mod.rs
@@ -24,8 +24,10 @@ use crate::data_type::private::ParquetValueType;
 use crate::data_type::*;
 use crate::encodings::rle::RleEncoder;
 use crate::errors::{ParquetError, Result};
+use crate::schema::types::ColumnDescPtr;
 use crate::util::bit_util::{num_required_bits, BitWriter};
 
+use byte_stream_split_encoder::{ByteStreamSplitEncoder, VariableWidthByteStreamSplitEncoder};
 use bytes::Bytes;
 pub use dict_encoder::DictEncoder;
 
@@ -78,7 +80,10 @@ pub trait Encoder<T: DataType>: Send {
 
 /// Gets a encoder for the particular data type `T` and encoding `encoding`. Memory usage
 /// for the encoder instance is tracked by `mem_tracker`.
-pub fn get_encoder<T: DataType>(encoding: Encoding) -> Result<Box<dyn Encoder<T>>> {
+pub fn get_encoder<T: DataType>(
+    encoding: Encoding,
+    descr: &ColumnDescPtr,
+) -> Result<Box<dyn Encoder<T>>> {
     let encoder: Box<dyn Encoder<T>> = match encoding {
         Encoding::PLAIN => Box::new(PlainEncoder::new()),
         Encoding::RLE_DICTIONARY | Encoding::PLAIN_DICTIONARY => {
@@ -90,9 +95,12 @@ pub fn get_encoder<T: DataType>(encoding: Encoding) -> Result<Box<dyn Encoder<T>
         Encoding::DELTA_BINARY_PACKED => Box::new(DeltaBitPackEncoder::new()),
         Encoding::DELTA_LENGTH_BYTE_ARRAY => Box::new(DeltaLengthByteArrayEncoder::new()),
         Encoding::DELTA_BYTE_ARRAY => Box::new(DeltaByteArrayEncoder::new()),
-        Encoding::BYTE_STREAM_SPLIT => {
-            Box::new(byte_stream_split_encoder::ByteStreamSplitEncoder::new())
-        }
+        Encoding::BYTE_STREAM_SPLIT => match T::get_physical_type() {
+            Type::FIXED_LEN_BYTE_ARRAY => Box::new(VariableWidthByteStreamSplitEncoder::new(
+                descr.type_length(),
+            )),
+            _ => Box::new(ByteStreamSplitEncoder::new()),
+        },
         e => return Err(nyi_err!("Encoding {} is not supported", e)),
     };
     Ok(encoder)
@@ -498,8 +506,7 @@ impl<T: DataType> Encoder<T> for DeltaBitPackEncoder<T> {
         self.page_header_writer.estimated_memory_size()
             + self.bit_writer.estimated_memory_size()
             + self.deltas.capacity() * std::mem::size_of::<i64>()
-        + std::mem::size_of::<Self>()
-
+            + std::mem::size_of::<Self>()
     }
 }
 
@@ -641,10 +648,7 @@ impl<T: DataType> Encoder<T> for DeltaLengthByteArrayEncoder<T> {
 
     /// return the estimated memory size of this encoder.
     fn estimated_memory_size(&self) -> usize {
-        self.len_encoder.estimated_memory_size()
-            + self.data.len()
-        + std::mem::size_of::<Self>()
-
+        self.len_encoder.estimated_memory_size() + self.data.len() + std::mem::size_of::<Self>()
     }
 }
 
@@ -754,8 +758,7 @@ impl<T: DataType> Encoder<T> for DeltaByteArrayEncoder<T> {
     fn estimated_memory_size(&self) -> usize {
         self.prefix_len_encoder.estimated_memory_size()
             + self.suffix_writer.estimated_memory_size()
-        + (self.previous.capacity() * std::mem::size_of::<u8>())
-
+            + (self.previous.capacity() * std::mem::size_of::<u8>())
     }
 }
 
@@ -767,28 +770,30 @@ mod tests {
 
     use crate::encodings::decoding::{get_decoder, Decoder, DictDecoder, PlainDecoder};
     use crate::schema::types::{ColumnDescPtr, ColumnDescriptor, ColumnPath, Type as SchemaType};
-    use crate::util::test_common::rand_gen::{random_bytes, RandGen};
     use crate::util::bit_util;
+    use crate::util::test_common::rand_gen::{random_bytes, RandGen};
 
     const TEST_SET_SIZE: usize = 1024;
 
     #[test]
     fn test_get_encoders() {
         // supported encodings
-        create_and_check_encoder::<Int32Type>(Encoding::PLAIN, None);
-        create_and_check_encoder::<Int32Type>(Encoding::DELTA_BINARY_PACKED, None);
-        create_and_check_encoder::<Int32Type>(Encoding::DELTA_LENGTH_BYTE_ARRAY, None);
-        create_and_check_encoder::<Int32Type>(Encoding::DELTA_BYTE_ARRAY, None);
-        create_and_check_encoder::<BoolType>(Encoding::RLE, None);
+        create_and_check_encoder::<Int32Type>(0, Encoding::PLAIN, None);
+        create_and_check_encoder::<Int32Type>(0, Encoding::DELTA_BINARY_PACKED, None);
+        create_and_check_encoder::<Int32Type>(0, Encoding::DELTA_LENGTH_BYTE_ARRAY, None);
+        create_and_check_encoder::<Int32Type>(0, Encoding::DELTA_BYTE_ARRAY, None);
+        create_and_check_encoder::<BoolType>(0, Encoding::RLE, None);
 
         // error when initializing
         create_and_check_encoder::<Int32Type>(
+            0,
             Encoding::RLE_DICTIONARY,
             Some(general_err!(
                 "Cannot initialize this encoding through this function"
             )),
         );
         create_and_check_encoder::<Int32Type>(
+            0,
             Encoding::PLAIN_DICTIONARY,
             Some(general_err!(
                 "Cannot initialize this encoding through this function"
@@ -798,6 +803,7 @@ mod tests {
         // unsupported
         #[allow(deprecated)]
         create_and_check_encoder::<Int32Type>(
+            0,
             Encoding::BIT_PACKED,
             Some(nyi_err!("Encoding BIT_PACKED is not supported")),
         );
@@ -815,6 +821,7 @@ mod tests {
         Int32Type::test(Encoding::PLAIN, TEST_SET_SIZE, -1);
         Int32Type::test(Encoding::PLAIN_DICTIONARY, TEST_SET_SIZE, -1);
         Int32Type::test(Encoding::DELTA_BINARY_PACKED, TEST_SET_SIZE, -1);
+        Int32Type::test(Encoding::BYTE_STREAM_SPLIT, TEST_SET_SIZE, -1);
     }
 
     #[test]
@@ -822,6 +829,7 @@ mod tests {
         Int64Type::test(Encoding::PLAIN, TEST_SET_SIZE, -1);
         Int64Type::test(Encoding::PLAIN_DICTIONARY, TEST_SET_SIZE, -1);
         Int64Type::test(Encoding::DELTA_BINARY_PACKED, TEST_SET_SIZE, -1);
+        Int64Type::test(Encoding::BYTE_STREAM_SPLIT, TEST_SET_SIZE, -1);
     }
 
     #[test]
@@ -853,10 +861,11 @@ mod tests {
     }
 
     #[test]
-    fn test_fixed_lenbyte_array() {
+    fn test_fixed_len_byte_array() {
         FixedLenByteArrayType::test(Encoding::PLAIN, TEST_SET_SIZE, 100);
         FixedLenByteArrayType::test(Encoding::PLAIN_DICTIONARY, TEST_SET_SIZE, 100);
         FixedLenByteArrayType::test(Encoding::DELTA_BYTE_ARRAY, TEST_SET_SIZE, 100);
+        FixedLenByteArrayType::test(Encoding::BYTE_STREAM_SPLIT, TEST_SET_SIZE, 100);
     }
 
     #[test]
@@ -905,7 +914,7 @@ mod tests {
                 Encoding::PLAIN_DICTIONARY | Encoding::RLE_DICTIONARY => {
                     Box::new(create_test_dict_encoder::<T>(type_length))
                 }
-                _ => create_test_encoder::<T>(encoding),
+                _ => create_test_encoder::<T>(type_length, encoding),
             };
             assert_eq!(encoder.estimated_data_encoded_size(), initial_size);
 
@@ -960,7 +969,7 @@ mod tests {
     #[test]
     fn test_byte_stream_split_example_f32() {
         // Test data from https://github.com/apache/parquet-format/blob/2a481fe1aad64ff770e21734533bb7ef5a057dac/Encodings.md#byte-stream-split-byte_stream_split--9
-        let mut encoder = create_test_encoder::<FloatType>(Encoding::BYTE_STREAM_SPLIT);
+        let mut encoder = create_test_encoder::<FloatType>(0, Encoding::BYTE_STREAM_SPLIT);
         let mut decoder = create_test_decoder::<FloatType>(0, Encoding::BYTE_STREAM_SPLIT);
 
         let input = vec![
@@ -989,7 +998,7 @@ mod tests {
     // See: https://github.com/sunchao/parquet-rs/issues/47
     #[test]
     fn test_issue_47() {
-        let mut encoder = create_test_encoder::<ByteArrayType>(Encoding::DELTA_BYTE_ARRAY);
+        let mut encoder = create_test_encoder::<ByteArrayType>(0, Encoding::DELTA_BYTE_ARRAY);
         let mut decoder = create_test_decoder::<ByteArrayType>(0, Encoding::DELTA_BYTE_ARRAY);
 
         let input = vec![
@@ -1039,7 +1048,7 @@ mod tests {
 
     impl<T: DataType + RandGen<T>> EncodingTester<T> for T {
         fn test_internal(enc: Encoding, total: usize, type_length: i32) -> Result<()> {
-            let mut encoder = create_test_encoder::<T>(enc);
+            let mut encoder = create_test_encoder::<T>(type_length, enc);
             let mut decoder = create_test_decoder::<T>(type_length, enc);
             let mut values = <T as RandGen<T>>::gen_vec(type_length, total);
             let mut result_data = vec![T::T::default(); total];
@@ -1137,8 +1146,13 @@ mod tests {
         decoder.get(output)
     }
 
-    fn create_and_check_encoder<T: DataType>(encoding: Encoding, err: Option<ParquetError>) {
-        let encoder = get_encoder::<T>(encoding);
+    fn create_and_check_encoder<T: DataType>(
+        type_length: i32,
+        encoding: Encoding,
+        err: Option<ParquetError>,
+    ) {
+        let desc = create_test_col_desc_ptr(type_length, T::get_physical_type());
+        let encoder = get_encoder::<T>(encoding, &desc);
         match err {
             Some(parquet_error) => {
                 assert_eq!(
@@ -1164,8 +1178,9 @@ mod tests {
         ))
     }
 
-    fn create_test_encoder<T: DataType>(enc: Encoding) -> Box<dyn Encoder<T>> {
-        get_encoder(enc).unwrap()
+    fn create_test_encoder<T: DataType>(type_len: i32, enc: Encoding) -> Box<dyn Encoder<T>> {
+        let desc = create_test_col_desc_ptr(type_len, T::get_physical_type());
+        get_encoder(enc, &desc).unwrap()
     }
 
     fn create_test_decoder<T: DataType>(type_len: i32, enc: Encoding) -> Box<dyn Decoder<T>> {
diff --git a/parquet/src/file/metadata/mod.rs b/parquet/src/file/metadata/mod.rs
index 86c673bbd..45ef0c546 100644
--- a/parquet/src/file/metadata/mod.rs
+++ b/parquet/src/file/metadata/mod.rs
@@ -33,6 +33,67 @@
 //! * [`ColumnChunkMetaData`]: Metadata for each column chunk (primitive leaf)
 //!   within a Row Group including encoding and compression information,
 //!   number of values, statistics, etc.
+//!
+//! # APIs for working with Parquet Metadata
+//!
+//! The Parquet readers and writers in this crate read and write
+//! metadata into parquet files. To work with metadata directly,
+//! the following APIs are available.
+//!
+//! Reading:
+//! * Read from bytes to `ParquetMetaData`: [`decode_footer`]
+//!   and [`decode_metadata`]
+//! * Read from an `async` source to `ParquetMetadata`: [`MetadataLoader`]
+//!
+//! [`MetadataLoader`]: https://docs.rs/parquet/latest/parquet/arrow/async_reader/struct.MetadataLoader.html
+//! [`decode_footer`]: crate::file::footer::decode_footer
+//! [`decode_metadata`]: crate::file::footer::decode_metadata
+//!
+//! Writing:
+//! * Write `ParquetMetaData` to bytes in memory: Not yet supported (see [#6002])
+//! * Writes `ParquetMetaData` to an async target: Not yet supported
+//!
+//! [#6002]: https://github.com/apache/arrow-rs/issues/6002
+//!
+//! # Metadata Encodings and Structures
+//!
+//! There are three different encodings of Parquet Metadata in this crate:
+//!
+//! 1. `bytes`:encoded with the Thrift TCompactProtocol as defined in
+//!    [parquet.thrift]
+//!
+//! 2. [`format`]: Rust structures automatically generated by the thrift compiler
+//!    from [parquet.thrift]. These structures are low level and mirror
+//!    the thrift definitions.
+//!
+//! 3. [`file::metadata`] (this module): Easier to use Rust structures
+//!    with a more idiomatic API. Note that, confusingly, some but not all
+//!    of these structures have the same name as the [`format`] structures.
+//!
+//! [`format`]: crate::format
+//! [`file::metadata`]: crate::file::metadata
+//! [parquet.thrift]:  https://github.com/apache/parquet-format/blob/master/src/main/thrift/parquet.thrift
+//!
+//! Graphically, this is how the different structures relate to each other:
+//!
+//! ```text
+//!                          ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─         ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─
+//!                            ┌──────────────┐     │         ┌───────────────────────┐ │
+//!                          │ │ ColumnIndex  │              ││    ParquetMetaData    │
+//!                            └──────────────┘     │         └───────────────────────┘ │
+//! ┌──────────────┐         │ ┌────────────────┐            │┌───────────────────────┐
+//! │   ..0x24..   │ ◀────▶    │  OffsetIndex   │   │ ◀────▶  │    ParquetMetaData    │ │
+//! └──────────────┘         │ └────────────────┘            │└───────────────────────┘
+//!                                     ...         │                   ...             │
+//!                          │ ┌──────────────────┐          │ ┌──────────────────┐
+//! bytes                      │  FileMetaData*   │ │          │  FileMetaData*   │     │
+//! (thrift encoded)         │ └──────────────────┘          │ └──────────────────┘
+//!                           ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┘         ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┘
+//!
+//!                          format::meta structures          file::metadata structures
+//!
+//!                         * Same name, different struct
+//! ```
 mod memory;
 
 use std::ops::Range;
diff --git a/parquet/src/file/page_index/index.rs b/parquet/src/file/page_index/index.rs
index cebb602b3..0c23e4aa3 100644
--- a/parquet/src/file/page_index/index.rs
+++ b/parquet/src/file/page_index/index.rs
@@ -225,6 +225,53 @@ impl<T: ParquetValueType> NativeIndex<T> {
             boundary_order: index.boundary_order,
         })
     }
+
+    pub(crate) fn to_thrift(&self) -> ColumnIndex {
+        let min_values = self
+            .indexes
+            .iter()
+            .map(|x| x.min_bytes().map(|x| x.to_vec()))
+            .collect::<Option<Vec<_>>>()
+            .unwrap_or_else(|| vec![vec![]; self.indexes.len()]);
+
+        let max_values = self
+            .indexes
+            .iter()
+            .map(|x| x.max_bytes().map(|x| x.to_vec()))
+            .collect::<Option<Vec<_>>>()
+            .unwrap_or_else(|| vec![vec![]; self.indexes.len()]);
+
+        let null_counts = self
+            .indexes
+            .iter()
+            .map(|x| x.null_count())
+            .collect::<Option<Vec<_>>>();
+
+        // Concatenate page histograms into a single Option<Vec>
+        let repetition_level_histograms = self
+            .indexes
+            .iter()
+            .map(|x| x.repetition_level_histogram().map(|v| v.values()))
+            .collect::<Option<Vec<&[i64]>>>()
+            .map(|hists| hists.concat());
+
+        let definition_level_histograms = self
+            .indexes
+            .iter()
+            .map(|x| x.definition_level_histogram().map(|v| v.values()))
+            .collect::<Option<Vec<&[i64]>>>()
+            .map(|hists| hists.concat());
+
+        ColumnIndex::new(
+            self.indexes.iter().map(|x| x.min().is_none()).collect(),
+            min_values,
+            max_values,
+            self.boundary_order,
+            null_counts,
+            repetition_level_histograms,
+            definition_level_histograms,
+        )
+    }
 }
 
 #[cfg(test)]
diff --git a/parquet/src/file/serialized_reader.rs b/parquet/src/file/serialized_reader.rs
index 70aea6fd5..07e80f7f6 100644
--- a/parquet/src/file/serialized_reader.rs
+++ b/parquet/src/file/serialized_reader.rs
@@ -1765,4 +1765,42 @@ mod tests {
             _ => unreachable!(),
         }
     }
+
+    #[test]
+    fn test_byte_stream_split_extended() {
+        let path = format!(
+            "{}/byte_stream_split_extended.gzip.parquet",
+            arrow::util::test_util::parquet_test_data(),
+        );
+        let file = File::open(path).unwrap();
+        let reader = Box::new(SerializedFileReader::new(file).expect("Failed to create reader"));
+
+        // Use full schema as projected schema
+        let mut iter = reader
+            .get_row_iter(None)
+            .expect("Failed to create row iterator");
+
+        let mut start = 0;
+        let end = reader.metadata().file_metadata().num_rows();
+
+        let check_row = |row: Result<Row, ParquetError>| {
+            assert!(row.is_ok());
+            let r = row.unwrap();
+            assert_eq!(r.get_float16(0).unwrap(), r.get_float16(1).unwrap());
+            assert_eq!(r.get_float(2).unwrap(), r.get_float(3).unwrap());
+            assert_eq!(r.get_double(4).unwrap(), r.get_double(5).unwrap());
+            assert_eq!(r.get_int(6).unwrap(), r.get_int(7).unwrap());
+            assert_eq!(r.get_long(8).unwrap(), r.get_long(9).unwrap());
+            assert_eq!(r.get_bytes(10).unwrap(), r.get_bytes(11).unwrap());
+            assert_eq!(r.get_decimal(12).unwrap(), r.get_decimal(13).unwrap());
+        };
+
+        while start < end {
+            match iter.next() {
+                Some(row) => check_row(row),
+                None => break,
+            };
+            start += 1;
+        }
+    }
 }
diff --git a/parquet/src/file/writer.rs b/parquet/src/file/writer.rs
index bcf8449b4..c84d06a2c 100644
--- a/parquet/src/file/writer.rs
+++ b/parquet/src/file/writer.rs
@@ -19,6 +19,7 @@
 //! using row group writers and column writers respectively.
 
 use crate::bloom_filter::Sbbf;
+use crate::file::page_index::index::Index;
 use crate::format as parquet;
 use crate::format::{ColumnIndex, OffsetIndex, RowGroup};
 use crate::thrift::TSerializable;
@@ -261,122 +262,41 @@ impl<W: Write + Send> SerializedFileWriter<W> {
         Ok(())
     }
 
-    /// Serialize all the offset index to the file
-    fn write_offset_indexes(&mut self, row_groups: &mut [RowGroup]) -> Result<()> {
-        // iter row group
-        // iter each column
-        // write offset index to the file
-        for (row_group_idx, row_group) in row_groups.iter_mut().enumerate() {
-            for (column_idx, column_metadata) in row_group.columns.iter_mut().enumerate() {
-                match &self.offset_indexes[row_group_idx][column_idx] {
-                    Some(offset_index) => {
-                        let start_offset = self.buf.bytes_written();
-                        let mut protocol = TCompactOutputProtocol::new(&mut self.buf);
-                        offset_index.write_to_out_protocol(&mut protocol)?;
-                        let end_offset = self.buf.bytes_written();
-                        // set offset and index for offset index
-                        column_metadata.offset_index_offset = Some(start_offset as i64);
-                        column_metadata.offset_index_length =
-                            Some((end_offset - start_offset) as i32);
-                    }
-                    None => {}
-                }
-            }
-        }
-        Ok(())
-    }
-
-    /// Serialize all the column index to the file
-    fn write_column_indexes(&mut self, row_groups: &mut [RowGroup]) -> Result<()> {
-        // iter row group
-        // iter each column
-        // write column index to the file
-        for (row_group_idx, row_group) in row_groups.iter_mut().enumerate() {
-            for (column_idx, column_metadata) in row_group.columns.iter_mut().enumerate() {
-                match &self.column_indexes[row_group_idx][column_idx] {
-                    Some(column_index) => {
-                        let start_offset = self.buf.bytes_written();
-                        let mut protocol = TCompactOutputProtocol::new(&mut self.buf);
-                        column_index.write_to_out_protocol(&mut protocol)?;
-                        let end_offset = self.buf.bytes_written();
-                        // set offset and index for offset index
-                        column_metadata.column_index_offset = Some(start_offset as i64);
-                        column_metadata.column_index_length =
-                            Some((end_offset - start_offset) as i32);
-                    }
-                    None => {}
-                }
-            }
-        }
-        Ok(())
-    }
-
     /// Assembles and writes metadata at the end of the file.
     fn write_metadata(&mut self) -> Result<parquet::FileMetaData> {
         self.finished = true;
-        let num_rows = self.row_groups.iter().map(|x| x.num_rows()).sum();
 
         // write out any remaining bloom filters after all row groups
         for row_group in &mut self.row_groups {
             write_bloom_filters(&mut self.buf, &mut self.bloom_filters, row_group)?;
         }
 
-        let mut row_groups = self
-            .row_groups
-            .as_slice()
-            .iter()
-            .map(|v| v.to_thrift())
-            .collect::<Vec<_>>();
-
-        // Write column indexes and offset indexes
-        self.write_column_indexes(&mut row_groups)?;
-        self.write_offset_indexes(&mut row_groups)?;
-
         let key_value_metadata = match self.props.key_value_metadata() {
             Some(kv) => Some(kv.iter().chain(&self.kv_metadatas).cloned().collect()),
             None if self.kv_metadatas.is_empty() => None,
             None => Some(self.kv_metadatas.clone()),
         };
 
-        // We only include ColumnOrder for leaf nodes.
-        // Currently only supported ColumnOrder is TypeDefinedOrder so we set this
-        // for all leaf nodes.
-        // Even if the column has an undefined sort order, such as INTERVAL, this
-        // is still technically the defined TYPEORDER so it should still be set.
-        let column_orders = (0..self.schema_descr().num_columns())
-            .map(|_| parquet::ColumnOrder::TYPEORDER(parquet::TypeDefinedOrder {}))
-            .collect();
-        // This field is optional, perhaps in cases where no min/max fields are set
-        // in any Statistics or ColumnIndex object in the whole file.
-        // But for simplicity we always set this field.
-        let column_orders = Some(column_orders);
+        let row_groups = self
+            .row_groups
+            .iter()
+            .map(|v| v.to_thrift())
+            .collect::<Vec<_>>();
 
-        let file_metadata = parquet::FileMetaData {
-            num_rows,
+        let mut encoder = ThriftMetadataWriter::new(
+            &mut self.buf,
+            &self.schema,
+            &self.descr,
             row_groups,
-            key_value_metadata,
-            version: self.props.writer_version().as_num(),
-            schema: types::to_thrift(self.schema.as_ref())?,
-            created_by: Some(self.props.created_by().to_owned()),
-            column_orders,
-            encryption_algorithm: None,
-            footer_signing_key_metadata: None,
-        };
-
-        // Write file metadata
-        let start_pos = self.buf.bytes_written();
-        {
-            let mut protocol = TCompactOutputProtocol::new(&mut self.buf);
-            file_metadata.write_to_out_protocol(&mut protocol)?;
+            Some(self.props.created_by().to_string()),
+            self.props.writer_version().as_num(),
+        );
+        if let Some(key_value_metadata) = key_value_metadata {
+            encoder = encoder.with_key_value_metadata(key_value_metadata)
         }
-        let end_pos = self.buf.bytes_written();
-
-        // Write footer
-        let metadata_len = (end_pos - start_pos) as u32;
-
-        self.buf.write_all(&metadata_len.to_le_bytes())?;
-        self.buf.write_all(&PARQUET_MAGIC)?;
-        Ok(file_metadata)
+        encoder = encoder.with_column_indexes(&self.column_indexes);
+        encoder = encoder.with_offset_indexes(&self.offset_indexes);
+        encoder.finish()
     }
 
     #[inline]
@@ -810,13 +730,294 @@ impl<'a, W: Write + Send> PageWriter for SerializedPageWriter<'a, W> {
     }
 }
 
+/// Writes `crate::file::metadata` structures to a thrift encdoded byte streams
+///
+/// This structure handles the details of writing the various parts of parquet
+/// metadata into a byte stream. It is used to write the metadata into a
+/// parquet file and can also write metadata into other locations (such as a
+/// store of bytes).
+///
+/// This is somewhat trickey because the metadata is not store as a single inline
+/// thrift struture. It can have several "out of band" structures such as the OffsetIndex
+/// and BloomFilters which are stored separately whose locations are stored as offsets
+struct ThriftMetadataWriter<'a, W: Write> {
+    buf: &'a mut TrackedWrite<W>,
+    schema: &'a TypePtr,
+    schema_descr: &'a SchemaDescPtr,
+    row_groups: Vec<RowGroup>,
+    column_indexes: Option<&'a [Vec<Option<ColumnIndex>>]>,
+    offset_indexes: Option<&'a [Vec<Option<OffsetIndex>>]>,
+    key_value_metadata: Option<Vec<KeyValue>>,
+    created_by: Option<String>,
+    writer_version: i32,
+}
+
+impl<'a, W: Write> ThriftMetadataWriter<'a, W> {
+    /// Serialize all the offset index to the file
+    fn write_offset_indexes(&mut self, offset_indexes: &[Vec<Option<OffsetIndex>>]) -> Result<()> {
+        // iter row group
+        // iter each column
+        // write offset index to the file
+        for (row_group_idx, row_group) in self.row_groups.iter_mut().enumerate() {
+            for (column_idx, column_metadata) in row_group.columns.iter_mut().enumerate() {
+                match &offset_indexes[row_group_idx][column_idx] {
+                    Some(offset_index) => {
+                        let start_offset = self.buf.bytes_written();
+                        let mut protocol = TCompactOutputProtocol::new(&mut self.buf);
+                        offset_index.write_to_out_protocol(&mut protocol)?;
+                        let end_offset = self.buf.bytes_written();
+                        // set offset and index for offset index
+                        column_metadata.offset_index_offset = Some(start_offset as i64);
+                        column_metadata.offset_index_length =
+                            Some((end_offset - start_offset) as i32);
+                    }
+                    None => {}
+                }
+            }
+        }
+        Ok(())
+    }
+
+    /// Serialize all the column index to the file
+    fn write_column_indexes(&mut self, column_indexes: &[Vec<Option<ColumnIndex>>]) -> Result<()> {
+        // iter row group
+        // iter each column
+        // write column index to the file
+        for (row_group_idx, row_group) in self.row_groups.iter_mut().enumerate() {
+            for (column_idx, column_metadata) in row_group.columns.iter_mut().enumerate() {
+                match &column_indexes[row_group_idx][column_idx] {
+                    Some(column_index) => {
+                        let start_offset = self.buf.bytes_written();
+                        let mut protocol = TCompactOutputProtocol::new(&mut self.buf);
+                        column_index.write_to_out_protocol(&mut protocol)?;
+                        let end_offset = self.buf.bytes_written();
+                        // set offset and index for offset index
+                        column_metadata.column_index_offset = Some(start_offset as i64);
+                        column_metadata.column_index_length =
+                            Some((end_offset - start_offset) as i32);
+                    }
+                    None => {}
+                }
+            }
+        }
+        Ok(())
+    }
+
+    /// Assembles and writes the final metadata to self.buf
+    pub fn finish(mut self) -> Result<parquet::FileMetaData> {
+        let num_rows = self.row_groups.iter().map(|x| x.num_rows).sum();
+
+        // Write column indexes and offset indexes
+        if let Some(column_indexes) = self.column_indexes {
+            self.write_column_indexes(column_indexes)?;
+        }
+        if let Some(offset_indexes) = self.offset_indexes {
+            self.write_offset_indexes(offset_indexes)?;
+        }
+
+        // We only include ColumnOrder for leaf nodes.
+        // Currently only supported ColumnOrder is TypeDefinedOrder so we set this
+        // for all leaf nodes.
+        // Even if the column has an undefined sort order, such as INTERVAL, this
+        // is still technically the defined TYPEORDER so it should still be set.
+        let column_orders = (0..self.schema_descr.num_columns())
+            .map(|_| parquet::ColumnOrder::TYPEORDER(parquet::TypeDefinedOrder {}))
+            .collect();
+        // This field is optional, perhaps in cases where no min/max fields are set
+        // in any Statistics or ColumnIndex object in the whole file.
+        // But for simplicity we always set this field.
+        let column_orders = Some(column_orders);
+
+        let file_metadata = parquet::FileMetaData {
+            num_rows,
+            row_groups: self.row_groups,
+            key_value_metadata: self.key_value_metadata.clone(),
+            version: self.writer_version,
+            schema: types::to_thrift(self.schema.as_ref())?,
+            created_by: self.created_by.clone(),
+            column_orders,
+            encryption_algorithm: None,
+            footer_signing_key_metadata: None,
+        };
+
+        // Write file metadata
+        let start_pos = self.buf.bytes_written();
+        {
+            let mut protocol = TCompactOutputProtocol::new(&mut self.buf);
+            file_metadata.write_to_out_protocol(&mut protocol)?;
+        }
+        let end_pos = self.buf.bytes_written();
+
+        // Write footer
+        let metadata_len = (end_pos - start_pos) as u32;
+
+        self.buf.write_all(&metadata_len.to_le_bytes())?;
+        self.buf.write_all(&PARQUET_MAGIC)?;
+        Ok(file_metadata)
+    }
+
+    pub(self) fn new(
+        buf: &'a mut TrackedWrite<W>,
+        schema: &'a TypePtr,
+        schema_descr: &'a SchemaDescPtr,
+        row_groups: Vec<RowGroup>,
+        created_by: Option<String>,
+        writer_version: i32,
+    ) -> Self {
+        Self {
+            buf,
+            schema,
+            schema_descr,
+            row_groups,
+            column_indexes: None,
+            offset_indexes: None,
+            key_value_metadata: None,
+            created_by,
+            writer_version,
+        }
+    }
+
+    pub fn with_column_indexes(mut self, column_indexes: &'a [Vec<Option<ColumnIndex>>]) -> Self {
+        self.column_indexes = Some(column_indexes);
+        self
+    }
+
+    pub fn with_offset_indexes(mut self, offset_indexes: &'a [Vec<Option<OffsetIndex>>]) -> Self {
+        self.offset_indexes = Some(offset_indexes);
+        self
+    }
+
+    pub fn with_key_value_metadata(mut self, key_value_metadata: Vec<KeyValue>) -> Self {
+        self.key_value_metadata = Some(key_value_metadata);
+        self
+    }
+}
+
+pub struct ParquetMetadataWriter<'a, W: Write> {
+    buf: TrackedWrite<W>,
+    write_page_index: bool,
+    metadata: &'a ParquetMetaData,
+}
+
+impl<'a, W: Write> ParquetMetadataWriter<'a, W> {
+    pub fn new(buf: W, metadata: &'a ParquetMetaData) -> Self {
+        Self {
+            buf: TrackedWrite::new(buf),
+            write_page_index: true,
+            metadata,
+        }
+    }
+
+    pub fn write_page_index(&mut self, write_page_index: bool) -> &mut Self {
+        self.write_page_index = write_page_index;
+        self
+    }
+
+    pub fn finish(&mut self) -> Result<()> {
+        let file_metadata = self.metadata.file_metadata();
+
+        let schema = Arc::new(file_metadata.schema().clone());
+        let schema_descr = Arc::new(SchemaDescriptor::new(schema.clone()));
+        let created_by = file_metadata.created_by().map(str::to_string);
+
+        let row_groups = self
+            .metadata
+            .row_groups()
+            .iter()
+            .map(|rg| rg.to_thrift())
+            .collect::<Vec<_>>();
+
+        let key_value_metadata = file_metadata.key_value_metadata().cloned();
+
+        let column_indexes = self.convert_column_indexes();
+        let offset_indexes = self.convert_offset_index();
+
+        let mut encoder = ThriftMetadataWriter::new(
+            &mut self.buf,
+            &schema,
+            &schema_descr,
+            row_groups,
+            created_by,
+            file_metadata.version(),
+        );
+        encoder = encoder.with_column_indexes(&column_indexes);
+        encoder = encoder.with_offset_indexes(&offset_indexes);
+        if let Some(key_value_metadata) = key_value_metadata {
+            encoder = encoder.with_key_value_metadata(key_value_metadata);
+        }
+        encoder.finish()?;
+
+        Ok(())
+    }
+
+    fn convert_column_indexes(&self) -> Vec<Vec<Option<ColumnIndex>>> {
+        if let Some(row_group_column_indexes) = self.metadata.column_index() {
+            (0..self.metadata.row_groups().len())
+                .map(|rg_idx| {
+                    let column_indexes = &row_group_column_indexes[rg_idx];
+                    column_indexes
+                        .iter()
+                        .map(|column_index| match column_index {
+                            Index::NONE => None,
+                            Index::BOOLEAN(column_index) => Some(column_index.to_thrift()),
+                            Index::BYTE_ARRAY(column_index) => Some(column_index.to_thrift()),
+                            Index::DOUBLE(column_index) => Some(column_index.to_thrift()),
+                            Index::FIXED_LEN_BYTE_ARRAY(column_index) => {
+                                Some(column_index.to_thrift())
+                            }
+                            Index::FLOAT(column_index) => Some(column_index.to_thrift()),
+                            Index::INT32(column_index) => Some(column_index.to_thrift()),
+                            Index::INT64(column_index) => Some(column_index.to_thrift()),
+                            Index::INT96(column_index) => Some(column_index.to_thrift()),
+                        })
+                        .collect()
+                })
+                .collect()
+        } else {
+            // make a None for each row group, for each column
+            self.metadata
+                .row_groups()
+                .iter()
+                .map(|rg| std::iter::repeat(None).take(rg.columns().len()).collect())
+                .collect()
+        }
+    }
+
+    fn convert_offset_index(&self) -> Vec<Vec<Option<OffsetIndex>>> {
+        if let Some(row_group_offset_indexes) = self.metadata.offset_index() {
+            (0..self.metadata.row_groups().len())
+                .map(|rg_idx| {
+                    let offset_indexes = &row_group_offset_indexes[rg_idx];
+                    offset_indexes
+                        .iter()
+                        .map(|offset_index| Some(offset_index.to_thrift()))
+                        .collect()
+                })
+                .collect()
+        } else {
+            // make a None for each row group, for each column
+            self.metadata
+                .row_groups()
+                .iter()
+                .map(|rg| std::iter::repeat(None).take(rg.columns().len()).collect())
+                .collect()
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
 
+    #[cfg(feature = "arrow")]
+    use arrow_array::RecordBatchReader;
     use bytes::Bytes;
     use std::fs::File;
 
+    #[cfg(feature = "arrow")]
+    use crate::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
+    #[cfg(feature = "arrow")]
+    use crate::arrow::ArrowWriter;
     use crate::basic::{
         ColumnOrder, Compression, ConvertedType, Encoding, LogicalType, Repetition, SortOrder, Type,
     };
@@ -2063,4 +2264,348 @@ mod tests {
         assert_eq!(offset_index[0].len(), 1);
         assert!(offset_index[0][0].unencoded_byte_array_data_bytes.is_none());
     }
+
+    #[cfg(feature = "async")]
+    mod async_tests {
+        use std::sync::Arc;
+
+        use crate::file::footer::parse_metadata;
+        use crate::file::properties::{EnabledStatistics, WriterProperties};
+        use crate::file::reader::{FileReader, SerializedFileReader};
+        use crate::file::writer::ParquetMetadataWriter;
+        use crate::{
+            arrow::ArrowWriter,
+            file::{page_index::index::Index, serialized_reader::ReadOptionsBuilder},
+        };
+        use arrow_array::{ArrayRef, Int32Array, RecordBatch};
+        use arrow_schema::{DataType as ArrowDataType, Field, Schema};
+        use bytes::{BufMut, Bytes, BytesMut};
+
+        use super::{ColumnChunkMetaData, ParquetMetaData, RowGroupMetaData};
+
+        struct TestMetadata {
+            #[allow(dead_code)]
+            file_size: usize,
+            metadata: ParquetMetaData,
+        }
+
+        fn has_page_index(metadata: &ParquetMetaData) -> bool {
+            match metadata.column_index() {
+                Some(column_index) => column_index
+                    .iter()
+                    .any(|rg_idx| rg_idx.iter().all(|col_idx| !matches!(col_idx, Index::NONE))),
+                None => false,
+            }
+        }
+
+        #[test]
+        fn test_roundtrip_parquet_metadata_without_page_index() {
+            // We currently don't have an ad-hoc ParquetMetadata loader that can load page indexes so
+            // we at least test round trip without them
+            let metadata = get_test_metadata(false, false);
+            assert!(!has_page_index(&metadata.metadata));
+
+            let mut buf = BytesMut::new().writer();
+            {
+                let mut writer = ParquetMetadataWriter::new(&mut buf, &metadata.metadata);
+                writer.finish().unwrap();
+            }
+
+            let data = buf.into_inner().freeze();
+
+            let decoded_metadata = parse_metadata(&data).unwrap();
+            assert!(!has_page_index(&metadata.metadata));
+
+            assert_eq!(metadata.metadata, decoded_metadata);
+        }
+
+        fn get_test_metadata(write_page_index: bool, read_page_index: bool) -> TestMetadata {
+            let mut buf = BytesMut::new().writer();
+            let schema: Arc<Schema> = Arc::new(Schema::new(vec![Field::new(
+                "a",
+                ArrowDataType::Int32,
+                true,
+            )]));
+
+            let a: ArrayRef = Arc::new(Int32Array::from(vec![Some(1), None, Some(2)]));
+
+            let batch = RecordBatch::try_from_iter(vec![("a", a)]).unwrap();
+
+            let writer_props = match write_page_index {
+                true => WriterProperties::builder()
+                    .set_statistics_enabled(EnabledStatistics::Page)
+                    .build(),
+                false => WriterProperties::builder()
+                    .set_statistics_enabled(EnabledStatistics::Chunk)
+                    .build(),
+            };
+
+            let mut writer = ArrowWriter::try_new(&mut buf, schema, Some(writer_props)).unwrap();
+            writer.write(&batch).unwrap();
+            writer.close().unwrap();
+
+            let data = buf.into_inner().freeze();
+
+            let reader_opts = match read_page_index {
+                true => ReadOptionsBuilder::new().with_page_index().build(),
+                false => ReadOptionsBuilder::new().build(),
+            };
+            let reader = SerializedFileReader::new_with_options(data.clone(), reader_opts).unwrap();
+            let metadata = reader.metadata().clone();
+            TestMetadata {
+                file_size: data.len(),
+                metadata,
+            }
+        }
+
+        /// Temporary function so we can test loading metadata with page indexes
+        /// while we haven't fully figured out how to load it cleanly
+        async fn load_metadata_from_bytes(file_size: usize, data: Bytes) -> ParquetMetaData {
+            use crate::arrow::async_reader::{MetadataFetch, MetadataLoader};
+            use crate::errors::Result as ParquetResult;
+            use bytes::Bytes;
+            use futures::future::BoxFuture;
+            use futures::FutureExt;
+            use std::ops::Range;
+
+            /// Adapt a `Bytes` to a `MetadataFetch` implementation.
+            struct AsyncBytes {
+                data: Bytes,
+            }
+
+            impl AsyncBytes {
+                fn new(data: Bytes) -> Self {
+                    Self { data }
+                }
+            }
+
+            impl MetadataFetch for AsyncBytes {
+                fn fetch(&mut self, range: Range<usize>) -> BoxFuture<'_, ParquetResult<Bytes>> {
+                    async move { Ok(self.data.slice(range.start..range.end)) }.boxed()
+                }
+            }
+
+            /// A `MetadataFetch` implementation that reads from a subset of the full data
+            /// while accepting ranges that address the full data.
+            struct MaskedBytes {
+                inner: Box<dyn MetadataFetch + Send>,
+                inner_range: Range<usize>,
+            }
+
+            impl MaskedBytes {
+                fn new(inner: Box<dyn MetadataFetch + Send>, inner_range: Range<usize>) -> Self {
+                    Self { inner, inner_range }
+                }
+            }
+
+            impl MetadataFetch for &mut MaskedBytes {
+                fn fetch(&mut self, range: Range<usize>) -> BoxFuture<'_, ParquetResult<Bytes>> {
+                    let inner_range = self.inner_range.clone();
+                    println!("inner_range: {:?}", inner_range);
+                    println!("range: {:?}", range);
+                    assert!(inner_range.start <= range.start && inner_range.end >= range.end);
+                    let range =
+                        range.start - self.inner_range.start..range.end - self.inner_range.start;
+                    self.inner.fetch(range)
+                }
+            }
+
+            let metadata_length = data.len();
+            let mut reader = MaskedBytes::new(
+                Box::new(AsyncBytes::new(data)),
+                file_size - metadata_length..file_size,
+            );
+            let metadata = MetadataLoader::load(&mut reader, file_size, None)
+                .await
+                .unwrap();
+            let loaded_metadata = metadata.finish();
+            let mut metadata = MetadataLoader::new(&mut reader, loaded_metadata);
+            metadata.load_page_index(true, true).await.unwrap();
+            metadata.finish()
+        }
+
+        fn check_columns_are_equivalent(left: &ColumnChunkMetaData, right: &ColumnChunkMetaData) {
+            assert_eq!(left.column_descr(), right.column_descr());
+            assert_eq!(left.encodings(), right.encodings());
+            assert_eq!(left.num_values(), right.num_values());
+            assert_eq!(left.compressed_size(), right.compressed_size());
+            assert_eq!(left.data_page_offset(), right.data_page_offset());
+            assert_eq!(left.statistics(), right.statistics());
+            assert_eq!(left.offset_index_length(), right.offset_index_length());
+            assert_eq!(left.column_index_length(), right.column_index_length());
+            assert_eq!(
+                left.unencoded_byte_array_data_bytes(),
+                right.unencoded_byte_array_data_bytes()
+            );
+        }
+
+        fn check_row_groups_are_equivalent(left: &RowGroupMetaData, right: &RowGroupMetaData) {
+            assert_eq!(left.num_rows(), right.num_rows());
+            assert_eq!(left.file_offset(), right.file_offset());
+            assert_eq!(left.total_byte_size(), right.total_byte_size());
+            assert_eq!(left.schema_descr(), right.schema_descr());
+            assert_eq!(left.num_columns(), right.num_columns());
+            left.columns()
+                .iter()
+                .zip(right.columns().iter())
+                .for_each(|(lc, rc)| {
+                    check_columns_are_equivalent(lc, rc);
+                });
+        }
+
+        #[tokio::test]
+        async fn test_encode_parquet_metadata_with_page_index() {
+            // Create a ParquetMetadata with page index information
+            let metadata = get_test_metadata(true, true);
+            assert!(has_page_index(&metadata.metadata));
+
+            let mut buf = BytesMut::new().writer();
+            {
+                let mut writer = ParquetMetadataWriter::new(&mut buf, &metadata.metadata);
+                writer.finish().unwrap();
+            }
+
+            let data = buf.into_inner().freeze();
+
+            let decoded_metadata = load_metadata_from_bytes(data.len(), data).await;
+
+            // Because the page index offsets will differ, compare invariant parts of the metadata
+            assert_eq!(
+                metadata.metadata.file_metadata(),
+                decoded_metadata.file_metadata()
+            );
+            assert_eq!(
+                metadata.metadata.column_index(),
+                decoded_metadata.column_index()
+            );
+            assert_eq!(
+                metadata.metadata.offset_index(),
+                decoded_metadata.offset_index()
+            );
+            assert_eq!(
+                metadata.metadata.num_row_groups(),
+                decoded_metadata.num_row_groups()
+            );
+
+            metadata
+                .metadata
+                .row_groups()
+                .iter()
+                .zip(decoded_metadata.row_groups().iter())
+                .for_each(|(left, right)| {
+                    check_row_groups_are_equivalent(left, right);
+                });
+        }
+    }
+
+    #[test]
+    #[cfg(feature = "arrow")]
+    fn test_byte_stream_split_extended_roundtrip() {
+        let path = format!(
+            "{}/byte_stream_split_extended.gzip.parquet",
+            arrow::util::test_util::parquet_test_data(),
+        );
+        let file = File::open(path).unwrap();
+
+        // Read in test file and rewrite to tmp
+        let parquet_reader = ParquetRecordBatchReaderBuilder::try_new(file)
+            .expect("parquet open")
+            .build()
+            .expect("parquet open");
+
+        let file = tempfile::tempfile().unwrap();
+        let props = WriterProperties::builder()
+            .set_dictionary_enabled(false)
+            .set_column_encoding(
+                ColumnPath::from("float16_byte_stream_split"),
+                Encoding::BYTE_STREAM_SPLIT,
+            )
+            .set_column_encoding(
+                ColumnPath::from("float_byte_stream_split"),
+                Encoding::BYTE_STREAM_SPLIT,
+            )
+            .set_column_encoding(
+                ColumnPath::from("double_byte_stream_split"),
+                Encoding::BYTE_STREAM_SPLIT,
+            )
+            .set_column_encoding(
+                ColumnPath::from("int32_byte_stream_split"),
+                Encoding::BYTE_STREAM_SPLIT,
+            )
+            .set_column_encoding(
+                ColumnPath::from("int64_byte_stream_split"),
+                Encoding::BYTE_STREAM_SPLIT,
+            )
+            .set_column_encoding(
+                ColumnPath::from("flba5_byte_stream_split"),
+                Encoding::BYTE_STREAM_SPLIT,
+            )
+            .set_column_encoding(
+                ColumnPath::from("decimal_byte_stream_split"),
+                Encoding::BYTE_STREAM_SPLIT,
+            )
+            .build();
+
+        let mut parquet_writer = ArrowWriter::try_new(
+            file.try_clone().expect("cannot open file"),
+            parquet_reader.schema(),
+            Some(props),
+        )
+        .expect("create arrow writer");
+
+        for maybe_batch in parquet_reader {
+            let batch = maybe_batch.expect("reading batch");
+            parquet_writer.write(&batch).expect("writing data");
+        }
+
+        parquet_writer.close().expect("finalizing file");
+
+        let reader = SerializedFileReader::new(file).expect("Failed to create reader");
+        let filemeta = reader.metadata();
+
+        // Make sure byte_stream_split encoding was used
+        let check_encoding = |x: usize, filemeta: &ParquetMetaData| {
+            assert!(filemeta
+                .row_group(0)
+                .column(x)
+                .encodings()
+                .contains(&Encoding::BYTE_STREAM_SPLIT));
+        };
+
+        check_encoding(1, filemeta);
+        check_encoding(3, filemeta);
+        check_encoding(5, filemeta);
+        check_encoding(7, filemeta);
+        check_encoding(9, filemeta);
+        check_encoding(11, filemeta);
+        check_encoding(13, filemeta);
+
+        // Read back tmpfile and make sure all values are correct
+        let mut iter = reader
+            .get_row_iter(None)
+            .expect("Failed to create row iterator");
+
+        let mut start = 0;
+        let end = reader.metadata().file_metadata().num_rows();
+
+        let check_row = |row: Result<Row, ParquetError>| {
+            assert!(row.is_ok());
+            let r = row.unwrap();
+            assert_eq!(r.get_float16(0).unwrap(), r.get_float16(1).unwrap());
+            assert_eq!(r.get_float(2).unwrap(), r.get_float(3).unwrap());
+            assert_eq!(r.get_double(4).unwrap(), r.get_double(5).unwrap());
+            assert_eq!(r.get_int(6).unwrap(), r.get_int(7).unwrap());
+            assert_eq!(r.get_long(8).unwrap(), r.get_long(9).unwrap());
+            assert_eq!(r.get_bytes(10).unwrap(), r.get_bytes(11).unwrap());
+            assert_eq!(r.get_decimal(12).unwrap(), r.get_decimal(13).unwrap());
+        };
+
+        while start < end {
+            match iter.next() {
+                Some(row) => check_row(row),
+                None => break,
+            };
+            start += 1;
+        }
+    }
 }
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(
diff --git a/parquet/src/thrift.rs b/parquet/src/thrift.rs
index ad6c3f688..abb2ac13c 100644
--- a/parquet/src/thrift.rs
+++ b/parquet/src/thrift.rs
@@ -17,6 +17,7 @@
 
 //! Custom thrift definitions
 
+pub use thrift::protocol::TCompactOutputProtocol;
 use thrift::protocol::{
     TFieldIdentifier, TInputProtocol, TListIdentifier, TMapIdentifier, TMessageIdentifier,
     TOutputProtocol, TSetIdentifier, TStructIdentifier, TType,


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/arrow/array_reader/test_util.rs b/parquet/src/arrow/array_reader/test_util.rs
index 050329201..a7ff8d6e4 100644
--- a/parquet/src/arrow/array_reader/test_util.rs
+++ b/parquet/src/arrow/array_reader/test_util.rs
@@ -46,7 +46,8 @@ pub fn utf8_column() -> ColumnDescPtr {
 
 /// Encode `data` with the provided `encoding`
 pub fn encode_byte_array(encoding: Encoding, data: &[ByteArray]) -> Bytes {
-    let mut encoder = get_encoder::<ByteArrayType>(encoding).unwrap();
+    let desc = utf8_column();
+    let mut encoder = get_encoder::<ByteArrayType>(encoding, &desc).unwrap();
 
     encoder.put(data).unwrap();
     encoder.flush_buffer().unwrap()
diff --git a/parquet/src/util/test_common/page_util.rs b/parquet/src/util/test_common/page_util.rs
index 858fd8c23..3db43aef0 100644
--- a/parquet/src/util/test_common/page_util.rs
+++ b/parquet/src/util/test_common/page_util.rs
@@ -24,9 +24,10 @@ use crate::data_type::DataType;
 use crate::encodings::encoding::{get_encoder, Encoder};
 use crate::encodings::levels::LevelEncoder;
 use crate::errors::Result;
-use crate::schema::types::ColumnDescPtr;
+use crate::schema::types::{ColumnDescPtr, ColumnDescriptor, ColumnPath, Type as SchemaType};
 use std::iter::Peekable;
 use std::mem;
+use std::sync::Arc;
 
 pub trait DataPageBuilder {
     fn add_rep_levels(&mut self, max_level: i16, rep_levels: &[i16]);
@@ -107,9 +108,22 @@ impl DataPageBuilder for DataPageBuilderImpl {
             self.num_values,
             values.len()
         );
+        // Create test column descriptor.
+        let desc = {
+            let ty = SchemaType::primitive_type_builder("t", T::get_physical_type())
+                .with_length(0)
+                .build()
+                .unwrap();
+            Arc::new(ColumnDescriptor::new(
+                Arc::new(ty),
+                0,
+                0,
+                ColumnPath::new(vec![]),
+            ))
+        };
         self.encoding = Some(encoding);
         let mut encoder: Box<dyn Encoder<T>> =
-            get_encoder::<T>(encoding).expect("get_encoder() should be OK");
+            get_encoder::<T>(encoding, &desc).expect("get_encoder() should be OK");
         encoder.put(values).expect("put() should be OK");
         let encoded_values = encoder
             .flush_buffer()
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