Repository: apache/arrow-rs
Pull request #6184 (2024-08-02 20:28:44)

Add (more) Parquet Metadata Documentation
# Which issue does this PR close?

Part of https://github.com/apache/arrow-rs/issues/6129



# Rationale for this change
 
I have big ambitious to improve parquet metadata handling, see https://github.com/apache/arrow-rs/issues/6129.

I think making the existing documentation more complete is a good first step as I think it both
1. Helps existing users
2. Makes the gaps easier to see


# What changes are included in this PR?
1. Add docs and ASCII art to parquet metadata crate documentation

# Are there any user-facing changes?
1. More docs and ASCII art, no functional changes
<img width="1072" alt="Screenshot 2024-08-02 at 4 27 12 PM" src="https://github.com/user-attachments/assets/570a1d03-91f7-4ec6-a572-0766587f92bd">




alamb: FYI @etseidl  and @adriangb 

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/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/filter_kernels.rs b/arrow/benches/filter_kernels.rs
index e48b53022..ae5eec386 100644
--- a/arrow/benches/filter_kernels.rs
+++ b/arrow/benches/filter_kernels.rs
@@ -204,6 +204,49 @@ fn add_benchmark(c: &mut Criterion) {
         |b| b.iter(|| bench_built_filter(&sparse_filter, &data_array)),
     );
 
+    let mut add_benchmark_for_fsb_with_length = |value_length: usize| {
+        let data_array = create_fsb_array(size, 0.0, value_length);
+        c.bench_function(
+            format!("filter fsb with value length {value_length} (kept 1/2)").as_str(),
+            |b| b.iter(|| bench_filter(&data_array, &filter_array)),
+        );
+        c.bench_function(
+            format!(
+                "filter fsb with value length {value_length} high selectivity (kept 1023/1024)"
+            )
+            .as_str(),
+            |b| b.iter(|| bench_filter(&data_array, &dense_filter_array)),
+        );
+        c.bench_function(
+            format!("filter fsb with value length {value_length} low selectivity (kept 1/1024)")
+                .as_str(),
+            |b| b.iter(|| bench_filter(&data_array, &sparse_filter_array)),
+        );
+
+        c.bench_function(
+            format!("filter context fsb with value length {value_length} (kept 1/2)").as_str(),
+            |b| b.iter(|| bench_built_filter(&filter, &filter_array)),
+        );
+        c.bench_function(
+            format!(
+                "filter context fsb with value length {value_length} high selectivity (kept 1023/1024)"
+            )
+            .as_str(),
+            |b| b.iter(|| bench_built_filter(&filter, &dense_filter_array)),
+        );
+        c.bench_function(
+            format!(
+                "filter context fsb with value length {value_length} low selectivity (kept 1/1024)"
+            )
+            .as_str(),
+            |b| b.iter(|| bench_built_filter(&filter, &sparse_filter_array)),
+        );
+    };
+
+    add_benchmark_for_fsb_with_length(5);
+    add_benchmark_for_fsb_with_length(20);
+    add_benchmark_for_fsb_with_length(50);
+
     let data_array = create_primitive_array::<Float32Type>(size, 0.0);
 
     let field = Field::new("c1", data_array.data_type().clone(), 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 6a1434bce..c42f92838 100644
--- a/parquet/src/arrow/arrow_reader/statistics.rs
+++ b/parquet/src/arrow/arrow_reader/statistics.rs
@@ -26,7 +26,8 @@ use crate::file::page_index::index::{Index, PageIndex};
 use crate::file::statistics::Statistics as ParquetStatistics;
 use crate::schema::types::SchemaDescriptor;
 use arrow_array::builder::{
-    BooleanBuilder, FixedSizeBinaryBuilder, LargeStringBuilder, StringBuilder,
+    BinaryViewBuilder, BooleanBuilder, FixedSizeBinaryBuilder, LargeStringBuilder, StringBuilder,
+    StringViewBuilder,
 };
 use arrow_array::{
     new_empty_array, new_null_array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Date64Array,
@@ -446,14 +447,43 @@ macro_rules! get_statistics {
             },
             DataType::Dictionary(_, value_type) => {
                 [<$stat_type_prefix:lower _ statistics>](value_type, $iterator)
+            },
+            DataType::Utf8View => {
+                let iterator = [<$stat_type_prefix ByteArrayStatsIterator>]::new($iterator);
+                let mut builder = StringViewBuilder::new();
+                for x in iterator {
+                    let Some(x) = x else {
+                        builder.append_null(); // no statistics value
+                        continue;
+                    };
+
+                    let Ok(x) = std::str::from_utf8(x) else {
+                        builder.append_null();
+                        continue;
+                    };
+
+                    builder.append_value(x);
+                }
+                Ok(Arc::new(builder.finish()))
+            },
+            DataType::BinaryView => {
+                let iterator = [<$stat_type_prefix ByteArrayStatsIterator>]::new($iterator);
+                let mut builder = BinaryViewBuilder::new();
+                for x in iterator {
+                    let Some(x) = x else {
+                        builder.append_null(); // no statistics value
+                        continue;
+                    };
+
+                    builder.append_value(x);
+                }
+                Ok(Arc::new(builder.finish()))
             }
 
             DataType::Map(_,_) |
             DataType::Duration(_) |
             DataType::Interval(_) |
             DataType::Null |
-            DataType::BinaryView |
-            DataType::Utf8View |
             DataType::List(_) |
             DataType::ListView(_) |
             DataType::FixedSizeList(_, _) |
@@ -919,7 +949,7 @@ macro_rules! get_data_page_statistics {
                         }
                     })
                 },
-               Some(DataType::FixedSizeBinary(size)) => {
+                Some(DataType::FixedSizeBinary(size)) => {
                     let mut builder = FixedSizeBinaryBuilder::new(*size);
                     let iterator = [<$stat_type_prefix FixedLenByteArrayDataPageStatsIterator>]::new($iterator);
                     for x in iterator {
@@ -943,7 +973,58 @@ macro_rules! get_data_page_statistics {
                     }
                     Ok(Arc::new(builder.finish()))
                 },
-                _ => unimplemented!()
+                Some(DataType::Utf8View) => {
+                    let mut builder = StringViewBuilder::new();
+                    let iterator = [<$stat_type_prefix ByteArrayDataPageStatsIterator>]::new($iterator);
+                    for x in iterator {
+                        for x in x.into_iter() {
+                            let Some(x) = x else {
+                                builder.append_null(); // no statistics value
+                                continue;
+                            };
+
+                            let Ok(x) = std::str::from_utf8(x.data()) else {
+                                builder.append_null();
+                                continue;
+                            };
+
+                            builder.append_value(x);
+                        }
+                    }
+                    Ok(Arc::new(builder.finish()))
+                },
+                Some(DataType::BinaryView) => {
+                    let mut builder = BinaryViewBuilder::new();
+                    let iterator = [<$stat_type_prefix ByteArrayDataPageStatsIterator>]::new($iterator);
+                    for x in iterator {
+                        for x in x.into_iter() {
+                            let Some(x) = x else {
+                                builder.append_null(); // no statistics value
+                                continue;
+                            };
+
+                            builder.append_value(x);
+                        }
+                    }
+                    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(_, _)) => {
+                    let len = $iterator.count();
+                    // don't know how to extract statistics, so return a null array
+                    Ok(new_null_array($data_type.unwrap(), len))
+                },
+                None => unimplemented!()  // not sure how to handle this
             }
         }
     }
@@ -1499,10 +1580,10 @@ mod test {
     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, BooleanArray, Date32Array,
-        Date64Array, Decimal128Array, Decimal256Array, Float32Array, Float64Array, Int16Array,
-        Int32Array, Int64Array, Int8Array, LargeBinaryArray, RecordBatch, StringArray, StructArray,
-        TimestampNanosecondArray,
+        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;
@@ -1916,6 +1997,65 @@ mod test {
         .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 {
@@ -2539,4 +2679,19 @@ mod test {
 
         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/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/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/mod.rs b/parquet/tests/arrow_reader/mod.rs
index 4f63a5054..7e979dcf3 100644
--- a/parquet/tests/arrow_reader/mod.rs
+++ b/parquet/tests/arrow_reader/mod.rs
@@ -17,13 +17,13 @@
 
 use arrow_array::types::{Int32Type, Int8Type};
 use arrow_array::{
-    Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Date64Array, Decimal128Array,
-    Decimal256Array, DictionaryArray, FixedSizeBinaryArray, Float16Array, Float32Array,
-    Float64Array, Int16Array, Int32Array, Int64Array, Int8Array, LargeBinaryArray,
-    LargeStringArray, RecordBatch, StringArray, StructArray, Time32MillisecondArray,
-    Time32SecondArray, Time64MicrosecondArray, Time64NanosecondArray, TimestampMicrosecondArray,
-    TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray, UInt16Array,
-    UInt32Array, UInt64Array, UInt8Array,
+    Array, ArrayRef, BinaryArray, BinaryViewArray, BooleanArray, Date32Array, Date64Array,
+    Decimal128Array, Decimal256Array, DictionaryArray, FixedSizeBinaryArray, Float16Array,
+    Float32Array, Float64Array, Int16Array, Int32Array, Int64Array, Int8Array, LargeBinaryArray,
+    LargeStringArray, RecordBatch, StringArray, StringViewArray, StructArray,
+    Time32MillisecondArray, Time32SecondArray, Time64MicrosecondArray, Time64NanosecondArray,
+    TimestampMicrosecondArray, TimestampMillisecondArray, TimestampNanosecondArray,
+    TimestampSecondArray, UInt16Array, UInt32Array, UInt64Array, UInt8Array,
 };
 use arrow_buffer::i256;
 use arrow_schema::{DataType, Field, Schema, TimeUnit};
@@ -88,6 +88,8 @@ enum Scenario {
     PeriodsInColumnNames,
     StructArray,
     UTF8,
+    UTF8View,
+    BinaryView,
 }
 
 fn make_boolean_batch(v: Vec<Option<bool>>) -> RecordBatch {
@@ -589,6 +591,16 @@ fn make_utf8_batch(value: Vec<Option<&str>>) -> RecordBatch {
     .unwrap()
 }
 
+fn make_utf8_view_batch(value: Vec<Option<&str>>) -> RecordBatch {
+    let utf8_view = StringViewArray::from(value);
+    RecordBatch::try_from_iter(vec![("utf8_view", Arc::new(utf8_view) as _)]).unwrap()
+}
+
+fn make_binary_view_batch(value: Vec<Option<&[u8]>>) -> RecordBatch {
+    let binary_view = BinaryViewArray::from(value);
+    RecordBatch::try_from_iter(vec![("binary_view", Arc::new(binary_view) as _)]).unwrap()
+}
+
 fn make_dict_batch() -> RecordBatch {
     let values = [
         Some("abc"),
@@ -972,6 +984,35 @@ fn create_data_batch(scenario: Scenario) -> Vec<RecordBatch> {
                 make_utf8_batch(vec![Some("e"), Some("f"), Some("g"), Some("h"), Some("i")]),
             ]
         }
+        Scenario::UTF8View => {
+            // Make utf8_view batch including string length <12 and >12 bytes
+            // as the internal representation of StringView is differed for strings
+            // shorter and longer than that length
+            vec![
+                make_utf8_view_batch(vec![Some("a"), Some("b"), Some("c"), Some("d"), None]),
+                make_utf8_view_batch(vec![Some("a"), Some("e_longerthan12"), None, None, None]),
+                make_utf8_view_batch(vec![
+                    Some("e_longerthan12"),
+                    Some("f_longerthan12"),
+                    Some("g_longerthan12"),
+                    Some("h_longerthan12"),
+                    Some("i_longerthan12"),
+                ]),
+            ]
+        }
+        Scenario::BinaryView => {
+            vec![
+                make_binary_view_batch(vec![Some(b"a"), Some(b"b"), Some(b"c"), Some(b"d"), None]),
+                make_binary_view_batch(vec![Some(b"a"), Some(b"e_longerthan12"), None, None, None]),
+                make_binary_view_batch(vec![
+                    Some(b"e_longerthan12"),
+                    Some(b"f_longerthan12"),
+                    Some(b"g_longerthan12"),
+                    Some(b"h_longerthan12"),
+                    Some(b"i_longerthan12"),
+                ]),
+            ]
+        }
     }
 }
 
diff --git a/parquet/tests/arrow_reader/statistics.rs b/parquet/tests/arrow_reader/statistics.rs
index 5702967ff..75a73ac13 100644
--- a/parquet/tests/arrow_reader/statistics.rs
+++ b/parquet/tests/arrow_reader/statistics.rs
@@ -29,11 +29,11 @@ use arrow::datatypes::{
     TimestampNanosecondType, TimestampSecondType,
 };
 use arrow_array::{
-    make_array, new_null_array, Array, ArrayRef, BinaryArray, BooleanArray, Date32Array,
-    Date64Array, Decimal128Array, Decimal256Array, FixedSizeBinaryArray, Float16Array,
+    make_array, new_null_array, Array, ArrayRef, BinaryArray, BinaryViewArray, BooleanArray,
+    Date32Array, Date64Array, Decimal128Array, Decimal256Array, FixedSizeBinaryArray, Float16Array,
     Float32Array, Float64Array, Int16Array, Int32Array, Int64Array, Int8Array, LargeBinaryArray,
-    LargeStringArray, RecordBatch, StringArray, Time32MillisecondArray, Time32SecondArray,
-    Time64MicrosecondArray, Time64NanosecondArray, TimestampMicrosecondArray,
+    LargeStringArray, RecordBatch, StringArray, StringViewArray, Time32MillisecondArray,
+    Time32SecondArray, Time64MicrosecondArray, Time64NanosecondArray, TimestampMicrosecondArray,
     TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray, UInt16Array,
     UInt32Array, UInt64Array, UInt8Array,
 };
@@ -2059,6 +2059,60 @@ async fn test_utf8() {
     .run();
 }
 
+// UTF8View
+#[tokio::test]
+async fn test_utf8_view() {
+    let reader = TestReader {
+        scenario: Scenario::UTF8View,
+        row_per_group: 5,
+    }
+    .build()
+    .await;
+
+    // test for utf8_view
+    Test {
+        reader: &reader,
+        expected_min: Arc::new(StringViewArray::from(vec!["a", "a", "e_longerthan12"])),
+        expected_max: Arc::new(StringViewArray::from(vec![
+            "d",
+            "e_longerthan12",
+            "i_longerthan12",
+        ])),
+        expected_null_counts: UInt64Array::from(vec![1, 3, 0]),
+        expected_row_counts: Some(UInt64Array::from(vec![5, 5, 5])),
+        column_name: "utf8_view",
+        check: Check::Both,
+    }
+    .run()
+}
+
+// BinaryView
+#[tokio::test]
+async fn test_binary_view() {
+    let reader = TestReader {
+        scenario: Scenario::BinaryView,
+        row_per_group: 5,
+    }
+    .build()
+    .await;
+
+    let expected_min: Vec<Option<&[u8]>> = vec![Some(b"a"), Some(b"a"), Some(b"e_longerthan12")];
+    let expected_max: Vec<Option<&[u8]>> =
+        vec![Some(b"d"), Some(b"e_longerthan12"), Some(b"i_longerthan12")];
+
+    // test for utf8_view
+    Test {
+        reader: &reader,
+        expected_min: Arc::new(BinaryViewArray::from(expected_min)),
+        expected_max: Arc::new(BinaryViewArray::from(expected_max)),
+        expected_null_counts: UInt64Array::from(vec![1, 3, 0]),
+        expected_row_counts: Some(UInt64Array::from(vec![5, 5, 5])),
+        column_name: "binary_view",
+        check: Check::Both,
+    }
+    .run()
+}
+
 ////// Files with missing statistics ///////
 
 #[tokio::test]