Repository: apache/arrow-rs
Pull request #5965 (2024-06-26 22:07:45)

Add ParquetMetadata::memory_size size estimation
~Draft as I need to implement memory size calculations for the schema structs as well~

# Which issue does this PR close?

Closes https://github.com/apache/arrow-rs/issues/1729


# Rationale for this change
 
For systems that want to do low latency queries on parquet files stored in
object store, it is important to somehow provide the ParquetMetadata to the reader to
avoid the overhead of fetching the file footer and re-parsing the metadata. 


For example, when using the `ArrowReaderMetadata` API:

https://github.com/apache/arrow-rs/blob/ed018a34d996590544fe5e833cb601bf46e9758e/parquet/src/arrow/arrow_reader/mod.rs#L293-L315

One way to provide `ParquetMetadata` to the Arrow reader is to cache it in memory and for large numbers
of parquet files this can consume non trivial memory. Thus accurately understanding the memory requirements
is important

# What changes are included in this PR?
1. Add `ParquetMetadata::memory_size()`, named similarly to [`arrow::Array::get_array_memory_size`](https://docs.rs/arrow/latest/arrow/array/trait.Array.html#tymethod.get_array_memory_size)
2. Add a non public trait to help calculate memory usage
3. Tests

# Are there any user-facing changes?

There is a new function in the API

<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->

<!---
If there are any breaking changes to public APIs, please add the `breaking change` label.
-->



Patch:
diff --git a/.github/workflows/docs.yml b/.github/workflows/docs.yml
index accb9f881..08d287bcc 100644
--- a/.github/workflows/docs.yml
+++ b/.github/workflows/docs.yml
@@ -95,3 +95,5 @@ jobs:
           github_token: ${{ secrets.GITHUB_TOKEN }}
           publish_dir: website/build
           publish_branch: asf-site
+          # Avoid accumulating history of in progress API jobs: https://github.com/apache/arrow-rs/issues/5908
+          force_orphan: true
diff --git a/arrow-arith/src/numeric.rs b/arrow-arith/src/numeric.rs
index 17b794762..91aaf6284 100644
--- a/arrow-arith/src/numeric.rs
+++ b/arrow-arith/src/numeric.rs
@@ -888,15 +888,15 @@ mod tests {
 
         test_neg_primitive::<Int32Type>(
             &[i32::MIN],
-            Err("Compute error: Overflow happened on: -2147483648"),
+            Err("Compute error: Overflow happened on: - -2147483648"),
         );
         test_neg_primitive::<Int64Type>(
             &[i64::MIN],
-            Err("Compute error: Overflow happened on: -9223372036854775808"),
+            Err("Compute error: Overflow happened on: - -9223372036854775808"),
         );
         test_neg_primitive::<DurationSecondType>(
             &[i64::MIN],
-            Err("Compute error: Overflow happened on: -9223372036854775808"),
+            Err("Compute error: Overflow happened on: - -9223372036854775808"),
         );
 
         let r = neg_wrapping(&Int32Array::from(vec![i32::MIN])).unwrap();
@@ -911,7 +911,7 @@ mod tests {
 
         assert_eq!(
             err,
-            "Compute error: Overflow happened on: -9223372036854775808"
+            "Compute error: Overflow happened on: - -9223372036854775808"
         );
 
         let a = Decimal128Array::from(vec![1, 3, -44, 2, 4])
diff --git a/arrow-array/src/arithmetic.rs b/arrow-array/src/arithmetic.rs
index 72989ad7d..078c2e3bc 100644
--- a/arrow-array/src/arithmetic.rs
+++ b/arrow-array/src/arithmetic.rs
@@ -237,7 +237,7 @@ macro_rules! native_type_op {
             #[inline]
             fn neg_checked(self) -> Result<Self, ArrowError> {
                 self.checked_neg().ok_or_else(|| {
-                    ArrowError::ComputeError(format!("Overflow happened on: {:?}", self))
+                    ArrowError::ComputeError(format!("Overflow happened on: - {:?}", self))
                 })
             }
 
diff --git a/arrow-array/src/builder/generic_bytes_view_builder.rs b/arrow-array/src/builder/generic_bytes_view_builder.rs
index 6ec34bf5a..2bcc5a3f3 100644
--- a/arrow-array/src/builder/generic_bytes_view_builder.rs
+++ b/arrow-array/src/builder/generic_bytes_view_builder.rs
@@ -128,21 +128,8 @@ impl<T: ByteViewType + ?Sized> GenericByteViewBuilder<T> {
         let end = start.saturating_add(len as usize);
         let b = b.get_unchecked(start..end);
 
-        if len <= 12 {
-            let mut view_buffer = [0; 16];
-            view_buffer[0..4].copy_from_slice(&len.to_le_bytes());
-            view_buffer[4..4 + b.len()].copy_from_slice(b);
-            self.views_builder.append(u128::from_le_bytes(view_buffer));
-        } else {
-            let view = ByteView {
-                length: len,
-                prefix: u32::from_le_bytes(b[0..4].try_into().unwrap()),
-                buffer_index: block,
-                offset,
-            };
-            self.views_builder.append(view.into());
-        }
-
+        let view = make_view(b, block, offset);
+        self.views_builder.append(view);
         self.null_buffer_builder.append_non_null();
     }
 
@@ -345,6 +332,26 @@ pub type StringViewBuilder = GenericByteViewBuilder<StringViewType>;
 /// [`GenericByteViewBuilder::append_null`] as normal.
 pub type BinaryViewBuilder = GenericByteViewBuilder<BinaryViewType>;
 
+/// Create a view based on the given data, block id and offset
+#[inline(always)]
+pub fn make_view(data: &[u8], block_id: u32, offset: u32) -> u128 {
+    let len = data.len() as u32;
+    if len <= 12 {
+        let mut view_buffer = [0; 16];
+        view_buffer[0..4].copy_from_slice(&len.to_le_bytes());
+        view_buffer[4..4 + data.len()].copy_from_slice(data);
+        u128::from_le_bytes(view_buffer)
+    } else {
+        let view = ByteView {
+            length: len,
+            prefix: u32::from_le_bytes(data[0..4].try_into().unwrap()),
+            buffer_index: block_id,
+            offset,
+        };
+        view.into()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/arrow-cast/src/cast/mod.rs b/arrow-cast/src/cast/mod.rs
index e5ab304bb..c9de714e7 100644
--- a/arrow-cast/src/cast/mod.rs
+++ b/arrow-cast/src/cast/mod.rs
@@ -4403,17 +4403,17 @@ mod tests {
         test_unsafe_string_to_interval_err!(
             vec![Some("foobar")],
             IntervalUnit::YearMonth,
-            r#"Not yet implemented: Unsupported Interval Expression with value "foobar""#
+            r#"Parser error: Invalid input syntax for type interval: "foobar""#
         );
         test_unsafe_string_to_interval_err!(
             vec![Some("foobar")],
             IntervalUnit::DayTime,
-            r#"Not yet implemented: Unsupported Interval Expression with value "foobar""#
+            r#"Parser error: Invalid input syntax for type interval: "foobar""#
         );
         test_unsafe_string_to_interval_err!(
             vec![Some("foobar")],
             IntervalUnit::MonthDayNano,
-            r#"Not yet implemented: Unsupported Interval Expression with value "foobar""#
+            r#"Parser error: Invalid input syntax for type interval: "foobar""#
         );
         test_unsafe_string_to_interval_err!(
             vec![Some("2 months 31 days 1 second")],
diff --git a/arrow-cast/src/parse.rs b/arrow-cast/src/parse.rs
index a44c7ad4f..427672a6a 100644
--- a/arrow-cast/src/parse.rs
+++ b/arrow-cast/src/parse.rs
@@ -1381,8 +1381,8 @@ fn parse_interval_components(
 
     // invalid amounts?
     if !invalid_amounts.is_empty() {
-        return Err(ArrowError::NotYetImplemented(format!(
-            "Unsupported Interval Expression with value {value:?}"
+        return Err(ArrowError::ParseError(format!(
+            "Invalid input syntax for type interval: {value:?}"
         )));
     }
 
diff --git a/arrow-flight/src/encode.rs b/arrow-flight/src/encode.rs
index f59c29e68..e7722fd7f 100644
--- a/arrow-flight/src/encode.rs
+++ b/arrow-flight/src/encode.rs
@@ -304,7 +304,11 @@ impl FlightDataEncoder {
         // The first message is the schema message, and all
         // batches have the same schema
         let send_dictionaries = self.dictionary_handling == DictionaryHandling::Resend;
-        let schema = Arc::new(prepare_schema_for_flight(schema, send_dictionaries));
+        let schema = Arc::new(prepare_schema_for_flight(
+            schema,
+            &mut self.encoder.dictionary_tracker,
+            send_dictionaries,
+        ));
         let mut schema_flight_data = self.encoder.encode_schema(&schema);
 
         // attach any metadata requested
@@ -438,24 +442,28 @@ pub enum DictionaryHandling {
     Resend,
 }
 
-fn prepare_field_for_flight(field: &FieldRef, send_dictionaries: bool) -> Field {
+fn prepare_field_for_flight(
+    field: &FieldRef,
+    dictionary_tracker: &mut DictionaryTracker,
+    send_dictionaries: bool,
+) -> Field {
     match field.data_type() {
         DataType::List(inner) => Field::new_list(
             field.name(),
-            prepare_field_for_flight(inner, send_dictionaries),
+            prepare_field_for_flight(inner, dictionary_tracker, send_dictionaries),
             field.is_nullable(),
         )
         .with_metadata(field.metadata().clone()),
         DataType::LargeList(inner) => Field::new_list(
             field.name(),
-            prepare_field_for_flight(inner, send_dictionaries),
+            prepare_field_for_flight(inner, dictionary_tracker, send_dictionaries),
             field.is_nullable(),
         )
         .with_metadata(field.metadata().clone()),
         DataType::Struct(fields) => {
             let new_fields: Vec<Field> = fields
                 .iter()
-                .map(|f| prepare_field_for_flight(f, send_dictionaries))
+                .map(|f| prepare_field_for_flight(f, dictionary_tracker, send_dictionaries))
                 .collect();
             Field::new_struct(field.name(), new_fields, field.is_nullable())
                 .with_metadata(field.metadata().clone())
@@ -463,17 +471,37 @@ fn prepare_field_for_flight(field: &FieldRef, send_dictionaries: bool) -> Field
         DataType::Union(fields, mode) => {
             let (type_ids, new_fields): (Vec<i8>, Vec<Field>) = fields
                 .iter()
-                .map(|(type_id, f)| (type_id, prepare_field_for_flight(f, send_dictionaries)))
+                .map(|(type_id, f)| {
+                    (
+                        type_id,
+                        prepare_field_for_flight(f, dictionary_tracker, send_dictionaries),
+                    )
+                })
                 .unzip();
 
             Field::new_union(field.name(), type_ids, new_fields, *mode)
         }
-        DataType::Dictionary(_, value_type) if !send_dictionaries => Field::new(
-            field.name(),
-            value_type.as_ref().clone(),
-            field.is_nullable(),
-        )
-        .with_metadata(field.metadata().clone()),
+        DataType::Dictionary(_, value_type) => {
+            if !send_dictionaries {
+                Field::new(
+                    field.name(),
+                    value_type.as_ref().clone(),
+                    field.is_nullable(),
+                )
+                .with_metadata(field.metadata().clone())
+            } else {
+                let dict_id = dictionary_tracker.set_dict_id(field.as_ref());
+
+                Field::new_dict(
+                    field.name(),
+                    field.data_type().clone(),
+                    field.is_nullable(),
+                    dict_id,
+                    field.dict_is_ordered().unwrap_or_default(),
+                )
+                .with_metadata(field.metadata().clone())
+            }
+        }
         _ => field.as_ref().clone(),
     }
 }
@@ -483,18 +511,38 @@ fn prepare_field_for_flight(field: &FieldRef, send_dictionaries: bool) -> Field
 /// Convert dictionary types to underlying types
 ///
 /// See hydrate_dictionary for more information
-fn prepare_schema_for_flight(schema: &Schema, send_dictionaries: bool) -> Schema {
+fn prepare_schema_for_flight(
+    schema: &Schema,
+    dictionary_tracker: &mut DictionaryTracker,
+    send_dictionaries: bool,
+) -> Schema {
     let fields: Fields = schema
         .fields()
         .iter()
         .map(|field| match field.data_type() {
-            DataType::Dictionary(_, value_type) if !send_dictionaries => Field::new(
-                field.name(),
-                value_type.as_ref().clone(),
-                field.is_nullable(),
-            )
-            .with_metadata(field.metadata().clone()),
-            tpe if tpe.is_nested() => prepare_field_for_flight(field, send_dictionaries),
+            DataType::Dictionary(_, value_type) => {
+                if !send_dictionaries {
+                    Field::new(
+                        field.name(),
+                        value_type.as_ref().clone(),
+                        field.is_nullable(),
+                    )
+                    .with_metadata(field.metadata().clone())
+                } else {
+                    let dict_id = dictionary_tracker.set_dict_id(field.as_ref());
+                    Field::new_dict(
+                        field.name(),
+                        field.data_type().clone(),
+                        field.is_nullable(),
+                        dict_id,
+                        field.dict_is_ordered().unwrap_or_default(),
+                    )
+                    .with_metadata(field.metadata().clone())
+                }
+            }
+            tpe if tpe.is_nested() => {
+                prepare_field_for_flight(field, dictionary_tracker, send_dictionaries)
+            }
             _ => field.as_ref().clone(),
         })
         .collect();
@@ -548,10 +596,14 @@ struct FlightIpcEncoder {
 
 impl FlightIpcEncoder {
     fn new(options: IpcWriteOptions, error_on_replacement: bool) -> Self {
+        let preserve_dict_id = options.preserve_dict_id();
         Self {
             options,
             data_gen: IpcDataGenerator::default(),
-            dictionary_tracker: DictionaryTracker::new(error_on_replacement),
+            dictionary_tracker: DictionaryTracker::new_with_preserve_dict_id(
+                error_on_replacement,
+                preserve_dict_id,
+            ),
         }
     }
 
@@ -619,7 +671,10 @@ fn hydrate_dictionary(array: &ArrayRef, data_type: &DataType) -> Result<ArrayRef
 
 #[cfg(test)]
 mod tests {
-    use arrow_array::builder::StringDictionaryBuilder;
+    use crate::decode::{DecodedPayload, FlightDataDecoder};
+    use arrow_array::builder::{
+        GenericByteDictionaryBuilder, ListBuilder, StringDictionaryBuilder, StructBuilder,
+    };
     use arrow_array::*;
     use arrow_array::{cast::downcast_array, types::*};
     use arrow_buffer::ScalarBuffer;
@@ -628,8 +683,6 @@ mod tests {
     use arrow_schema::{UnionFields, UnionMode};
     use std::collections::HashMap;
 
-    use crate::decode::{DecodedPayload, FlightDataDecoder};
-
     use super::*;
 
     #[test]
@@ -708,9 +761,52 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn test_dictionary_resend() {
+        let arr1: DictionaryArray<UInt16Type> = vec!["a", "a", "b"].into_iter().collect();
+        let arr2: DictionaryArray<UInt16Type> = vec!["c", "c", "d"].into_iter().collect();
+
+        let schema = Arc::new(Schema::new(vec![Field::new_dictionary(
+            "dict",
+            DataType::UInt16,
+            DataType::Utf8,
+            false,
+        )]));
+        let batch1 = RecordBatch::try_new(schema.clone(), vec![Arc::new(arr1)]).unwrap();
+        let batch2 = RecordBatch::try_new(schema, vec![Arc::new(arr2)]).unwrap();
+
+        verify_flight_round_trip(vec![batch1, batch2]).await;
+    }
+
+    #[tokio::test]
+    async fn test_multiple_dictionaries_resend() {
+        // Create a schema with two dictionary fields that have the same dict ID
+        let schema = Arc::new(Schema::new(vec![
+            Field::new_dictionary("dict_1", DataType::UInt16, DataType::Utf8, false),
+            Field::new_dictionary("dict_2", DataType::UInt16, DataType::Utf8, false),
+        ]));
+
+        let arr_one_1: Arc<DictionaryArray<UInt16Type>> =
+            Arc::new(vec!["a", "a", "b"].into_iter().collect());
+        let arr_one_2: Arc<DictionaryArray<UInt16Type>> =
+            Arc::new(vec!["c", "c", "d"].into_iter().collect());
+        let arr_two_1: Arc<DictionaryArray<UInt16Type>> =
+            Arc::new(vec!["b", "a", "c"].into_iter().collect());
+        let arr_two_2: Arc<DictionaryArray<UInt16Type>> =
+            Arc::new(vec!["k", "d", "e"].into_iter().collect());
+        let batch1 =
+            RecordBatch::try_new(schema.clone(), vec![arr_one_1.clone(), arr_one_2.clone()])
+                .unwrap();
+        let batch2 =
+            RecordBatch::try_new(schema.clone(), vec![arr_two_1.clone(), arr_two_2.clone()])
+                .unwrap();
+
+        verify_flight_round_trip(vec![batch1, batch2]).await;
+    }
+
     #[tokio::test]
     async fn test_dictionary_list_hydration() {
-        let mut builder = builder::ListBuilder::new(StringDictionaryBuilder::<UInt16Type>::new());
+        let mut builder = ListBuilder::new(StringDictionaryBuilder::<UInt16Type>::new());
 
         builder.append_value(vec![Some("a"), None, Some("b")]);
 
@@ -766,6 +862,30 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn test_dictionary_list_resend() {
+        let mut builder = ListBuilder::new(StringDictionaryBuilder::<UInt16Type>::new());
+
+        builder.append_value(vec![Some("a"), None, Some("b")]);
+
+        let arr1 = builder.finish();
+
+        builder.append_value(vec![Some("c"), None, Some("d")]);
+
+        let arr2 = builder.finish();
+
+        let schema = Arc::new(Schema::new(vec![Field::new_list(
+            "dict_list",
+            Field::new_dictionary("item", DataType::UInt16, DataType::Utf8, true),
+            true,
+        )]));
+
+        let batch1 = RecordBatch::try_new(schema.clone(), vec![Arc::new(arr1)]).unwrap();
+        let batch2 = RecordBatch::try_new(schema.clone(), vec![Arc::new(arr2)]).unwrap();
+
+        verify_flight_round_trip(vec![batch1, batch2]).await;
+    }
+
     #[tokio::test]
     async fn test_dictionary_struct_hydration() {
         let struct_fields = vec![Field::new_list(
@@ -774,26 +894,38 @@ mod tests {
             true,
         )];
 
-        let mut builder = builder::ListBuilder::new(StringDictionaryBuilder::<UInt16Type>::new());
+        let mut struct_builder = StructBuilder::new(
+            struct_fields.clone(),
+            vec![Box::new(builder::ListBuilder::new(
+                StringDictionaryBuilder::<UInt16Type>::new(),
+            ))],
+        );
 
-        builder.append_value(vec![Some("a"), None, Some("b")]);
+        struct_builder
+            .field_builder::<ListBuilder<GenericByteDictionaryBuilder<UInt16Type,GenericStringType<i32>>>>(0)
+            .unwrap()
+            .append_value(vec![Some("a"), None, Some("b")]);
 
-        let arr1 = Arc::new(builder.finish());
-        let arr1 = StructArray::new(struct_fields.clone().into(), vec![arr1], None);
+        struct_builder.append(true);
 
-        builder.append_value(vec![Some("c"), None, Some("d")]);
+        let arr1 = struct_builder.finish();
 
-        let arr2 = Arc::new(builder.finish());
-        let arr2 = StructArray::new(struct_fields.clone().into(), vec![arr2], None);
+        struct_builder
+            .field_builder::<ListBuilder<GenericByteDictionaryBuilder<UInt16Type,GenericStringType<i32>>>>(0)
+            .unwrap()
+            .append_value(vec![Some("c"), None, Some("d")]);
+        struct_builder.append(true);
+
+        let arr2 = struct_builder.finish();
 
         let schema = Arc::new(Schema::new(vec![Field::new_struct(
             "struct",
-            struct_fields.clone(),
+            struct_fields,
             true,
         )]));
 
         let batch1 = RecordBatch::try_new(schema.clone(), vec![Arc::new(arr1)]).unwrap();
-        let batch2 = RecordBatch::try_new(schema.clone(), vec![Arc::new(arr2)]).unwrap();
+        let batch2 = RecordBatch::try_new(schema, vec![Arc::new(arr2)]).unwrap();
 
         let stream = futures::stream::iter(vec![Ok(batch1), Ok(batch2)]);
 
@@ -838,6 +970,43 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn test_dictionary_struct_resend() {
+        let struct_fields = vec![Field::new_list(
+            "dict_list",
+            Field::new_dictionary("item", DataType::UInt16, DataType::Utf8, true),
+            true,
+        )];
+
+        let mut struct_builder = StructBuilder::new(
+            struct_fields.clone(),
+            vec![Box::new(builder::ListBuilder::new(
+                StringDictionaryBuilder::<UInt16Type>::new(),
+            ))],
+        );
+
+        struct_builder.field_builder::<ListBuilder<GenericByteDictionaryBuilder<UInt16Type,GenericStringType<i32>>>>(0).unwrap().append_value(vec![Some("a"), None, Some("b")]);
+        struct_builder.append(true);
+
+        let arr1 = struct_builder.finish();
+
+        struct_builder.field_builder::<ListBuilder<GenericByteDictionaryBuilder<UInt16Type,GenericStringType<i32>>>>(0).unwrap().append_value(vec![Some("c"), None, Some("d")]);
+        struct_builder.append(true);
+
+        let arr2 = struct_builder.finish();
+
+        let schema = Arc::new(Schema::new(vec![Field::new_struct(
+            "struct",
+            struct_fields,
+            true,
+        )]));
+
+        let batch1 = RecordBatch::try_new(schema.clone(), vec![Arc::new(arr1)]).unwrap();
+        let batch2 = RecordBatch::try_new(schema, vec![Arc::new(arr2)]).unwrap();
+
+        verify_flight_round_trip(vec![batch1, batch2]).await;
+    }
+
     #[tokio::test]
     async fn test_dictionary_union_hydration() {
         let struct_fields = vec![Field::new_list(
@@ -1004,42 +1173,124 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn test_send_dictionaries() {
-        let schema = Arc::new(Schema::new(vec![Field::new_dictionary(
-            "dict",
-            DataType::UInt16,
-            DataType::Utf8,
-            false,
+    async fn test_dictionary_union_resend() {
+        let struct_fields = vec![Field::new_list(
+            "dict_list",
+            Field::new_dictionary("item", DataType::UInt16, DataType::Utf8, true),
+            true,
+        )];
+
+        let union_fields = [
+            (
+                0,
+                Arc::new(Field::new_list(
+                    "dict_list",
+                    Field::new_dictionary("item", DataType::UInt16, DataType::Utf8, true),
+                    true,
+                )),
+            ),
+            (
+                1,
+                Arc::new(Field::new_struct("struct", struct_fields.clone(), true)),
+            ),
+            (2, Arc::new(Field::new("string", DataType::Utf8, true))),
+        ]
+        .into_iter()
+        .collect::<UnionFields>();
+
+        let struct_fields = vec![Field::new_list(
+            "dict_list",
+            Field::new_dictionary("item", DataType::UInt16, DataType::Utf8, true),
+            true,
+        )];
+
+        let mut builder = builder::ListBuilder::new(StringDictionaryBuilder::<UInt16Type>::new());
+
+        builder.append_value(vec![Some("a"), None, Some("b")]);
+
+        let arr1 = builder.finish();
+
+        let type_id_buffer = [0].into_iter().collect::<ScalarBuffer<i8>>();
+        let arr1 = UnionArray::try_new(
+            union_fields.clone(),
+            type_id_buffer,
+            None,
+            vec![
+                Arc::new(arr1) as Arc<dyn Array>,
+                new_null_array(union_fields.iter().nth(1).unwrap().1.data_type(), 1),
+                new_null_array(union_fields.iter().nth(2).unwrap().1.data_type(), 1),
+            ],
+        )
+        .unwrap();
+
+        builder.append_value(vec![Some("c"), None, Some("d")]);
+
+        let arr2 = Arc::new(builder.finish());
+        let arr2 = StructArray::new(struct_fields.clone().into(), vec![arr2], None);
+
+        let type_id_buffer = [1].into_iter().collect::<ScalarBuffer<i8>>();
+        let arr2 = UnionArray::try_new(
+            union_fields.clone(),
+            type_id_buffer,
+            None,
+            vec![
+                new_null_array(union_fields.iter().next().unwrap().1.data_type(), 1),
+                Arc::new(arr2),
+                new_null_array(union_fields.iter().nth(2).unwrap().1.data_type(), 1),
+            ],
+        )
+        .unwrap();
+
+        let type_id_buffer = [2].into_iter().collect::<ScalarBuffer<i8>>();
+        let arr3 = UnionArray::try_new(
+            union_fields.clone(),
+            type_id_buffer,
+            None,
+            vec![
+                new_null_array(union_fields.iter().next().unwrap().1.data_type(), 1),
+                new_null_array(union_fields.iter().nth(1).unwrap().1.data_type(), 1),
+                Arc::new(StringArray::from(vec!["e"])),
+            ],
+        )
+        .unwrap();
+
+        let (type_ids, union_fields): (Vec<_>, Vec<_>) = union_fields
+            .iter()
+            .map(|(type_id, field_ref)| (type_id, (*Arc::clone(field_ref)).clone()))
+            .unzip();
+        let schema = Arc::new(Schema::new(vec![Field::new_union(
+            "union",
+            type_ids.clone(),
+            union_fields.clone(),
+            UnionMode::Sparse,
         )]));
 
-        let arr_one: Arc<DictionaryArray<UInt16Type>> =
-            Arc::new(vec!["a", "a", "b"].into_iter().collect());
-        let arr_two: Arc<DictionaryArray<UInt16Type>> =
-            Arc::new(vec!["b", "a", "c"].into_iter().collect());
-        let batch_one = RecordBatch::try_new(schema.clone(), vec![arr_one.clone()]).unwrap();
-        let batch_two = RecordBatch::try_new(schema.clone(), vec![arr_two.clone()]).unwrap();
+        let batch1 = RecordBatch::try_new(schema.clone(), vec![Arc::new(arr1)]).unwrap();
+        let batch2 = RecordBatch::try_new(schema.clone(), vec![Arc::new(arr2)]).unwrap();
+        let batch3 = RecordBatch::try_new(schema.clone(), vec![Arc::new(arr3)]).unwrap();
+
+        verify_flight_round_trip(vec![batch1, batch2, batch3]).await;
+    }
+
+    async fn verify_flight_round_trip(mut batches: Vec<RecordBatch>) {
+        let expected_schema = batches.first().unwrap().schema();
 
         let encoder = FlightDataEncoderBuilder::default()
+            .with_options(IpcWriteOptions::default().with_preserve_dict_id(false))
             .with_dictionary_handling(DictionaryHandling::Resend)
-            .build(futures::stream::iter(vec![Ok(batch_one), Ok(batch_two)]));
+            .build(futures::stream::iter(batches.clone().into_iter().map(Ok)));
+
+        let mut expected_batches = batches.drain(..);
 
         let mut decoder = FlightDataDecoder::new(encoder);
-        let mut expected_array = arr_one;
         while let Some(decoded) = decoder.next().await {
             let decoded = decoded.unwrap();
             match decoded.payload {
                 DecodedPayload::None => {}
-                DecodedPayload::Schema(s) => assert_eq!(s, schema),
+                DecodedPayload::Schema(s) => assert_eq!(s, expected_schema),
                 DecodedPayload::RecordBatch(b) => {
-                    assert_eq!(b.schema(), schema);
-
-                    let actual_array = Arc::new(downcast_array::<DictionaryArray<UInt16Type>>(
-                        b.column_by_name("dict").unwrap(),
-                    ));
-
-                    assert_eq!(actual_array, expected_array);
-
-                    expected_array = arr_two.clone();
+                    let expected_batch = expected_batches.next().unwrap();
+                    assert_eq!(b, expected_batch);
                 }
             }
         }
@@ -1051,7 +1302,9 @@ mod tests {
             HashMap::from([("some_key".to_owned(), "some_value".to_owned())]),
         );
 
-        let got = prepare_schema_for_flight(&schema, false);
+        let mut dictionary_tracker = DictionaryTracker::new_with_preserve_dict_id(false, true);
+
+        let got = prepare_schema_for_flight(&schema, &mut dictionary_tracker, false);
         assert!(got.metadata().contains_key("some_key"));
     }
 
@@ -1072,7 +1325,7 @@ mod tests {
         options: &IpcWriteOptions,
     ) -> (Vec<FlightData>, FlightData) {
         let data_gen = IpcDataGenerator::default();
-        let mut dictionary_tracker = DictionaryTracker::new(false);
+        let mut dictionary_tracker = DictionaryTracker::new_with_preserve_dict_id(false, true);
 
         let (encoded_dictionaries, encoded_batch) = data_gen
             .encoded_batch(batch, &mut dictionary_tracker, options)
diff --git a/arrow-flight/src/utils.rs b/arrow-flight/src/utils.rs
index 32716a52e..c1e2d61fc 100644
--- a/arrow-flight/src/utils.rs
+++ b/arrow-flight/src/utils.rs
@@ -39,7 +39,8 @@ pub fn flight_data_from_arrow_batch(
     options: &IpcWriteOptions,
 ) -> (Vec<FlightData>, FlightData) {
     let data_gen = writer::IpcDataGenerator::default();
-    let mut dictionary_tracker = writer::DictionaryTracker::new(false);
+    let mut dictionary_tracker =
+        writer::DictionaryTracker::new_with_preserve_dict_id(false, options.preserve_dict_id());
 
     let (encoded_dictionaries, encoded_batch) = data_gen
         .encoded_batch(batch, &mut dictionary_tracker, options)
@@ -149,7 +150,8 @@ pub fn batches_to_flight_data(
     let mut flight_data = vec![];
 
     let data_gen = writer::IpcDataGenerator::default();
-    let mut dictionary_tracker = writer::DictionaryTracker::new(false);
+    let mut dictionary_tracker =
+        writer::DictionaryTracker::new_with_preserve_dict_id(false, options.preserve_dict_id());
 
     for batch in batches.iter() {
         let (encoded_dictionaries, encoded_batch) =
diff --git a/arrow-ipc/src/reader.rs b/arrow-ipc/src/reader.rs
index 3423d06b6..1f83200d6 100644
--- a/arrow-ipc/src/reader.rs
+++ b/arrow-ipc/src/reader.rs
@@ -617,7 +617,7 @@ fn read_dictionary_impl(
     let id = batch.id();
     let fields_using_this_dictionary = schema.fields_with_dict_id(id);
     let first_field = fields_using_this_dictionary.first().ok_or_else(|| {
-        ArrowError::InvalidArgumentError("dictionary id not found in schema".to_string())
+        ArrowError::InvalidArgumentError(format!("dictionary id {id} not found in schema"))
     })?;
 
     // As the dictionary batch does not contain the type of the
@@ -643,7 +643,7 @@ fn read_dictionary_impl(
         _ => None,
     }
     .ok_or_else(|| {
-        ArrowError::InvalidArgumentError("dictionary id not found in schema".to_string())
+        ArrowError::InvalidArgumentError(format!("dictionary id {id} not found in schema"))
     })?;
 
     // We don't currently record the isOrdered field. This could be general
@@ -1812,7 +1812,7 @@ mod tests {
             "values",
             DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Utf8)),
             true,
-            1,
+            2,
             false,
         ));
         let entry_struct = StructArray::from(vec![
@@ -2082,7 +2082,7 @@ mod tests {
         .unwrap();
 
         let gen = IpcDataGenerator {};
-        let mut dict_tracker = DictionaryTracker::new(false);
+        let mut dict_tracker = DictionaryTracker::new_with_preserve_dict_id(false, true);
         let (_, encoded) = gen
             .encoded_batch(&batch, &mut dict_tracker, &Default::default())
             .unwrap();
@@ -2119,7 +2119,7 @@ mod tests {
         .unwrap();
 
         let gen = IpcDataGenerator {};
-        let mut dict_tracker = DictionaryTracker::new(false);
+        let mut dict_tracker = DictionaryTracker::new_with_preserve_dict_id(false, true);
         let (_, encoded) = gen
             .encoded_batch(&batch, &mut dict_tracker, &Default::default())
             .unwrap();
diff --git a/arrow-ipc/src/writer.rs b/arrow-ipc/src/writer.rs
index f74a86e01..c07821959 100644
--- a/arrow-ipc/src/writer.rs
+++ b/arrow-ipc/src/writer.rs
@@ -59,6 +59,11 @@ pub struct IpcWriteOptions {
     /// Compression, if desired. Will result in a runtime error
     /// if the corresponding feature is not enabled
     batch_compression_type: Option<crate::CompressionType>,
+    /// Flag indicating whether the writer should preserver the dictionary IDs defined in the
+    /// schema or generate unique dictionary IDs internally during encoding.
+    ///
+    /// Defaults to `true`
+    preserve_dict_id: bool,
 }
 
 impl IpcWriteOptions {
@@ -81,7 +86,7 @@ impl IpcWriteOptions {
         }
         Ok(self)
     }
-    /// Try create IpcWriteOptions, checking for incompatible settings
+    /// Try to create IpcWriteOptions, checking for incompatible settings
     pub fn try_new(
         alignment: usize,
         write_legacy_ipc_format: bool,
@@ -106,6 +111,7 @@ impl IpcWriteOptions {
                 write_legacy_ipc_format,
                 metadata_version,
                 batch_compression_type: None,
+                preserve_dict_id: true,
             }),
             crate::MetadataVersion::V5 => {
                 if write_legacy_ipc_format {
@@ -118,6 +124,7 @@ impl IpcWriteOptions {
                         write_legacy_ipc_format,
                         metadata_version,
                         batch_compression_type: None,
+                        preserve_dict_id: true,
                     })
                 }
             }
@@ -126,6 +133,22 @@ impl IpcWriteOptions {
             ))),
         }
     }
+
+    pub fn preserve_dict_id(&self) -> bool {
+        self.preserve_dict_id
+    }
+
+    /// Set whether the IPC writer should preserve the dictionary IDs in the schema
+    /// or auto-assign unique dictionary IDs during encoding (defaults to true)
+    ///
+    /// If this option is true,  the application must handle assigning ids
+    /// to the dictionary batches in order to encode them correctly
+    ///
+    /// The default will change to `false`  in future releases
+    pub fn with_preserve_dict_id(mut self, preserve_dict_id: bool) -> Self {
+        self.preserve_dict_id = preserve_dict_id;
+        self
+    }
 }
 
 impl Default for IpcWriteOptions {
@@ -135,6 +158,7 @@ impl Default for IpcWriteOptions {
             write_legacy_ipc_format: false,
             metadata_version: crate::MetadataVersion::V5,
             batch_compression_type: None,
+            preserve_dict_id: true,
         }
     }
 }
@@ -163,7 +187,7 @@ impl Default for IpcWriteOptions {
 ///
 /// // encode the batch into zero or more encoded dictionaries
 /// // and the data for the actual array.
-/// let data_gen = IpcDataGenerator {};
+/// let data_gen = IpcDataGenerator::default();
 /// let (encoded_dictionaries, encoded_message) = data_gen
 ///   .encoded_batch(&batch, &mut dictionary_tracker, &options)
 ///   .unwrap();
@@ -198,12 +222,13 @@ impl IpcDataGenerator {
         }
     }
 
-    fn _encode_dictionaries(
+    fn _encode_dictionaries<I: Iterator<Item = i64>>(
         &self,
         column: &ArrayRef,
         encoded_dictionaries: &mut Vec<EncodedData>,
         dictionary_tracker: &mut DictionaryTracker,
         write_options: &IpcWriteOptions,
+        dict_id: &mut I,
     ) -> Result<(), ArrowError> {
         match column.data_type() {
             DataType::Struct(fields) => {
@@ -215,6 +240,7 @@ impl IpcDataGenerator {
                         encoded_dictionaries,
                         dictionary_tracker,
                         write_options,
+                        dict_id,
                     )?;
                 }
             }
@@ -235,6 +261,7 @@ impl IpcDataGenerator {
                     encoded_dictionaries,
                     dictionary_tracker,
                     write_options,
+                    dict_id,
                 )?;
             }
             DataType::List(field) => {
@@ -245,6 +272,7 @@ impl IpcDataGenerator {
                     encoded_dictionaries,
                     dictionary_tracker,
                     write_options,
+                    dict_id,
                 )?;
             }
             DataType::LargeList(field) => {
@@ -255,6 +283,7 @@ impl IpcDataGenerator {
                     encoded_dictionaries,
                     dictionary_tracker,
                     write_options,
+                    dict_id,
                 )?;
             }
             DataType::FixedSizeList(field, _) => {
@@ -268,6 +297,7 @@ impl IpcDataGenerator {
                     encoded_dictionaries,
                     dictionary_tracker,
                     write_options,
+                    dict_id,
                 )?;
             }
             DataType::Map(field, _) => {
@@ -285,6 +315,7 @@ impl IpcDataGenerator {
                     encoded_dictionaries,
                     dictionary_tracker,
                     write_options,
+                    dict_id,
                 )?;
 
                 // values
@@ -294,6 +325,7 @@ impl IpcDataGenerator {
                     encoded_dictionaries,
                     dictionary_tracker,
                     write_options,
+                    dict_id,
                 )?;
             }
             DataType::Union(fields, _) => {
@@ -306,6 +338,7 @@ impl IpcDataGenerator {
                         encoded_dictionaries,
                         dictionary_tracker,
                         write_options,
+                        dict_id,
                     )?;
                 }
             }
@@ -315,19 +348,24 @@ impl IpcDataGenerator {
         Ok(())
     }
 
-    fn encode_dictionaries(
+    fn encode_dictionaries<I: Iterator<Item = i64>>(
         &self,
         field: &Field,
         column: &ArrayRef,
         encoded_dictionaries: &mut Vec<EncodedData>,
         dictionary_tracker: &mut DictionaryTracker,
         write_options: &IpcWriteOptions,
+        dict_id_seq: &mut I,
     ) -> Result<(), ArrowError> {
         match column.data_type() {
             DataType::Dictionary(_key_type, _value_type) => {
-                let dict_id = field
-                    .dict_id()
-                    .expect("All Dictionary types have `dict_id`");
+                let dict_id = dict_id_seq
+                    .next()
+                    .or_else(|| field.dict_id())
+                    .ok_or_else(|| {
+                        ArrowError::IpcError(format!("no dict id for field {}", field.name()))
+                    })?;
+
                 let dict_data = column.to_data();
                 let dict_values = &dict_data.child_data()[0];
 
@@ -338,6 +376,7 @@ impl IpcDataGenerator {
                     encoded_dictionaries,
                     dictionary_tracker,
                     write_options,
+                    dict_id_seq,
                 )?;
 
                 let emit = dictionary_tracker.insert(dict_id, column)?;
@@ -355,6 +394,7 @@ impl IpcDataGenerator {
                 encoded_dictionaries,
                 dictionary_tracker,
                 write_options,
+                dict_id_seq,
             )?,
         }
 
@@ -373,6 +413,8 @@ impl IpcDataGenerator {
         let schema = batch.schema();
         let mut encoded_dictionaries = Vec::with_capacity(schema.all_fields().len());
 
+        let mut dict_id = dictionary_tracker.dict_ids.clone().into_iter();
+
         for (i, field) in schema.fields().iter().enumerate() {
             let column = batch.column(i);
             self.encode_dictionaries(
@@ -381,6 +423,7 @@ impl IpcDataGenerator {
                 &mut encoded_dictionaries,
                 dictionary_tracker,
                 write_options,
+                &mut dict_id,
             )?;
         }
 
@@ -671,20 +714,73 @@ fn into_zero_offset_run_array<R: RunEndIndexType>(
 /// isn't allowed in the `FileWriter`.
 pub struct DictionaryTracker {
     written: HashMap<i64, ArrayData>,
+    dict_ids: Vec<i64>,
     error_on_replacement: bool,
+    preserve_dict_id: bool,
 }
 
 impl DictionaryTracker {
-    /// Create a new [`DictionaryTracker`]. If `error_on_replacement`
+    /// Create a new [`DictionaryTracker`].
+    ///
+    /// If `error_on_replacement`
     /// is true, an error will be generated if an update to an
     /// existing dictionary is attempted.
+    ///
+    /// If `preserve_dict_id` is true, the dictionary ID defined in the schema
+    /// is used, otherwise a unique dictionary ID will be assigned by incrementing
+    /// the last seen dictionary ID (or using `0` if no other dictionary IDs have been
+    /// seen)
     pub fn new(error_on_replacement: bool) -> Self {
         Self {
             written: HashMap::new(),
+            dict_ids: Vec::new(),
             error_on_replacement,
+            preserve_dict_id: true,
         }
     }
 
+    /// Create a new [`DictionaryTracker`].
+    ///
+    /// If `error_on_replacement`
+    /// is true, an error will be generated if an update to an
+    /// existing dictionary is attempted.
+    pub fn new_with_preserve_dict_id(error_on_replacement: bool, preserve_dict_id: bool) -> Self {
+        Self {
+            written: HashMap::new(),
+            dict_ids: Vec::new(),
+            error_on_replacement,
+            preserve_dict_id,
+        }
+    }
+
+    /// Set the dictionary ID for `field`.
+    ///
+    /// If `preserve_dict_id` is true, this will return the `dict_id` in `field` (or panic if `field` does
+    /// not have a `dict_id` defined).
+    ///
+    /// If `preserve_dict_id` is false, this will return the value of the last `dict_id` assigned incremented by 1
+    /// or 0 in the case where no dictionary IDs have yet been assigned
+    pub fn set_dict_id(&mut self, field: &Field) -> i64 {
+        let next = if self.preserve_dict_id {
+            field.dict_id().expect("no dict_id in field")
+        } else {
+            self.dict_ids
+                .last()
+                .copied()
+                .map(|i| i + 1)
+                .unwrap_or_default()
+        };
+
+        self.dict_ids.push(next);
+        next
+    }
+
+    /// Return the sequence of dictionary IDs in the order they should be observed while
+    /// traversing the schema
+    pub fn dict_id(&mut self) -> &[i64] {
+        &self.dict_ids
+    }
+
     /// Keep track of the dictionary with the given ID and values. Behavior:
     ///
     /// * If this ID has been written already and has the same data, return `Ok(false)` to indicate
@@ -771,6 +867,7 @@ impl<W: Write> FileWriter<W> {
         // write the schema, set the written bytes to the schema + header
         let encoded_message = data_gen.schema_to_bytes(schema, &write_options);
         let (meta, data) = write_message(&mut writer, encoded_message, &write_options)?;
+        let preserve_dict_id = write_options.preserve_dict_id;
         Ok(Self {
             writer,
             write_options,
@@ -779,7 +876,10 @@ impl<W: Write> FileWriter<W> {
             dictionary_blocks: vec![],
             record_blocks: vec![],
             finished: false,
-            dictionary_tracker: DictionaryTracker::new(true),
+            dictionary_tracker: DictionaryTracker::new_with_preserve_dict_id(
+                true,
+                preserve_dict_id,
+            ),
             custom_metadata: HashMap::new(),
             data_gen,
         })
@@ -936,11 +1036,15 @@ impl<W: Write> StreamWriter<W> {
         // write the schema, set the written bytes to the schema
         let encoded_message = data_gen.schema_to_bytes(schema, &write_options);
         write_message(&mut writer, encoded_message, &write_options)?;
+        let preserve_dict_id = write_options.preserve_dict_id;
         Ok(Self {
             writer,
             write_options,
             finished: false,
-            dictionary_tracker: DictionaryTracker::new(false),
+            dictionary_tracker: DictionaryTracker::new_with_preserve_dict_id(
+                false,
+                preserve_dict_id,
+            ),
             data_gen,
         })
     }
@@ -1817,11 +1921,12 @@ mod tests {
         let batch = RecordBatch::try_new(schema, vec![Arc::new(union)]).unwrap();
 
         let gen = IpcDataGenerator {};
-        let mut dict_tracker = DictionaryTracker::new(false);
+        let mut dict_tracker = DictionaryTracker::new_with_preserve_dict_id(false, true);
         gen.encoded_batch(&batch, &mut dict_tracker, &Default::default())
             .unwrap();
 
-        // Dictionary with id 2 should have been written to the dict tracker
+        // The encoder will assign dict IDs itself to ensure uniqueness and ignore the dict ID in the schema
+        // so we expect the dict will be keyed to 0
         assert!(dict_tracker.written.contains_key(&2));
     }
 
@@ -1852,11 +1957,10 @@ mod tests {
         let batch = RecordBatch::try_new(schema, vec![struct_array]).unwrap();
 
         let gen = IpcDataGenerator {};
-        let mut dict_tracker = DictionaryTracker::new(false);
+        let mut dict_tracker = DictionaryTracker::new_with_preserve_dict_id(false, true);
         gen.encoded_batch(&batch, &mut dict_tracker, &Default::default())
             .unwrap();
 
-        // Dictionary with id 2 should have been written to the dict tracker
         assert!(dict_tracker.written.contains_key(&2));
     }
 
diff --git a/object_store/Cargo.toml b/object_store/Cargo.toml
index c18509aaf..5a0df233c 100644
--- a/object_store/Cargo.toml
+++ b/object_store/Cargo.toml
@@ -46,7 +46,7 @@ walkdir = "2"
 # Cloud storage support
 base64 = { version = "0.22", default-features = false, features = ["std"], optional = true }
 hyper = { version = "1.2", default-features = false, optional = true }
-quick-xml = { version = "0.34.0", features = ["serialize", "overlapped-lists"], optional = true }
+quick-xml = { version = "0.35.0", features = ["serialize", "overlapped-lists"], optional = true }
 serde = { version = "1.0", default-features = false, features = ["derive"], optional = true }
 serde_json = { version = "1.0", default-features = false, optional = true }
 rand = { version = "0.8", default-features = false, features = ["std", "std_rng"], optional = true }
diff --git a/object_store/src/attributes.rs b/object_store/src/attributes.rs
index ecef32eb4..11cf27c85 100644
--- a/object_store/src/attributes.rs
+++ b/object_store/src/attributes.rs
@@ -45,6 +45,10 @@ pub enum Attribute {
     ///
     /// See [Cache-Control](https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Cache-Control)
     CacheControl,
+    /// Specifies a user-defined metadata field for the object
+    ///
+    /// The String is a user-defined key
+    Metadata(Cow<'static, str>),
 }
 
 /// The value of an [`Attribute`]
@@ -194,10 +198,11 @@ mod tests {
             (Attribute::ContentLanguage, "en-US"),
             (Attribute::ContentType, "test"),
             (Attribute::CacheControl, "control"),
+            (Attribute::Metadata("key1".into()), "value1"),
         ]);
 
         assert!(!attributes.is_empty());
-        assert_eq!(attributes.len(), 5);
+        assert_eq!(attributes.len(), 6);
 
         assert_eq!(
             attributes.get(&Attribute::ContentType),
@@ -210,18 +215,18 @@ mod tests {
             attributes.insert(Attribute::CacheControl, "v1".into()),
             Some(metav)
         );
-        assert_eq!(attributes.len(), 5);
+        assert_eq!(attributes.len(), 6);
 
         assert_eq!(
             attributes.remove(&Attribute::CacheControl).unwrap(),
             "v1".into()
         );
-        assert_eq!(attributes.len(), 4);
+        assert_eq!(attributes.len(), 5);
 
         let metav: AttributeValue = "v2".into();
         attributes.insert(Attribute::CacheControl, metav.clone());
         assert_eq!(attributes.get(&Attribute::CacheControl), Some(&metav));
-        assert_eq!(attributes.len(), 5);
+        assert_eq!(attributes.len(), 6);
 
         assert_eq!(
             attributes.get(&Attribute::ContentDisposition),
@@ -235,5 +240,9 @@ mod tests {
             attributes.get(&Attribute::ContentLanguage),
             Some(&"en-US".into())
         );
+        assert_eq!(
+            attributes.get(&Attribute::Metadata("key1".into())),
+            Some(&"value1".into())
+        );
     }
 }
diff --git a/object_store/src/aws/client.rs b/object_store/src/aws/client.rs
index 98226c4b6..ab4da86f5 100644
--- a/object_store/src/aws/client.rs
+++ b/object_store/src/aws/client.rs
@@ -61,6 +61,7 @@ use std::sync::Arc;
 
 const VERSION_HEADER: &str = "x-amz-version-id";
 const SHA256_CHECKSUM: &str = "x-amz-checksum-sha256";
+const USER_DEFINED_METADATA_HEADER_PREFIX: &str = "x-amz-meta-";
 
 /// A specialized `Error` for object store-related errors
 #[derive(Debug, Snafu)]
@@ -326,6 +327,10 @@ impl<'a> Request<'a> {
                     has_content_type = true;
                     builder.header(CONTENT_TYPE, v.as_ref())
                 }
+                Attribute::Metadata(k_suffix) => builder.header(
+                    &format!("{}{}", USER_DEFINED_METADATA_HEADER_PREFIX, k_suffix),
+                    v.as_ref(),
+                ),
             };
         }
 
@@ -642,6 +647,7 @@ impl GetClient for S3Client {
         etag_required: false,
         last_modified_required: false,
         version_header: Some(VERSION_HEADER),
+        user_defined_metadata_prefix: Some(USER_DEFINED_METADATA_HEADER_PREFIX),
     };
 
     /// Make an S3 GET request <https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetObject.html>
diff --git a/object_store/src/azure/client.rs b/object_store/src/azure/client.rs
index be760c7dd..b5e82c2a8 100644
--- a/object_store/src/azure/client.rs
+++ b/object_store/src/azure/client.rs
@@ -48,6 +48,7 @@ use std::time::Duration;
 use url::Url;
 
 const VERSION_HEADER: &str = "x-ms-version-id";
+const USER_DEFINED_METADATA_HEADER_PREFIX: &str = "x-ms-meta-";
 static MS_CACHE_CONTROL: HeaderName = HeaderName::from_static("x-ms-blob-cache-control");
 static MS_CONTENT_TYPE: HeaderName = HeaderName::from_static("x-ms-blob-content-type");
 static MS_CONTENT_DISPOSITION: HeaderName =
@@ -208,6 +209,10 @@ impl<'a> PutRequest<'a> {
                     has_content_type = true;
                     builder.header(&MS_CONTENT_TYPE, v.as_ref())
                 }
+                Attribute::Metadata(k_suffix) => builder.header(
+                    &format!("{}{}", USER_DEFINED_METADATA_HEADER_PREFIX, k_suffix),
+                    v.as_ref(),
+                ),
             };
         }
 
@@ -499,6 +504,7 @@ impl GetClient for AzureClient {
         etag_required: true,
         last_modified_required: true,
         version_header: Some(VERSION_HEADER),
+        user_defined_metadata_prefix: Some(USER_DEFINED_METADATA_HEADER_PREFIX),
     };
 
     /// Make an Azure GET request
diff --git a/object_store/src/buffered.rs b/object_store/src/buffered.rs
index c7b71aa1c..fcd7e064e 100644
--- a/object_store/src/buffered.rs
+++ b/object_store/src/buffered.rs
@@ -438,7 +438,12 @@ impl AsyncWrite for BufWriter {
                 }
                 BufWriterState::Flush(f) => return f.poll_unpin(cx).map_err(std::io::Error::from),
                 BufWriterState::Write(x) => {
-                    let upload = x.take().unwrap();
+                    let upload = x.take().ok_or_else(|| {
+                        std::io::Error::new(
+                            ErrorKind::InvalidInput,
+                            "Cannot shutdown a writer that has already been shut down",
+                        )
+                    })?;
                     self.state = BufWriterState::Flush(
                         async move {
                             upload.finish().await?;
diff --git a/object_store/src/client/get.rs b/object_store/src/client/get.rs
index 430b87b1b..b45eaa143 100644
--- a/object_store/src/client/get.rs
+++ b/object_store/src/client/get.rs
@@ -135,6 +135,9 @@ enum GetResultError {
     #[snafu(display("Content-Type header contained non UTF-8 characters"))]
     InvalidContentType { source: ToStrError },
 
+    #[snafu(display("Metadata value for \"{key:?}\" contained non UTF-8 characters"))]
+    InvalidMetadata { key: String },
+
     #[snafu(display("Requested {expected:?}, got {actual:?}"))]
     UnexpectedRange {
         expected: Range<usize>,
@@ -192,7 +195,7 @@ fn get_result<T: GetClient>(
         }}
     }
 
-    let attributes = parse_attributes!(
+    let mut attributes = parse_attributes!(
         response.headers(),
         (
             CACHE_CONTROL,
@@ -221,6 +224,24 @@ fn get_result<T: GetClient>(
         )
     );
 
+    // Add attributes that match the user-defined metadata prefix (e.g. x-amz-meta-)
+    if let Some(prefix) = T::HEADER_CONFIG.user_defined_metadata_prefix {
+        for (key, val) in response.headers() {
+            if let Some(suffix) = key.as_str().strip_prefix(prefix) {
+                if let Ok(val_str) = val.to_str() {
+                    attributes.insert(
+                        Attribute::Metadata(suffix.to_string().into()),
+                        val_str.to_string().into(),
+                    );
+                } else {
+                    return Err(GetResultError::InvalidMetadata {
+                        key: key.to_string(),
+                    });
+                }
+            }
+        }
+    }
+
     let stream = response
         .bytes_stream()
         .map_err(|source| crate::Error::Generic {
@@ -253,6 +274,7 @@ mod tests {
             etag_required: false,
             last_modified_required: false,
             version_header: None,
+            user_defined_metadata_prefix: Some("x-test-meta-"),
         };
 
         async fn get_request(&self, _: &Path, _: GetOptions) -> Result<Response> {
@@ -265,6 +287,7 @@ mod tests {
         range: Option<Range<usize>>,
         status: StatusCode,
         content_range: Option<&str>,
+        headers: Option<Vec<(&str, &str)>>,
     ) -> Response {
         let mut builder = http::Response::builder();
         if let Some(range) = content_range {
@@ -276,6 +299,12 @@ mod tests {
             None => vec![0_u8; object_size],
         };
 
+        if let Some(headers) = headers {
+            for (key, value) in headers {
+                builder = builder.header(key, value);
+            }
+        }
+
         builder
             .status(status)
             .header(CONTENT_LENGTH, object_size)
@@ -288,7 +317,7 @@ mod tests {
     async fn test_get_result() {
         let path = Path::from("test");
 
-        let resp = make_response(12, None, StatusCode::OK, None);
+        let resp = make_response(12, None, StatusCode::OK, None, None);
         let res = get_result::<TestClient>(&path, None, resp).unwrap();
         assert_eq!(res.meta.size, 12);
         assert_eq!(res.range, 0..12);
@@ -302,6 +331,7 @@ mod tests {
             Some(2..3),
             StatusCode::PARTIAL_CONTENT,
             Some("bytes 2-2/12"),
+            None,
         );
         let res = get_result::<TestClient>(&path, Some(get_range.clone()), resp).unwrap();
         assert_eq!(res.meta.size, 12);
@@ -309,7 +339,7 @@ mod tests {
         let bytes = res.bytes().await.unwrap();
         assert_eq!(bytes.len(), 1);
 
-        let resp = make_response(12, Some(2..3), StatusCode::OK, None);
+        let resp = make_response(12, Some(2..3), StatusCode::OK, None, None);
         let err = get_result::<TestClient>(&path, Some(get_range.clone()), resp).unwrap_err();
         assert_eq!(
             err.to_string(),
@@ -321,6 +351,7 @@ mod tests {
             Some(2..3),
             StatusCode::PARTIAL_CONTENT,
             Some("bytes 2-3/12"),
+            None,
         );
         let err = get_result::<TestClient>(&path, Some(get_range.clone()), resp).unwrap_err();
         assert_eq!(err.to_string(), "Requested 2..3, got 2..4");
@@ -330,6 +361,7 @@ mod tests {
             Some(2..3),
             StatusCode::PARTIAL_CONTENT,
             Some("bytes 2-2/*"),
+            None,
         );
         let err = get_result::<TestClient>(&path, Some(get_range.clone()), resp).unwrap_err();
         assert_eq!(
@@ -337,7 +369,7 @@ mod tests {
             "Failed to parse value for CONTENT_RANGE header: \"bytes 2-2/*\""
         );
 
-        let resp = make_response(12, Some(2..3), StatusCode::PARTIAL_CONTENT, None);
+        let resp = make_response(12, Some(2..3), StatusCode::PARTIAL_CONTENT, None, None);
         let err = get_result::<TestClient>(&path, Some(get_range.clone()), resp).unwrap_err();
         assert_eq!(
             err.to_string(),
@@ -349,6 +381,7 @@ mod tests {
             Some(2..3),
             StatusCode::PARTIAL_CONTENT,
             Some("bytes 2-3/2"),
+            None,
         );
         let err = get_result::<TestClient>(&path, Some(get_range.clone()), resp).unwrap_err();
         assert_eq!(
@@ -361,6 +394,7 @@ mod tests {
             Some(2..6),
             StatusCode::PARTIAL_CONTENT,
             Some("bytes 2-5/6"),
+            None,
         );
         let res = get_result::<TestClient>(&path, Some(GetRange::Suffix(4)), resp).unwrap();
         assert_eq!(res.meta.size, 6);
@@ -373,8 +407,26 @@ mod tests {
             Some(2..6),
             StatusCode::PARTIAL_CONTENT,
             Some("bytes 2-3/6"),
+            None,
         );
         let err = get_result::<TestClient>(&path, Some(GetRange::Suffix(4)), resp).unwrap_err();
         assert_eq!(err.to_string(), "Requested 2..6, got 2..4");
+
+        let resp = make_response(
+            12,
+            None,
+            StatusCode::OK,
+            None,
+            Some(vec![("x-test-meta-foo", "bar")]),
+        );
+        let res = get_result::<TestClient>(&path, None, resp).unwrap();
+        assert_eq!(res.meta.size, 12);
+        assert_eq!(res.range, 0..12);
+        assert_eq!(
+            res.attributes.get(&Attribute::Metadata("foo".into())),
+            Some(&"bar".into())
+        );
+        let bytes = res.bytes().await.unwrap();
+        assert_eq!(bytes.len(), 12);
     }
 }
diff --git a/object_store/src/client/header.rs b/object_store/src/client/header.rs
index e85bf6ba5..9ce5db433 100644
--- a/object_store/src/client/header.rs
+++ b/object_store/src/client/header.rs
@@ -31,6 +31,7 @@ pub struct HeaderConfig {
     ///
     /// Defaults to `true`
     pub etag_required: bool,
+
     /// Whether to require a Last-Modified header when extracting [`ObjectMeta`] from headers.
     ///
     /// Defaults to `true`
@@ -38,6 +39,9 @@ pub struct HeaderConfig {
 
     /// The version header name if any
     pub version_header: Option<&'static str>,
+
+    /// The user defined metadata prefix if any
+    pub user_defined_metadata_prefix: Option<&'static str>,
 }
 
 #[derive(Debug, Snafu)]
diff --git a/object_store/src/gcp/client.rs b/object_store/src/gcp/client.rs
index 35c64cc3d..0045383ee 100644
--- a/object_store/src/gcp/client.rs
+++ b/object_store/src/gcp/client.rs
@@ -49,6 +49,7 @@ use std::sync::Arc;
 
 const VERSION_HEADER: &str = "x-goog-generation";
 const DEFAULT_CONTENT_TYPE: &str = "application/octet-stream";
+const USER_DEFINED_METADATA_HEADER_PREFIX: &str = "x-goog-meta-";
 
 static VERSION_MATCH: HeaderName = HeaderName::from_static("x-goog-if-generation-match");
 
@@ -199,6 +200,10 @@ impl<'a> Request<'a> {
                     has_content_type = true;
                     builder.header(CONTENT_TYPE, v.as_ref())
                 }
+                Attribute::Metadata(k_suffix) => builder.header(
+                    &format!("{}{}", USER_DEFINED_METADATA_HEADER_PREFIX, k_suffix),
+                    v.as_ref(),
+                ),
             };
         }
 
@@ -567,6 +572,7 @@ impl GetClient for GoogleCloudStorageClient {
         etag_required: true,
         last_modified_required: true,
         version_header: Some(VERSION_HEADER),
+        user_defined_metadata_prefix: Some(USER_DEFINED_METADATA_HEADER_PREFIX),
     };
 
     /// Perform a get request <https://cloud.google.com/storage/docs/xml-api/get-object-download>
diff --git a/object_store/src/http/client.rs b/object_store/src/http/client.rs
index 4dccef880..5def93120 100644
--- a/object_store/src/http/client.rs
+++ b/object_store/src/http/client.rs
@@ -183,6 +183,8 @@ impl Client {
                         has_content_type = true;
                         builder.header(CONTENT_TYPE, v.as_ref())
                     }
+                    // Ignore metadata attributes
+                    Attribute::Metadata(_) => builder,
                 };
             }
 
@@ -319,6 +321,7 @@ impl GetClient for Client {
         etag_required: false,
         last_modified_required: false,
         version_header: None,
+        user_defined_metadata_prefix: None,
     };
 
     async fn get_request(&self, path: &Path, options: GetOptions) -> Result<Response> {
diff --git a/object_store/src/integration.rs b/object_store/src/integration.rs
index 31b074f49..89b21bc61 100644
--- a/object_store/src/integration.rs
+++ b/object_store/src/integration.rs
@@ -480,6 +480,7 @@ pub async fn put_get_attributes(integration: &dyn ObjectStore) {
         (Attribute::ContentEncoding, "gzip"),
         (Attribute::ContentLanguage, "en-US"),
         (Attribute::ContentType, "text/html; charset=utf-8"),
+        (Attribute::Metadata("test_key".into()), "test_value"),
     ]);
 
     let path = Path::from("attributes");
diff --git a/object_store/src/lib.rs b/object_store/src/lib.rs
index 1339d1c01..efbfe0bd4 100644
--- a/object_store/src/lib.rs
+++ b/object_store/src/lib.rs
@@ -550,7 +550,7 @@ pub mod integration;
 
 pub use attributes::*;
 
-pub use parse::{parse_url, parse_url_opts};
+pub use parse::{parse_url, parse_url_opts, ObjectStoreScheme};
 pub use payload::*;
 pub use upload::*;
 pub use util::{coalesce_ranges, collect_bytes, GetRange, OBJECT_STORE_COALESCE_DEFAULT};
diff --git a/object_store/src/local.rs b/object_store/src/local.rs
index 1ce588af2..d3bfab8ed 100644
--- a/object_store/src/local.rs
+++ b/object_store/src/local.rs
@@ -724,7 +724,7 @@ struct LocalUpload {
 #[derive(Debug)]
 struct UploadState {
     dest: PathBuf,
-    file: Mutex<Option<File>>,
+    file: Mutex<File>,
 }
 
 impl LocalUpload {
@@ -732,7 +732,7 @@ impl LocalUpload {
         Self {
             state: Arc::new(UploadState {
                 dest,
-                file: Mutex::new(Some(file)),
+                file: Mutex::new(file),
             }),
             src: Some(src),
             offset: 0,
@@ -748,8 +748,7 @@ impl MultipartUpload for LocalUpload {
 
         let s = Arc::clone(&self.state);
         maybe_spawn_blocking(move || {
-            let mut f = s.file.lock();
-            let file = f.as_mut().context(AbortedSnafu)?;
+            let mut file = s.file.lock();
             file.seek(SeekFrom::Start(offset))
                 .context(SeekSnafu { path: &s.dest })?;
 
@@ -767,9 +766,9 @@ impl MultipartUpload for LocalUpload {
         let s = Arc::clone(&self.state);
         maybe_spawn_blocking(move || {
             // Ensure no inflight writes
-            let f = s.file.lock().take().context(AbortedSnafu)?;
+            let file = s.file.lock();
             std::fs::rename(&src, &s.dest).context(UnableToRenameFileSnafu)?;
-            let metadata = f.metadata().map_err(|e| Error::Metadata {
+            let metadata = file.metadata().map_err(|e| Error::Metadata {
                 source: e.into(),
                 path: src.to_string_lossy().to_string(),
             })?;
diff --git a/object_store/src/parse.rs b/object_store/src/parse.rs
index e5d5149fb..debc9e529 100644
--- a/object_store/src/parse.rs
+++ b/object_store/src/parse.rs
@@ -24,7 +24,7 @@ use snafu::Snafu;
 use url::Url;
 
 #[derive(Debug, Snafu)]
-enum Error {
+pub enum Error {
     #[snafu(display("Unable to recognise URL \"{}\"", url))]
     Unrecognised { url: Url },
 
@@ -41,9 +41,27 @@ impl From<Error> for super::Error {
     }
 }
 
-/// Recognises various URL formats, identifying the relevant [`ObjectStore`]
-#[derive(Debug, Eq, PartialEq)]
-enum ObjectStoreScheme {
+/// Recognizes various URL formats, identifying the relevant [`ObjectStore`]
+///
+/// See [`ObjectStoreScheme::parse`] for more details
+///
+/// # Supported formats:
+/// - `file:///path/to/my/file` -> [`LocalFileSystem`]
+/// - `memory:///` -> [`InMemory`]
+/// - `s3://bucket/path` -> [`AmazonS3`](crate::aws::AmazonS3) (also supports `s3a`)
+/// - `gs://bucket/path` -> [`GoogleCloudStorage`](crate::gcp::GoogleCloudStorage)
+/// - `az://account/container/path` -> [`MicrosoftAzure`](crate::azure::MicrosoftAzure) (also supports `adl`, `azure`, `abfs`, `abfss`)
+/// - `http://mydomain/path` -> [`HttpStore`](crate::http::HttpStore)
+/// - `https://mydomain/path` -> [`HttpStore`](crate::http::HttpStore)
+///
+/// There are also special cases for AWS and Azure for `https://{host?}/path` paths:
+/// - `dfs.core.windows.net`, `blob.core.windows.net`, `dfs.fabric.microsoft.com`, `blob.fabric.microsoft.com` -> [`MicrosoftAzure`](crate::azure::MicrosoftAzure)
+/// - `amazonaws.com` -> [`AmazonS3`](crate::aws::AmazonS3)
+/// - `r2.cloudflarestorage.com` -> [`AmazonS3`](crate::aws::AmazonS3)
+///
+#[non_exhaustive] // permit new variants
+#[derive(Debug, Eq, PartialEq, Clone)]
+pub enum ObjectStoreScheme {
     /// Url corresponding to [`LocalFileSystem`]
     Local,
     /// Url corresponding to [`InMemory`]
@@ -62,7 +80,27 @@ impl ObjectStoreScheme {
     /// Create an [`ObjectStoreScheme`] from the provided [`Url`]
     ///
     /// Returns the [`ObjectStoreScheme`] and the remaining [`Path`]
-    fn parse(url: &Url) -> Result<(Self, Path), Error> {
+    ///
+    /// # Example
+    /// ```
+    /// # use url::Url;
+    /// # use object_store::ObjectStoreScheme;
+    /// let url: Url = "file:///path/to/my/file".parse().unwrap();
+    /// let (scheme, path) = ObjectStoreScheme::parse(&url).unwrap();
+    /// assert_eq!(scheme, ObjectStoreScheme::Local);
+    /// assert_eq!(path.as_ref(), "path/to/my/file");
+    ///
+    /// let url: Url = "https://blob.core.windows.net/path/to/my/file".parse().unwrap();
+    /// let (scheme, path) = ObjectStoreScheme::parse(&url).unwrap();
+    /// assert_eq!(scheme, ObjectStoreScheme::MicrosoftAzure);
+    /// assert_eq!(path.as_ref(), "path/to/my/file");
+    ///
+    /// let url: Url = "https://example.com/path/to/my/file".parse().unwrap();
+    /// let (scheme, path) = ObjectStoreScheme::parse(&url).unwrap();
+    /// assert_eq!(scheme, ObjectStoreScheme::Http);
+    /// assert_eq!(path.as_ref(), "path/to/my/file");
+    /// ```
+    pub fn parse(url: &Url) -> Result<(Self, Path), Error> {
         let strip_bucket = || Some(url.path().strip_prefix('/')?.split_once('/')?.1);
 
         let (scheme, path) = match (url.scheme(), url.host_str()) {
diff --git a/object_store/src/upload.rs b/object_store/src/upload.rs
index dc499e2e0..4df4d8fd4 100644
--- a/object_store/src/upload.rs
+++ b/object_store/src/upload.rs
@@ -226,7 +226,15 @@ impl WriteMultipart {
         }
 
         self.wait_for_capacity(0).await?;
-        self.upload.complete().await
+
+        match self.upload.complete().await {
+            Err(e) => {
+                self.tasks.shutdown().await;
+                self.upload.abort().await?;
+                Err(e)
+            }
+            Ok(result) => Ok(result),
+        }
     }
 }
 
diff --git a/parquet/benches/arrow_reader.rs b/parquet/benches/arrow_reader.rs
index e1853d755..927998ac2 100644
--- a/parquet/benches/arrow_reader.rs
+++ b/parquet/benches/arrow_reader.rs
@@ -63,6 +63,8 @@ fn build_test_schema() -> SchemaDescPtr {
                     optional BYTE_ARRAY element (UTF8);
                 }
             }
+            REQUIRED BYTE_ARRAY mandatory_binary_leaf;
+            OPTIONAL BYTE_ARRAY optional_binary_leaf;
         }
         ";
     parse_message_type(message_type)
@@ -71,8 +73,8 @@ fn build_test_schema() -> SchemaDescPtr {
 }
 
 // test data params
-const NUM_ROW_GROUPS: usize = 1;
-const PAGES_PER_GROUP: usize = 2;
+const NUM_ROW_GROUPS: usize = 2;
+const PAGES_PER_GROUP: usize = 4;
 const VALUES_PER_PAGE: usize = 10_000;
 const BATCH_SIZE: usize = 8192;
 const MAX_LIST_LEN: usize = 10;
@@ -261,7 +263,7 @@ where
     InMemoryPageIterator::new(pages)
 }
 
-fn build_plain_encoded_string_page_iterator(
+fn build_plain_encoded_byte_array_page_iterator(
     column_desc: ColumnDescPtr,
     null_density: f32,
 ) -> impl PageIterator + Clone {
@@ -496,13 +498,20 @@ fn create_decimal_by_bytes_reader(
     }
 }
 
-fn create_string_byte_array_reader(
+fn create_byte_array_reader(
     page_iterator: impl PageIterator + 'static,
     column_desc: ColumnDescPtr,
 ) -> Box<dyn ArrayReader> {
     make_byte_array_reader(Box::new(page_iterator), column_desc, None).unwrap()
 }
 
+fn create_byte_view_array_reader(
+    page_iterator: impl PageIterator + 'static,
+    column_desc: ColumnDescPtr,
+) -> Box<dyn ArrayReader> {
+    make_byte_view_array_reader(Box::new(page_iterator), column_desc, None).unwrap()
+}
+
 fn create_string_view_byte_array_reader(
     page_iterator: impl PageIterator + 'static,
     column_desc: ColumnDescPtr,
@@ -525,7 +534,7 @@ fn create_string_list_reader(
     page_iterator: impl PageIterator + 'static,
     column_desc: ColumnDescPtr,
 ) -> Box<dyn ArrayReader> {
-    let items = create_string_byte_array_reader(page_iterator, column_desc);
+    let items = create_byte_array_reader(page_iterator, column_desc);
     let field = Field::new("item", DataType::Utf8, true);
     let data_type = DataType::List(Arc::new(field));
     Box::new(ListArrayReader::<i32>::new(items, data_type, 2, 1, true))
@@ -845,6 +854,8 @@ fn add_benches(c: &mut Criterion) {
     let mandatory_int64_column_desc = schema.column(4);
     let optional_int64_column_desc = schema.column(5);
     let string_list_desc = schema.column(14);
+    let mandatory_binary_column_desc = schema.column(15);
+    let optional_binary_column_desc = schema.column(16);
 
     // primitive / int32 benchmarks
     // =============================
@@ -879,10 +890,10 @@ fn add_benches(c: &mut Criterion) {
 
     // string, plain encoded, no NULLs
     let plain_string_no_null_data =
-        build_plain_encoded_string_page_iterator(mandatory_string_column_desc.clone(), 0.0);
+        build_plain_encoded_byte_array_page_iterator(mandatory_string_column_desc.clone(), 0.0);
     group.bench_function("plain encoded, mandatory, no NULLs", |b| {
         b.iter(|| {
-            let array_reader = create_string_byte_array_reader(
+            let array_reader = create_byte_array_reader(
                 plain_string_no_null_data.clone(),
                 mandatory_string_column_desc.clone(),
             );
@@ -892,10 +903,10 @@ fn add_benches(c: &mut Criterion) {
     });
 
     let plain_string_no_null_data =
-        build_plain_encoded_string_page_iterator(optional_string_column_desc.clone(), 0.0);
+        build_plain_encoded_byte_array_page_iterator(optional_string_column_desc.clone(), 0.0);
     group.bench_function("plain encoded, optional, no NULLs", |b| {
         b.iter(|| {
-            let array_reader = create_string_byte_array_reader(
+            let array_reader = create_byte_array_reader(
                 plain_string_no_null_data.clone(),
                 optional_string_column_desc.clone(),
             );
@@ -906,10 +917,10 @@ fn add_benches(c: &mut Criterion) {
 
     // string, plain encoded, half NULLs
     let plain_string_half_null_data =
-        build_plain_encoded_string_page_iterator(optional_string_column_desc.clone(), 0.5);
+        build_plain_encoded_byte_array_page_iterator(optional_string_column_desc.clone(), 0.5);
     group.bench_function("plain encoded, optional, half NULLs", |b| {
         b.iter(|| {
-            let array_reader = create_string_byte_array_reader(
+            let array_reader = create_byte_array_reader(
                 plain_string_half_null_data.clone(),
                 optional_string_column_desc.clone(),
             );
@@ -923,7 +934,7 @@ fn add_benches(c: &mut Criterion) {
         build_dictionary_encoded_string_page_iterator(mandatory_string_column_desc.clone(), 0.0);
     group.bench_function("dictionary encoded, mandatory, no NULLs", |b| {
         b.iter(|| {
-            let array_reader = create_string_byte_array_reader(
+            let array_reader = create_byte_array_reader(
                 dictionary_string_no_null_data.clone(),
                 mandatory_string_column_desc.clone(),
             );
@@ -936,7 +947,7 @@ fn add_benches(c: &mut Criterion) {
         build_dictionary_encoded_string_page_iterator(optional_string_column_desc.clone(), 0.0);
     group.bench_function("dictionary encoded, optional, no NULLs", |b| {
         b.iter(|| {
-            let array_reader = create_string_byte_array_reader(
+            let array_reader = create_byte_array_reader(
                 dictionary_string_no_null_data.clone(),
                 optional_string_column_desc.clone(),
             );
@@ -950,7 +961,7 @@ fn add_benches(c: &mut Criterion) {
         build_dictionary_encoded_string_page_iterator(optional_string_column_desc.clone(), 0.5);
     group.bench_function("dictionary encoded, optional, half NULLs", |b| {
         b.iter(|| {
-            let array_reader = create_string_byte_array_reader(
+            let array_reader = create_byte_array_reader(
                 dictionary_string_half_null_data.clone(),
                 optional_string_column_desc.clone(),
             );
@@ -961,6 +972,184 @@ fn add_benches(c: &mut Criterion) {
 
     group.finish();
 
+    // binary benchmarks
+    //==============================
+
+    let mut group = c.benchmark_group("arrow_array_reader/BinaryArray");
+
+    // byte array, plain encoded, no NULLs
+    let plain_byte_array_no_null_data =
+        build_plain_encoded_byte_array_page_iterator(mandatory_binary_column_desc.clone(), 0.0);
+    group.bench_function("plain encoded, mandatory, no NULLs", |b| {
+        b.iter(|| {
+            let array_reader = create_byte_array_reader(
+                plain_byte_array_no_null_data.clone(),
+                mandatory_binary_column_desc.clone(),
+            );
+            count = bench_array_reader(array_reader);
+        });
+        assert_eq!(count, EXPECTED_VALUE_COUNT);
+    });
+
+    let plain_byte_array_no_null_data =
+        build_plain_encoded_byte_array_page_iterator(optional_binary_column_desc.clone(), 0.0);
+    group.bench_function("plain encoded, optional, no NULLs", |b| {
+        b.iter(|| {
+            let array_reader = create_byte_array_reader(
+                plain_byte_array_no_null_data.clone(),
+                optional_binary_column_desc.clone(),
+            );
+            count = bench_array_reader(array_reader);
+        });
+        assert_eq!(count, EXPECTED_VALUE_COUNT);
+    });
+
+    // byte array, plain encoded, half NULLs
+    let plain_byte_array_half_null_data =
+        build_plain_encoded_byte_array_page_iterator(optional_binary_column_desc.clone(), 0.5);
+    group.bench_function("plain encoded, optional, half NULLs", |b| {
+        b.iter(|| {
+            let array_reader = create_byte_array_reader(
+                plain_byte_array_half_null_data.clone(),
+                optional_binary_column_desc.clone(),
+            );
+            count = bench_array_reader(array_reader);
+        });
+        assert_eq!(count, EXPECTED_VALUE_COUNT);
+    });
+
+    // byte array, dictionary encoded, no NULLs
+    let dictionary_byte_array_no_null_data =
+        build_dictionary_encoded_string_page_iterator(mandatory_binary_column_desc.clone(), 0.0);
+    group.bench_function("dictionary encoded, mandatory, no NULLs", |b| {
+        b.iter(|| {
+            let array_reader = create_byte_array_reader(
+                dictionary_byte_array_no_null_data.clone(),
+                mandatory_binary_column_desc.clone(),
+            );
+            count = bench_array_reader(array_reader);
+        });
+        assert_eq!(count, EXPECTED_VALUE_COUNT);
+    });
+
+    let dictionary_byte_array_no_null_data =
+        build_dictionary_encoded_string_page_iterator(optional_binary_column_desc.clone(), 0.0);
+    group.bench_function("dictionary encoded, optional, no NULLs", |b| {
+        b.iter(|| {
+            let array_reader = create_byte_array_reader(
+                dictionary_byte_array_no_null_data.clone(),
+                optional_binary_column_desc.clone(),
+            );
+            count = bench_array_reader(array_reader);
+        });
+        assert_eq!(count, EXPECTED_VALUE_COUNT);
+    });
+
+    // string, dictionary encoded, half NULLs
+    let dictionary_byte_array_half_null_data =
+        build_dictionary_encoded_string_page_iterator(optional_binary_column_desc.clone(), 0.5);
+    group.bench_function("dictionary encoded, optional, half NULLs", |b| {
+        b.iter(|| {
+            let array_reader = create_byte_array_reader(
+                dictionary_byte_array_half_null_data.clone(),
+                optional_binary_column_desc.clone(),
+            );
+            count = bench_array_reader(array_reader);
+        });
+        assert_eq!(count, EXPECTED_VALUE_COUNT);
+    });
+
+    group.finish();
+
+    // binary view benchmarks
+    //==============================
+
+    let mut group = c.benchmark_group("arrow_array_reader/BinaryViewArray");
+
+    // binary view, plain encoded, no NULLs
+    let plain_byte_array_no_null_data =
+        build_plain_encoded_byte_array_page_iterator(mandatory_binary_column_desc.clone(), 0.0);
+    group.bench_function("plain encoded, mandatory, no NULLs", |b| {
+        b.iter(|| {
+            let array_reader = create_byte_view_array_reader(
+                plain_byte_array_no_null_data.clone(),
+                mandatory_binary_column_desc.clone(),
+            );
+            count = bench_array_reader(array_reader);
+        });
+        assert_eq!(count, EXPECTED_VALUE_COUNT);
+    });
+
+    let plain_byte_array_no_null_data =
+        build_plain_encoded_byte_array_page_iterator(optional_binary_column_desc.clone(), 0.0);
+    group.bench_function("plain encoded, optional, no NULLs", |b| {
+        b.iter(|| {
+            let array_reader = create_byte_view_array_reader(
+                plain_byte_array_no_null_data.clone(),
+                optional_binary_column_desc.clone(),
+            );
+            count = bench_array_reader(array_reader);
+        });
+        assert_eq!(count, EXPECTED_VALUE_COUNT);
+    });
+
+    // binary view, plain encoded, half NULLs
+    let plain_byte_array_half_null_data =
+        build_plain_encoded_byte_array_page_iterator(optional_binary_column_desc.clone(), 0.5);
+    group.bench_function("plain encoded, optional, half NULLs", |b| {
+        b.iter(|| {
+            let array_reader = create_byte_view_array_reader(
+                plain_byte_array_half_null_data.clone(),
+                optional_binary_column_desc.clone(),
+            );
+            count = bench_array_reader(array_reader);
+        });
+        assert_eq!(count, EXPECTED_VALUE_COUNT);
+    });
+
+    // binary view, dictionary encoded, no NULLs
+    let dictionary_byte_array_no_null_data =
+        build_dictionary_encoded_string_page_iterator(mandatory_binary_column_desc.clone(), 0.0);
+    group.bench_function("dictionary encoded, mandatory, no NULLs", |b| {
+        b.iter(|| {
+            let array_reader = create_byte_view_array_reader(
+                dictionary_byte_array_no_null_data.clone(),
+                mandatory_binary_column_desc.clone(),
+            );
+            count = bench_array_reader(array_reader);
+        });
+        assert_eq!(count, EXPECTED_VALUE_COUNT);
+    });
+
+    let dictionary_byte_array_no_null_data =
+        build_dictionary_encoded_string_page_iterator(optional_binary_column_desc.clone(), 0.0);
+    group.bench_function("dictionary encoded, optional, no NULLs", |b| {
+        b.iter(|| {
+            let array_reader = create_byte_view_array_reader(
+                dictionary_byte_array_no_null_data.clone(),
+                optional_binary_column_desc.clone(),
+            );
+            count = bench_array_reader(array_reader);
+        });
+        assert_eq!(count, EXPECTED_VALUE_COUNT);
+    });
+
+    // binary view, dictionary encoded, half NULLs
+    let dictionary_byte_array_half_null_data =
+        build_dictionary_encoded_string_page_iterator(optional_binary_column_desc.clone(), 0.5);
+    group.bench_function("dictionary encoded, optional, half NULLs", |b| {
+        b.iter(|| {
+            let array_reader = create_byte_view_array_reader(
+                dictionary_byte_array_half_null_data.clone(),
+                optional_binary_column_desc.clone(),
+            );
+            count = bench_array_reader(array_reader);
+        });
+        assert_eq!(count, EXPECTED_VALUE_COUNT);
+    });
+
+    group.finish();
+
     // string dictionary benchmarks
     //==============================
 
@@ -969,7 +1158,7 @@ fn add_benches(c: &mut Criterion) {
     group.bench_function("dictionary encoded, mandatory, no NULLs", |b| {
         b.iter(|| {
             let array_reader = create_string_byte_array_dictionary_reader(
-                dictionary_string_no_null_data.clone(),
+                dictionary_byte_array_no_null_data.clone(),
                 mandatory_string_column_desc.clone(),
             );
             count = bench_array_reader(array_reader);
@@ -980,7 +1169,7 @@ fn add_benches(c: &mut Criterion) {
     group.bench_function("dictionary encoded, optional, no NULLs", |b| {
         b.iter(|| {
             let array_reader = create_string_byte_array_dictionary_reader(
-                dictionary_string_no_null_data.clone(),
+                dictionary_byte_array_no_null_data.clone(),
                 optional_string_column_desc.clone(),
             );
             count = bench_array_reader(array_reader);
@@ -991,7 +1180,7 @@ fn add_benches(c: &mut Criterion) {
     group.bench_function("dictionary encoded, optional, half NULLs", |b| {
         b.iter(|| {
             let array_reader = create_string_byte_array_dictionary_reader(
-                dictionary_string_half_null_data.clone(),
+                dictionary_byte_array_half_null_data.clone(),
                 optional_string_column_desc.clone(),
             );
             count = bench_array_reader(array_reader);
@@ -1008,7 +1197,7 @@ fn add_benches(c: &mut Criterion) {
 
     // string, plain encoded, no NULLs
     let plain_string_no_null_data =
-        build_plain_encoded_string_page_iterator(mandatory_string_column_desc.clone(), 0.0);
+        build_plain_encoded_byte_array_page_iterator(mandatory_string_column_desc.clone(), 0.0);
     group.bench_function("plain encoded, mandatory, no NULLs", |b| {
         b.iter(|| {
             let array_reader = create_string_view_byte_array_reader(
@@ -1021,7 +1210,7 @@ fn add_benches(c: &mut Criterion) {
     });
 
     let plain_string_no_null_data =
-        build_plain_encoded_string_page_iterator(optional_string_column_desc.clone(), 0.0);
+        build_plain_encoded_byte_array_page_iterator(optional_string_column_desc.clone(), 0.0);
     group.bench_function("plain encoded, optional, no NULLs", |b| {
         b.iter(|| {
             let array_reader = create_string_view_byte_array_reader(
@@ -1035,7 +1224,7 @@ fn add_benches(c: &mut Criterion) {
 
     // string, plain encoded, half NULLs
     let plain_string_half_null_data =
-        build_plain_encoded_string_page_iterator(optional_string_column_desc.clone(), 0.5);
+        build_plain_encoded_byte_array_page_iterator(optional_string_column_desc.clone(), 0.5);
     group.bench_function("plain encoded, optional, half NULLs", |b| {
         b.iter(|| {
             let array_reader = create_string_view_byte_array_reader(
diff --git a/parquet/src/arrow/array_reader/byte_view_array.rs b/parquet/src/arrow/array_reader/byte_view_array.rs
new file mode 100644
index 000000000..2d4e57414
--- /dev/null
+++ b/parquet/src/arrow/array_reader/byte_view_array.rs
@@ -0,0 +1,427 @@
+// 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 crate::arrow::array_reader::{read_records, skip_records, ArrayReader};
+use crate::arrow::buffer::view_buffer::ViewBuffer;
+use crate::arrow::record_reader::GenericRecordReader;
+use crate::arrow::schema::parquet_to_arrow_field;
+use crate::basic::{ConvertedType, Encoding};
+use crate::column::page::PageIterator;
+use crate::column::reader::decoder::ColumnValueDecoder;
+use crate::errors::{ParquetError, Result};
+use crate::schema::types::ColumnDescPtr;
+use arrow_array::ArrayRef;
+use arrow_schema::DataType as ArrowType;
+use bytes::Bytes;
+use std::any::Any;
+
+/// Returns an [`ArrayReader`] that decodes the provided byte array column to view types.
+#[allow(unused)]
+pub fn make_byte_view_array_reader(
+    pages: Box<dyn PageIterator>,
+    column_desc: ColumnDescPtr,
+    arrow_type: Option<ArrowType>,
+) -> Result<Box<dyn ArrayReader>> {
+    // Check if Arrow type is specified, else create it from Parquet type
+    let data_type = match arrow_type {
+        Some(t) => t,
+        None => match parquet_to_arrow_field(column_desc.as_ref())?.data_type() {
+            ArrowType::Utf8 | ArrowType::Utf8View => ArrowType::Utf8View,
+            _ => ArrowType::BinaryView,
+        },
+    };
+
+    match data_type {
+        ArrowType::BinaryView | ArrowType::Utf8View => {
+            let reader = GenericRecordReader::new(column_desc);
+            Ok(Box::new(ByteViewArrayReader::new(pages, data_type, reader)))
+        }
+
+        _ => Err(general_err!(
+            "invalid data type for byte array reader read to view type - {}",
+            data_type
+        )),
+    }
+}
+
+/// An [`ArrayReader`] for variable length byte arrays
+#[allow(unused)]
+struct ByteViewArrayReader {
+    data_type: ArrowType,
+    pages: Box<dyn PageIterator>,
+    def_levels_buffer: Option<Vec<i16>>,
+    rep_levels_buffer: Option<Vec<i16>>,
+    record_reader: GenericRecordReader<ViewBuffer, ByteViewArrayColumnValueDecoder>,
+}
+
+impl ByteViewArrayReader {
+    #[allow(unused)]
+    fn new(
+        pages: Box<dyn PageIterator>,
+        data_type: ArrowType,
+        record_reader: GenericRecordReader<ViewBuffer, ByteViewArrayColumnValueDecoder>,
+    ) -> Self {
+        Self {
+            data_type,
+            pages,
+            def_levels_buffer: None,
+            rep_levels_buffer: None,
+            record_reader,
+        }
+    }
+}
+
+impl ArrayReader for ByteViewArrayReader {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn get_data_type(&self) -> &ArrowType {
+        &self.data_type
+    }
+
+    fn read_records(&mut self, batch_size: usize) -> Result<usize> {
+        read_records(&mut self.record_reader, self.pages.as_mut(), batch_size)
+    }
+
+    fn consume_batch(&mut self) -> Result<ArrayRef> {
+        let buffer = self.record_reader.consume_record_data();
+        let null_buffer = self.record_reader.consume_bitmap_buffer();
+        self.def_levels_buffer = self.record_reader.consume_def_levels();
+        self.rep_levels_buffer = self.record_reader.consume_rep_levels();
+        self.record_reader.reset();
+
+        let array = buffer.into_array(null_buffer, &self.data_type);
+
+        Ok(array)
+    }
+
+    fn skip_records(&mut self, num_records: usize) -> Result<usize> {
+        skip_records(&mut self.record_reader, self.pages.as_mut(), num_records)
+    }
+
+    fn get_def_levels(&self) -> Option<&[i16]> {
+        self.def_levels_buffer.as_deref()
+    }
+
+    fn get_rep_levels(&self) -> Option<&[i16]> {
+        self.rep_levels_buffer.as_deref()
+    }
+}
+
+/// A [`ColumnValueDecoder`] for variable length byte arrays
+struct ByteViewArrayColumnValueDecoder {
+    dict: Option<ViewBuffer>,
+    decoder: Option<ByteViewArrayDecoder>,
+    validate_utf8: bool,
+}
+
+impl ColumnValueDecoder for ByteViewArrayColumnValueDecoder {
+    type Buffer = ViewBuffer;
+
+    fn new(desc: &ColumnDescPtr) -> Self {
+        let validate_utf8 = desc.converted_type() == ConvertedType::UTF8;
+        Self {
+            dict: None,
+            decoder: None,
+            validate_utf8,
+        }
+    }
+
+    fn set_dict(
+        &mut self,
+        buf: Bytes,
+        num_values: u32,
+        encoding: Encoding,
+        _is_sorted: bool,
+    ) -> Result<()> {
+        if !matches!(
+            encoding,
+            Encoding::PLAIN | Encoding::RLE_DICTIONARY | Encoding::PLAIN_DICTIONARY
+        ) {
+            return Err(nyi_err!(
+                "Invalid/Unsupported encoding type for dictionary: {}",
+                encoding
+            ));
+        }
+
+        let mut buffer = ViewBuffer::default();
+        let mut decoder = ByteViewArrayDecoderPlain::new(
+            buf,
+            num_values as usize,
+            Some(num_values as usize),
+            self.validate_utf8,
+        );
+        decoder.read(&mut buffer, usize::MAX)?;
+        self.dict = Some(buffer);
+        Ok(())
+    }
+
+    fn set_data(
+        &mut self,
+        encoding: Encoding,
+        data: Bytes,
+        num_levels: usize,
+        num_values: Option<usize>,
+    ) -> Result<()> {
+        self.decoder = Some(ByteViewArrayDecoder::new(
+            encoding,
+            data,
+            num_levels,
+            num_values,
+            self.validate_utf8,
+        )?);
+        Ok(())
+    }
+
+    fn read(&mut self, out: &mut Self::Buffer, num_values: usize) -> Result<usize> {
+        let decoder = self
+            .decoder
+            .as_mut()
+            .ok_or_else(|| general_err!("no decoder set"))?;
+
+        decoder.read(out, num_values, self.dict.as_ref())
+    }
+
+    fn skip_values(&mut self, num_values: usize) -> Result<usize> {
+        let decoder = self
+            .decoder
+            .as_mut()
+            .ok_or_else(|| general_err!("no decoder set"))?;
+
+        decoder.skip(num_values, self.dict.as_ref())
+    }
+}
+
+/// A generic decoder from uncompressed parquet value data to [`ViewBuffer`]
+pub enum ByteViewArrayDecoder {
+    Plain(ByteViewArrayDecoderPlain),
+}
+
+impl ByteViewArrayDecoder {
+    pub fn new(
+        encoding: Encoding,
+        data: Bytes,
+        num_levels: usize,
+        num_values: Option<usize>,
+        validate_utf8: bool,
+    ) -> Result<Self> {
+        let decoder = match encoding {
+            Encoding::PLAIN => ByteViewArrayDecoder::Plain(ByteViewArrayDecoderPlain::new(
+                data,
+                num_levels,
+                num_values,
+                validate_utf8,
+            )),
+            Encoding::RLE_DICTIONARY
+            | Encoding::PLAIN_DICTIONARY
+            | Encoding::DELTA_LENGTH_BYTE_ARRAY
+            | Encoding::DELTA_BYTE_ARRAY => unimplemented!("stay tuned!"),
+            _ => {
+                return Err(general_err!(
+                    "unsupported encoding for byte array: {}",
+                    encoding
+                ))
+            }
+        };
+
+        Ok(decoder)
+    }
+
+    /// Read up to `len` values to `out` with the optional dictionary
+    pub fn read(
+        &mut self,
+        out: &mut ViewBuffer,
+        len: usize,
+        _dict: Option<&ViewBuffer>,
+    ) -> Result<usize> {
+        match self {
+            ByteViewArrayDecoder::Plain(d) => d.read(out, len),
+        }
+    }
+
+    /// Skip `len` values
+    pub fn skip(&mut self, len: usize, _dict: Option<&ViewBuffer>) -> Result<usize> {
+        match self {
+            ByteViewArrayDecoder::Plain(d) => d.skip(len),
+        }
+    }
+}
+
+/// Decoder from [`Encoding::PLAIN`] data to [`ViewBuffer`]
+pub struct ByteViewArrayDecoderPlain {
+    buf: Bytes,
+    offset: usize,
+
+    validate_utf8: bool,
+
+    /// This is a maximum as the null count is not always known, e.g. value data from
+    /// a v1 data page
+    max_remaining_values: usize,
+}
+
+impl ByteViewArrayDecoderPlain {
+    pub fn new(
+        buf: Bytes,
+        num_levels: usize,
+        num_values: Option<usize>,
+        validate_utf8: bool,
+    ) -> Self {
+        Self {
+            buf,
+            offset: 0,
+            max_remaining_values: num_values.unwrap_or(num_levels),
+            validate_utf8,
+        }
+    }
+
+    pub fn read(&mut self, output: &mut ViewBuffer, len: usize) -> Result<usize> {
+        let block_id = output.append_block(self.buf.clone().into());
+
+        let to_read = len.min(self.max_remaining_values);
+
+        let buf = self.buf.as_ref();
+        let mut read = 0;
+        output.views.reserve(to_read);
+        while self.offset < self.buf.len() && read != to_read {
+            if self.offset + 4 > self.buf.len() {
+                return Err(ParquetError::EOF("eof decoding byte array".into()));
+            }
+            let len_bytes: [u8; 4] = unsafe {
+                buf.get_unchecked(self.offset..self.offset + 4)
+                    .try_into()
+                    .unwrap()
+            };
+            let len = u32::from_le_bytes(len_bytes);
+
+            let start_offset = self.offset + 4;
+            let end_offset = start_offset + len as usize;
+            if end_offset > buf.len() {
+                return Err(ParquetError::EOF("eof decoding byte array".into()));
+            }
+
+            if self.validate_utf8 {
+                check_valid_utf8(unsafe { buf.get_unchecked(start_offset..end_offset) })?;
+            }
+
+            unsafe {
+                output.append_view_unchecked(block_id, start_offset as u32, len);
+            }
+            self.offset = end_offset;
+            read += 1;
+        }
+
+        self.max_remaining_values -= to_read;
+        Ok(to_read)
+    }
+
+    pub fn skip(&mut self, to_skip: usize) -> Result<usize> {
+        let to_skip = to_skip.min(self.max_remaining_values);
+        let mut skip = 0;
+        let buf = self.buf.as_ref();
+
+        while self.offset < self.buf.len() && skip != to_skip {
+            if self.offset + 4 > buf.len() {
+                return Err(ParquetError::EOF("eof decoding byte array".into()));
+            }
+            let len_bytes: [u8; 4] = buf[self.offset..self.offset + 4].try_into().unwrap();
+            let len = u32::from_le_bytes(len_bytes) as usize;
+            skip += 1;
+            self.offset = self.offset + 4 + len;
+        }
+        self.max_remaining_values -= skip;
+        Ok(skip)
+    }
+}
+
+/// Check that `val` is a valid UTF-8 sequence
+pub fn check_valid_utf8(val: &[u8]) -> Result<()> {
+    match std::str::from_utf8(val) {
+        Ok(_) => Ok(()),
+        Err(e) => Err(general_err!("encountered non UTF-8 data: {}", e)),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use arrow_array::StringViewArray;
+    use arrow_buffer::Buffer;
+
+    use crate::{
+        arrow::{
+            array_reader::test_util::{byte_array_all_encodings, utf8_column},
+            buffer::view_buffer::ViewBuffer,
+            record_reader::buffer::ValuesBuffer,
+        },
+        basic::Encoding,
+        column::reader::decoder::ColumnValueDecoder,
+    };
+
+    use super::*;
+
+    #[test]
+    fn test_byte_array_string_view_decoder() {
+        let (pages, encoded_dictionary) =
+            byte_array_all_encodings(vec!["hello", "world", "large payload over 12 bytes", "b"]);
+
+        let column_desc = utf8_column();
+        let mut decoder = ByteViewArrayColumnValueDecoder::new(&column_desc);
+
+        decoder
+            .set_dict(encoded_dictionary, 4, Encoding::RLE_DICTIONARY, false)
+            .unwrap();
+
+        for (encoding, page) in pages {
+            if encoding != Encoding::PLAIN {
+                // skip non-plain encodings for now as they are not yet implemented
+                continue;
+            }
+            let mut output = ViewBuffer::default();
+            decoder.set_data(encoding, page, 4, Some(4)).unwrap();
+
+            assert_eq!(decoder.read(&mut output, 1).unwrap(), 1);
+            assert_eq!(decoder.read(&mut output, 1).unwrap(), 1);
+            assert_eq!(decoder.read(&mut output, 2).unwrap(), 2);
+            assert_eq!(decoder.read(&mut output, 4).unwrap(), 0);
+
+            assert_eq!(output.views.len(), 4);
+            assert_eq!(output.buffers.len(), 4);
+
+            let valid = [false, false, true, true, false, true, true, false, false];
+            let valid_buffer = Buffer::from_iter(valid.iter().cloned());
+
+            output.pad_nulls(0, 4, valid.len(), valid_buffer.as_slice());
+            let array = output.into_array(Some(valid_buffer), &ArrowType::Utf8View);
+            let strings = array.as_any().downcast_ref::<StringViewArray>().unwrap();
+
+            assert_eq!(
+                strings.iter().collect::<Vec<_>>(),
+                vec![
+                    None,
+                    None,
+                    Some("hello"),
+                    Some("world"),
+                    None,
+                    Some("large payload over 12 bytes"),
+                    Some("b"),
+                    None,
+                    None,
+                ]
+            );
+        }
+    }
+}
diff --git a/parquet/src/arrow/array_reader/mod.rs b/parquet/src/arrow/array_reader/mod.rs
index 8febe41e6..fef1e8722 100644
--- a/parquet/src/arrow/array_reader/mod.rs
+++ b/parquet/src/arrow/array_reader/mod.rs
@@ -32,6 +32,7 @@ use crate::file::reader::{FilePageIterator, FileReader};
 mod builder;
 mod byte_array;
 mod byte_array_dictionary;
+mod byte_view_array;
 mod empty_array;
 mod fixed_len_byte_array;
 mod fixed_size_list_array;
@@ -46,10 +47,10 @@ mod test_util;
 
 pub use builder::build_array_reader;
 pub use byte_array::make_byte_array_reader;
-#[allow(unused_imports)] // Only used for benchmarks
-pub use byte_array::make_byte_view_array_reader;
 pub use byte_array_dictionary::make_byte_array_dictionary_reader;
 #[allow(unused_imports)] // Only used for benchmarks
+pub use byte_view_array::make_byte_view_array_reader;
+#[allow(unused_imports)] // Only used for benchmarks
 pub use fixed_len_byte_array::make_fixed_len_byte_array_reader;
 pub use fixed_size_list_array::FixedSizeListArrayReader;
 pub use list_array::ListArrayReader;
diff --git a/parquet/src/arrow/arrow_reader/mod.rs b/parquet/src/arrow/arrow_reader/mod.rs
index 070dda97c..b38092dbc 100644
--- a/parquet/src/arrow/arrow_reader/mod.rs
+++ b/parquet/src/arrow/arrow_reader/mod.rs
@@ -25,25 +25,24 @@ use arrow_array::Array;
 use arrow_array::{RecordBatch, RecordBatchReader};
 use arrow_schema::{ArrowError, DataType as ArrowType, Schema, SchemaRef};
 use arrow_select::filter::prep_null_mask_filter;
+pub use filter::{ArrowPredicate, ArrowPredicateFn, RowFilter};
+pub use selection::{RowSelection, RowSelector};
 
+pub use crate::arrow::array_reader::RowGroups;
 use crate::arrow::array_reader::{build_array_reader, ArrayReader};
 use crate::arrow::schema::{parquet_to_arrow_schema_and_fields, ParquetField};
-use crate::arrow::{FieldLevels, ProjectionMask};
+use crate::arrow::{parquet_to_arrow_field_levels, FieldLevels, ProjectionMask};
+use crate::column::page::{PageIterator, PageReader};
 use crate::errors::{ParquetError, Result};
+use crate::file::footer;
 use crate::file::metadata::ParquetMetaData;
+use crate::file::page_index::index_reader;
 use crate::file::reader::{ChunkReader, SerializedPageReader};
 use crate::schema::types::SchemaDescriptor;
 
 mod filter;
 mod selection;
 
-pub use crate::arrow::array_reader::RowGroups;
-use crate::column::page::{PageIterator, PageReader};
-use crate::file::footer;
-use crate::file::page_index::index_reader;
-pub use filter::{ArrowPredicate, ArrowPredicateFn, RowFilter};
-pub use selection::{RowSelection, RowSelector};
-
 /// Builder for constructing parquet readers into arrow.
 ///
 /// Most users should use one of the following specializations:
@@ -250,6 +249,8 @@ impl<T> ArrowReaderBuilder<T> {
 pub struct ArrowReaderOptions {
     /// Should the reader strip any user defined metadata from the Arrow schema
     skip_arrow_metadata: bool,
+    /// If provided used as the schema for the file, otherwise the schema is read from the file
+    supplied_schema: Option<SchemaRef>,
     /// If true, attempt to read `OffsetIndex` and `ColumnIndex`
     pub(crate) page_index: bool,
 }
@@ -273,6 +274,60 @@ impl ArrowReaderOptions {
         }
     }
 
+    /// Provide a schema to use when reading the parquet file. If provided it
+    /// takes precedence over the schema inferred from the file or the schema defined
+    /// in the file's metadata. If the schema is not compatible with the file's
+    /// schema an error will be returned when constructing the builder.
+    ///
+    /// This option is only required if you want to cast columns to a different type.
+    /// For example, if you wanted to cast from an Int64 in the Parquet file to a Timestamp
+    /// in the Arrow schema.
+    ///
+    /// The supplied schema must have the same number of columns as the parquet schema and
+    /// the column names need to be the same.
+    ///
+    /// # Example
+    /// ```
+    /// use std::io::Bytes;
+    /// use std::sync::Arc;
+    /// use tempfile::tempfile;
+    /// use arrow_array::{ArrayRef, Int32Array, RecordBatch};
+    /// use arrow_schema::{DataType, Field, Schema, TimeUnit};
+    /// use parquet::arrow::arrow_reader::{ArrowReaderOptions, ParquetRecordBatchReaderBuilder};
+    /// use parquet::arrow::ArrowWriter;
+    ///
+    /// // Write data - schema is inferred from the data to be Int32
+    /// let file = tempfile().unwrap();
+    /// let batch = RecordBatch::try_from_iter(vec![
+    ///     ("col_1", Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef),
+    /// ]).unwrap();
+    /// let mut writer = ArrowWriter::try_new(file.try_clone().unwrap(), batch.schema(), None).unwrap();
+    /// writer.write(&batch).unwrap();
+    /// writer.close().unwrap();
+    ///
+    /// // Read the file back.
+    /// // Supply a schema that interprets the Int32 column as a Timestamp.
+    /// let supplied_schema = Arc::new(Schema::new(vec![
+    ///     Field::new("col_1", DataType::Timestamp(TimeUnit::Nanosecond, None), false)
+    /// ]));
+    /// let options = ArrowReaderOptions::new().with_schema(supplied_schema.clone());
+    /// let mut builder = ParquetRecordBatchReaderBuilder::try_new_with_options(
+    ///     file.try_clone().unwrap(),
+    ///     options
+    /// ).expect("Error if the schema is not compatible with the parquet file schema.");
+    ///
+    /// // Create the reader and read the data using the supplied schema.
+    /// let mut reader = builder.build().unwrap();
+    /// let _batch = reader.next().unwrap().unwrap();   
+    /// ```
+    pub fn with_schema(self, schema: SchemaRef) -> Self {
+        Self {
+            supplied_schema: Some(schema),
+            skip_arrow_metadata: true,
+            ..self
+        }
+    }
+
     /// Enable reading [`PageIndex`], if present (defaults to `false`)
     ///
     /// The `PageIndex` can be used to push down predicates to the parquet scan,
@@ -353,22 +408,77 @@ impl ArrowReaderMetadata {
     /// This function does not attempt to load the PageIndex if not present in the metadata.
     /// See [`Self::load`] for more details.
     pub fn try_new(metadata: Arc<ParquetMetaData>, options: ArrowReaderOptions) -> Result<Self> {
-        let kv_metadata = match options.skip_arrow_metadata {
-            true => None,
-            false => metadata.file_metadata().key_value_metadata(),
-        };
+        match options.supplied_schema {
+            Some(supplied_schema) => Self::with_supplied_schema(metadata, supplied_schema.clone()),
+            None => {
+                let kv_metadata = match options.skip_arrow_metadata {
+                    true => None,
+                    false => metadata.file_metadata().key_value_metadata(),
+                };
 
-        let (schema, fields) = parquet_to_arrow_schema_and_fields(
-            metadata.file_metadata().schema_descr(),
+                let (schema, fields) = parquet_to_arrow_schema_and_fields(
+                    metadata.file_metadata().schema_descr(),
+                    ProjectionMask::all(),
+                    kv_metadata,
+                )?;
+
+                Ok(Self {
+                    metadata,
+                    schema: Arc::new(schema),
+                    fields: fields.map(Arc::new),
+                })
+            }
+        }
+    }
+
+    fn with_supplied_schema(
+        metadata: Arc<ParquetMetaData>,
+        supplied_schema: SchemaRef,
+    ) -> Result<Self> {
+        let parquet_schema = metadata.file_metadata().schema_descr();
+        let field_levels = parquet_to_arrow_field_levels(
+            parquet_schema,
             ProjectionMask::all(),
-            kv_metadata,
+            Some(supplied_schema.fields()),
         )?;
-
-        Ok(Self {
-            metadata,
-            schema: Arc::new(schema),
-            fields: fields.map(Arc::new),
-        })
+        let fields = field_levels.fields;
+        let inferred_len = fields.len();
+        let supplied_len = supplied_schema.fields().len();
+        // Ensure the supplied schema has the same number of columns as the parquet schema.
+        // parquet_to_arrow_field_levels is expected to throw an error if the schemas have
+        // different lengths, but we check here to be safe.
+        if inferred_len != supplied_len {
+            Err(arrow_err!(format!(
+                "incompatible arrow schema, expected {} columns received {}",
+                inferred_len, supplied_len
+            )))
+        } else {
+            let diff_fields: Vec<_> = supplied_schema
+                .fields()
+                .iter()
+                .zip(fields.iter())
+                .filter_map(|(field1, field2)| {
+                    if field1 != field2 {
+                        Some(field1.name().clone())
+                    } else {
+                        None
+                    }
+                })
+                .collect();
+
+            if !diff_fields.is_empty() {
+                Err(ParquetError::ArrowError(format!(
+                    "incompatible arrow schema, the following fields could not be cast: [{}]",
+                    diff_fields.join(", ")
+                )))
+            } else {
+                Ok(Self {
+                    metadata,
+                    schema: supplied_schema,
+                    fields: field_levels.levels.map(Arc::new),
+                })
+            }
+        }
     }
 
     /// Returns a reference to the [`ParquetMetaData`] for this parquet file
@@ -842,7 +952,7 @@ mod tests {
     use arrow_array::*;
     use arrow_buffer::{i256, ArrowNativeType, Buffer, IntervalDayTime};
     use arrow_data::ArrayDataBuilder;
-    use arrow_schema::{ArrowError, DataType as ArrowDataType, Field, Fields, Schema};
+    use arrow_schema::{ArrowError, DataType as ArrowDataType, Field, Fields, Schema, SchemaRef};
     use arrow_select::concat::concat_batches;
 
     use crate::arrow::arrow_reader::{
@@ -2307,10 +2417,12 @@ mod tests {
     fn test_invalid_utf8_string_array() {
         test_invalid_utf8_string_array_inner::<i32>();
     }
+
     #[test]
     fn test_invalid_utf8_large_string_array() {
         test_invalid_utf8_string_array_inner::<i64>();
     }
+
     fn test_invalid_utf8_string_array_inner<O: OffsetSizeTrait>() {
         let cases = [
             (
@@ -2620,6 +2732,275 @@ mod tests {
         assert_eq!(reader.schema(), schema_without_metadata);
     }
 
+    fn write_parquet_from_iter<I, F>(value: I) -> File
+    where
+        I: IntoIterator<Item = (F, ArrayRef)>,
+        F: AsRef<str>,
+    {
+        let batch = RecordBatch::try_from_iter(value).unwrap();
+        let file = tempfile().unwrap();
+        let mut writer =
+            ArrowWriter::try_new(file.try_clone().unwrap(), batch.schema().clone(), None).unwrap();
+        writer.write(&batch).unwrap();
+        writer.close().unwrap();
+        file
+    }
+
+    fn run_schema_test_with_error<I, F>(value: I, schema: SchemaRef, expected_error: &str)
+    where
+        I: IntoIterator<Item = (F, ArrayRef)>,
+        F: AsRef<str>,
+    {
+        let file = write_parquet_from_iter(value);
+        let options_with_schema = ArrowReaderOptions::new().with_schema(schema.clone());
+        let builder = ParquetRecordBatchReaderBuilder::try_new_with_options(
+            file.try_clone().unwrap(),
+            options_with_schema,
+        );
+        assert_eq!(builder.err().unwrap().to_string(), expected_error);
+    }
+
+    #[test]
+    fn test_schema_too_few_columns() {
+        run_schema_test_with_error(
+            vec![
+                ("int64", Arc::new(Int64Array::from(vec![0])) as ArrayRef),
+                ("int32", Arc::new(Int32Array::from(vec![0])) as ArrayRef),
+            ],
+            Arc::new(Schema::new(vec![Field::new(
+                "int64",
+                ArrowDataType::Int64,
+                false,
+            )])),
+            "Arrow: incompatible arrow schema, expected 2 struct fields got 1",
+        );
+    }
+
+    #[test]
+    fn test_schema_too_many_columns() {
+        run_schema_test_with_error(
+            vec![("int64", Arc::new(Int64Array::from(vec![0])) as ArrayRef)],
+            Arc::new(Schema::new(vec![
+                Field::new("int64", ArrowDataType::Int64, false),
+                Field::new("int32", ArrowDataType::Int32, false),
+            ])),
+            "Arrow: incompatible arrow schema, expected 1 struct fields got 2",
+        );
+    }
+
+    #[test]
+    fn test_schema_mismatched_column_names() {
+        run_schema_test_with_error(
+            vec![("int64", Arc::new(Int64Array::from(vec![0])) as ArrayRef)],
+            Arc::new(Schema::new(vec![Field::new(
+                "other",
+                ArrowDataType::Int64,
+                false,
+            )])),
+            "Arrow: incompatible arrow schema, expected field named int64 got other",
+        );
+    }
+
+    #[test]
+    fn test_schema_incompatible_columns() {
+        run_schema_test_with_error(
+            vec![
+                (
+                    "col1_invalid",
+                    Arc::new(Int64Array::from(vec![0])) as ArrayRef,
+                ),
+                (
+                    "col2_valid",
+                    Arc::new(Int32Array::from(vec![0])) as ArrayRef,
+                ),
+                (
+                    "col3_invalid",
+                    Arc::new(Date64Array::from(vec![0])) as ArrayRef,
+                ),
+            ],
+            Arc::new(Schema::new(vec![
+                Field::new("col1_invalid", ArrowDataType::Int32, false),
+                Field::new("col2_valid", ArrowDataType::Int32, false),
+                Field::new("col3_invalid", ArrowDataType::Int32, false),
+            ])),
+            "Arrow: incompatible arrow schema, the following fields could not be cast: [col1_invalid, col3_invalid]",
+        );
+    }
+
+    #[test]
+    fn test_one_incompatible_nested_column() {
+        let nested_fields = Fields::from(vec![
+            Field::new("nested1_valid", ArrowDataType::Utf8, false),
+            Field::new("nested1_invalid", ArrowDataType::Int64, false),
+        ]);
+        let nested = StructArray::try_new(
+            nested_fields,
+            vec![
+                Arc::new(StringArray::from(vec!["a"])) as ArrayRef,
+                Arc::new(Int64Array::from(vec![0])) as ArrayRef,
+            ],
+            None,
+        )
+        .expect("struct array");
+        let supplied_nested_fields = Fields::from(vec![
+            Field::new("nested1_valid", ArrowDataType::Utf8, false),
+            Field::new("nested1_invalid", ArrowDataType::Int32, false),
+        ]);
+        run_schema_test_with_error(
+            vec![
+                ("col1", Arc::new(Int64Array::from(vec![0])) as ArrayRef),
+                ("col2", Arc::new(Int32Array::from(vec![0])) as ArrayRef),
+                ("nested", Arc::new(nested) as ArrayRef),
+            ],
+            Arc::new(Schema::new(vec![
+                Field::new("col1", ArrowDataType::Int64, false),
+                Field::new("col2", ArrowDataType::Int32, false),
+                Field::new(
+                    "nested",
+                    ArrowDataType::Struct(supplied_nested_fields),
+                    false,
+                ),
+            ])),
+            "Arrow: incompatible arrow schema, the following fields could not be cast: [nested]",
+        );
+    }
+
+    #[test]
+    fn test_with_schema() {
+        let nested_fields = Fields::from(vec![
+            Field::new("utf8_to_dict", ArrowDataType::Utf8, false),
+            Field::new("int64_to_ts_nano", ArrowDataType::Int64, false),
+        ]);
+
+        let nested_arrays: Vec<ArrayRef> = vec![
+            Arc::new(StringArray::from(vec!["a", "a", "a", "b"])) as ArrayRef,
+            Arc::new(Int64Array::from(vec![1, 2, 3, 4])) as ArrayRef,
+        ];
+
+        let nested = StructArray::try_new(nested_fields, nested_arrays, None).unwrap();
+
+        let file = write_parquet_from_iter(vec![
+            (
+                "int32_to_ts_second",
+                Arc::new(Int32Array::from(vec![0, 1, 2, 3])) as ArrayRef,
+            ),
+            (
+                "date32_to_date64",
+                Arc::new(Date32Array::from(vec![0, 1, 2, 3])) as ArrayRef,
+            ),
+            ("nested", Arc::new(nested) as ArrayRef),
+        ]);
+
+        let supplied_nested_fields = Fields::from(vec![
+            Field::new(
+                "utf8_to_dict",
+                ArrowDataType::Dictionary(
+                    Box::new(ArrowDataType::Int32),
+                    Box::new(ArrowDataType::Utf8),
+                ),
+                false,
+            ),
+            Field::new(
+                "int64_to_ts_nano",
+                ArrowDataType::Timestamp(
+                    arrow::datatypes::TimeUnit::Nanosecond,
+                    Some("+10:00".into()),
+                ),
+                false,
+            ),
+        ]);
+
+        let supplied_schema = Arc::new(Schema::new(vec![
+            Field::new(
+                "int32_to_ts_second",
+                ArrowDataType::Timestamp(arrow::datatypes::TimeUnit::Second, Some("+01:00".into())),
+                false,
+            ),
+            Field::new("date32_to_date64", ArrowDataType::Date64, false),
+            Field::new(
+                "nested",
+                ArrowDataType::Struct(supplied_nested_fields),
+                false,
+            ),
+        ]));
+
+        let options = ArrowReaderOptions::new().with_schema(supplied_schema.clone());
+        let mut arrow_reader = ParquetRecordBatchReaderBuilder::try_new_with_options(
+            file.try_clone().unwrap(),
+            options,
+        )
+        .expect("reader builder with schema")
+        .build()
+        .expect("reader with schema");
+
+        assert_eq!(arrow_reader.schema(), supplied_schema);
+        let batch = arrow_reader.next().unwrap().unwrap();
+        assert_eq!(batch.num_columns(), 3);
+        assert_eq!(batch.num_rows(), 4);
+        assert_eq!(
+            batch
+                .column(0)
+                .as_any()
+                .downcast_ref::<TimestampSecondArray>()
+                .expect("downcast to timestamp second")
+                .value_as_datetime_with_tz(0, "+01:00".parse().unwrap())
+                .map(|v| v.to_string())
+                .expect("value as datetime"),
+            "1970-01-01 01:00:00 +01:00"
+        );
+        assert_eq!(
+            batch
+                .column(1)
+                .as_any()
+                .downcast_ref::<Date64Array>()
+                .expect("downcast to date64")
+                .value_as_date(0)
+                .map(|v| v.to_string())
+                .expect("value as date"),
+            "1970-01-01"
+        );
+
+        let nested = batch
+            .column(2)
+            .as_any()
+            .downcast_ref::<StructArray>()
+            .expect("downcast to struct");
+
+        let nested_dict = nested
+            .column(0)
+            .as_any()
+            .downcast_ref::<Int32DictionaryArray>()
+            .expect("downcast to dictionary");
+
+        assert_eq!(
+            nested_dict
+                .values()
+                .as_any()
+                .downcast_ref::<StringArray>()
+                .expect("downcast to string")
+                .iter()
+                .collect::<Vec<_>>(),
+            vec![Some("a"), Some("b")]
+        );
+
+        assert_eq!(
+            nested_dict.keys().iter().collect::<Vec<_>>(),
+            vec![Some(0), Some(0), Some(0), Some(1)]
+        );
+
+        assert_eq!(
+            nested
+                .column(1)
+                .as_any()
+                .downcast_ref::<TimestampNanosecondArray>()
+                .expect("downcast to timestamp nanosecond")
+                .value_as_datetime_with_tz(0, "+10:00".parse().unwrap())
+                .map(|v| v.to_string())
+                .expect("value as datetime"),
+            "1970-01-01 10:00:00.000000001 +10:00"
+        );
+    }
+
     #[test]
     fn test_empty_projection() {
         let testdata = arrow::util::test_util::parquet_test_data();
diff --git a/parquet/src/arrow/buffer/mod.rs b/parquet/src/arrow/buffer/mod.rs
index cbc795d94..e7566d9a5 100644
--- a/parquet/src/arrow/buffer/mod.rs
+++ b/parquet/src/arrow/buffer/mod.rs
@@ -20,3 +20,4 @@
 pub mod bit_util;
 pub mod dictionary_buffer;
 pub mod offset_buffer;
+pub mod view_buffer;
diff --git a/parquet/src/arrow/buffer/view_buffer.rs b/parquet/src/arrow/buffer/view_buffer.rs
new file mode 100644
index 000000000..01e7c4aad
--- /dev/null
+++ b/parquet/src/arrow/buffer/view_buffer.rs
@@ -0,0 +1,183 @@
+// 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 crate::arrow::record_reader::buffer::ValuesBuffer;
+use arrow_array::{builder::make_view, make_array, ArrayRef};
+use arrow_buffer::Buffer;
+use arrow_data::ArrayDataBuilder;
+use arrow_schema::DataType as ArrowType;
+
+/// A buffer of view type byte arrays that can be converted into
+/// `GenericByteViewArray`
+///
+/// Note this does not reuse `GenericByteViewBuilder` due to the need to call `pad_nulls`
+/// and reuse the existing logic for Vec in the parquet crate
+#[derive(Debug, Default)]
+pub struct ViewBuffer {
+    pub views: Vec<u128>,
+    pub buffers: Vec<Buffer>,
+}
+
+impl ViewBuffer {
+    #[allow(unused)]
+    pub fn append_block(&mut self, block: Buffer) -> u32 {
+        let block_id = self.buffers.len() as u32;
+        self.buffers.push(block);
+        block_id
+    }
+
+    /// # Safety
+    /// This method is only safe when:
+    /// - `block` is a valid index, i.e., the return value of `append_block`
+    /// - `offset` and `offset + len` are valid indices into the buffer
+    /// - The `(offset, offset + len)` is valid value for the native type.
+    #[allow(unused)]
+    pub unsafe fn append_view_unchecked(&mut self, block: u32, offset: u32, len: u32) {
+        let b = self.buffers.get_unchecked(block as usize);
+        let end = offset.saturating_add(len);
+        let b = b.get_unchecked(offset as usize..end as usize);
+
+        let view = make_view(b, block, offset);
+
+        self.views.push(view);
+    }
+
+    /// Converts this into an [`ArrayRef`] with the provided `data_type` and `null_buffer`
+    #[allow(unused)]
+    pub fn into_array(self, null_buffer: Option<Buffer>, data_type: &ArrowType) -> ArrayRef {
+        let len = self.views.len();
+        let views = Buffer::from_vec(self.views);
+        match data_type {
+            ArrowType::Utf8View => {
+                let builder = ArrayDataBuilder::new(ArrowType::Utf8View)
+                    .len(len)
+                    .add_buffer(views)
+                    .add_buffers(self.buffers)
+                    .null_bit_buffer(null_buffer);
+                // We have checked that the data is utf8 when building the buffer, so it is safe
+                let array = unsafe { builder.build_unchecked() };
+                make_array(array)
+            }
+            ArrowType::BinaryView => {
+                let builder = ArrayDataBuilder::new(ArrowType::BinaryView)
+                    .len(len)
+                    .add_buffer(views)
+                    .add_buffers(self.buffers)
+                    .null_bit_buffer(null_buffer);
+                let array = unsafe { builder.build_unchecked() };
+                make_array(array)
+            }
+            _ => panic!("Unsupported data type: {:?}", data_type),
+        }
+    }
+}
+
+impl ValuesBuffer for ViewBuffer {
+    fn pad_nulls(
+        &mut self,
+        read_offset: usize,
+        values_read: usize,
+        levels_read: usize,
+        valid_mask: &[u8],
+    ) {
+        self.views
+            .pad_nulls(read_offset, values_read, levels_read, valid_mask);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+
+    use arrow_array::Array;
+
+    use super::*;
+
+    #[test]
+    fn test_view_buffer_empty() {
+        let buffer = ViewBuffer::default();
+        let array = buffer.into_array(None, &ArrowType::Utf8View);
+        let strings = array
+            .as_any()
+            .downcast_ref::<arrow::array::StringViewArray>()
+            .unwrap();
+        assert_eq!(strings.len(), 0);
+    }
+
+    #[test]
+    fn test_view_buffer_append_view() {
+        let mut buffer = ViewBuffer::default();
+        let string_buffer = Buffer::from(&b"0123456789long string to test string view"[..]);
+        let block_id = buffer.append_block(string_buffer);
+
+        unsafe {
+            buffer.append_view_unchecked(block_id, 0, 1);
+            buffer.append_view_unchecked(block_id, 1, 9);
+            buffer.append_view_unchecked(block_id, 10, 31);
+        }
+
+        let array = buffer.into_array(None, &ArrowType::Utf8View);
+        let string_array = array
+            .as_any()
+            .downcast_ref::<arrow::array::StringViewArray>()
+            .unwrap();
+        assert_eq!(
+            string_array.iter().collect::<Vec<_>>(),
+            vec![
+                Some("0"),
+                Some("123456789"),
+                Some("long string to test string view"),
+            ]
+        );
+    }
+
+    #[test]
+    fn test_view_buffer_pad_null() {
+        let mut buffer = ViewBuffer::default();
+        let string_buffer = Buffer::from(&b"0123456789long string to test string view"[..]);
+        let block_id = buffer.append_block(string_buffer);
+
+        unsafe {
+            buffer.append_view_unchecked(block_id, 0, 1);
+            buffer.append_view_unchecked(block_id, 1, 9);
+            buffer.append_view_unchecked(block_id, 10, 31);
+        }
+
+        let valid = [true, false, false, true, false, false, true];
+        let valid_mask = Buffer::from_iter(valid.iter().copied());
+
+        buffer.pad_nulls(1, 2, valid.len() - 1, valid_mask.as_slice());
+
+        let array = buffer.into_array(Some(valid_mask), &ArrowType::Utf8View);
+        let strings = array
+            .as_any()
+            .downcast_ref::<arrow::array::StringViewArray>()
+            .unwrap();
+
+        assert_eq!(
+            strings.iter().collect::<Vec<_>>(),
+            vec![
+                Some("0"),
+                None,
+                None,
+                Some("123456789"),
+                None,
+                None,
+                Some("long string to test string view"),
+            ]
+        );
+    }
+}
diff --git a/parquet/src/data_type.rs b/parquet/src/data_type.rs
index 5e1d53bad..5bcd2062c 100644
--- a/parquet/src/data_type.rs
+++ b/parquet/src/data_type.rs
@@ -586,9 +586,9 @@ pub(crate) mod private {
     use crate::encodings::decoding::PlainDecoderDetails;
     use crate::util::bit_util::{read_num_bytes, BitReader, BitWriter};
 
-    use crate::basic::Type;
-
     use super::{ParquetError, Result, SliceAsBytes};
+    use crate::basic::Type;
+    use crate::file::metadata::HeapSize;
 
     /// Sealed trait to start to remove specialisation from implementations
     ///
@@ -606,6 +606,7 @@ pub(crate) mod private {
         + SliceAsBytes
         + PartialOrd
         + Send
+        + HeapSize
         + crate::encodings::decoding::private::GetDecoder
         + crate::file::statistics::private::MakeStatistics
     {
@@ -886,6 +887,12 @@ pub(crate) mod private {
         }
     }
 
+    impl HeapSize for super::Int96 {
+        fn heap_size(&self) -> usize {
+            0 // no heap allocations
+        }
+    }
+
     impl ParquetValueType for super::ByteArray {
         const PHYSICAL_TYPE: Type = Type::BYTE_ARRAY;
 
@@ -970,6 +977,15 @@ pub(crate) mod private {
         }
     }
 
+    impl HeapSize for super::ByteArray {
+        fn heap_size(&self) -> usize {
+            // note: this is an estimate, not exact, so just return the size
+            // of the actual data used, don't try to handle the fact that it may
+            // be shared.
+            self.data.as_ref().map(|data| data.len()).unwrap_or(0)
+        }
+    }
+
     impl ParquetValueType for super::FixedLenByteArray {
         const PHYSICAL_TYPE: Type = Type::FIXED_LEN_BYTE_ARRAY;
 
@@ -1055,6 +1071,12 @@ pub(crate) mod private {
             self
         }
     }
+
+    impl HeapSize for super::FixedLenByteArray {
+        fn heap_size(&self) -> usize {
+            self.0.heap_size()
+        }
+    }
 }
 
 /// Contains the Parquet physical type information as well as the Rust primitive type
diff --git a/parquet/src/file/metadata/memory.rs b/parquet/src/file/metadata/memory.rs
new file mode 100644
index 000000000..57b2f7eec
--- /dev/null
+++ b/parquet/src/file/metadata/memory.rs
@@ -0,0 +1,228 @@
+// 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.
+
+//! Memory calculations for [`ParquetMetadata::memory_size`]
+//!
+//! [`ParquetMetadata::memory_size`]: crate::file::metadata::ParquetMetaData::memory_size
+use crate::basic::{ColumnOrder, Compression, Encoding, PageType};
+use crate::data_type::private::ParquetValueType;
+use crate::file::metadata::{ColumnChunkMetaData, FileMetaData, KeyValue, RowGroupMetaData};
+use crate::file::page_encoding_stats::PageEncodingStats;
+use crate::file::page_index::index::{Index, NativeIndex, PageIndex};
+use crate::file::statistics::{Statistics, ValueStatistics};
+use crate::format::{BoundaryOrder, PageLocation, SortingColumn};
+use std::sync::Arc;
+
+/// Trait for calculating the size of various containers
+pub trait HeapSize {
+    /// Return the size of any bytes allocated on the heap by this object,
+    /// including heap memory in those structures
+    ///
+    /// Note that the size of the type itself is not included in the result --
+    /// instead, that size is added by the caller (e.g. container).
+    fn heap_size(&self) -> usize;
+}
+
+impl<T: HeapSize> HeapSize for Vec<T> {
+    fn heap_size(&self) -> usize {
+        let item_size = std::mem::size_of::<T>();
+        // account for the contents of the Vec
+        (self.capacity() * item_size) +
+        // add any heap allocations by contents
+        self.iter().map(|t| t.heap_size()).sum::<usize>()
+    }
+}
+
+impl<T: HeapSize> HeapSize for Arc<T> {
+    fn heap_size(&self) -> usize {
+        self.as_ref().heap_size()
+    }
+}
+
+impl<T: HeapSize> HeapSize for Option<T> {
+    fn heap_size(&self) -> usize {
+        self.as_ref().map(|inner| inner.heap_size()).unwrap_or(0)
+    }
+}
+
+impl HeapSize for String {
+    fn heap_size(&self) -> usize {
+        self.capacity()
+    }
+}
+
+impl HeapSize for FileMetaData {
+    fn heap_size(&self) -> usize {
+        self.created_by.heap_size()
+            + self.key_value_metadata.heap_size()
+            + self.schema_descr.heap_size()
+            + self.column_orders.heap_size()
+    }
+}
+
+impl HeapSize for KeyValue {
+    fn heap_size(&self) -> usize {
+        self.key.heap_size() + self.value.heap_size()
+    }
+}
+
+impl HeapSize for RowGroupMetaData {
+    fn heap_size(&self) -> usize {
+        // don't count schema_descr here because it is already
+        // counted in FileMetaData
+        self.columns.heap_size() + self.sorting_columns.heap_size()
+    }
+}
+
+impl HeapSize for ColumnChunkMetaData {
+    fn heap_size(&self) -> usize {
+        // don't count column_descr here because it is already counted in
+        // FileMetaData
+        self.encodings.heap_size()
+            + self.file_path.heap_size()
+            + self.compression.heap_size()
+            + self.statistics.heap_size()
+            + self.encoding_stats.heap_size()
+    }
+}
+
+impl HeapSize for Encoding {
+    fn heap_size(&self) -> usize {
+        0 // no heap allocations
+    }
+}
+
+impl HeapSize for PageEncodingStats {
+    fn heap_size(&self) -> usize {
+        self.page_type.heap_size() + self.encoding.heap_size()
+    }
+}
+
+impl HeapSize for SortingColumn {
+    fn heap_size(&self) -> usize {
+        0 // no heap allocations
+    }
+}
+impl HeapSize for Compression {
+    fn heap_size(&self) -> usize {
+        0 // no heap allocations
+    }
+}
+
+impl HeapSize for PageType {
+    fn heap_size(&self) -> usize {
+        0 // no heap allocations
+    }
+}
+impl HeapSize for Statistics {
+    fn heap_size(&self) -> usize {
+        match self {
+            Statistics::Boolean(value_statistics) => value_statistics.heap_size(),
+            Statistics::Int32(value_statistics) => value_statistics.heap_size(),
+            Statistics::Int64(value_statistics) => value_statistics.heap_size(),
+            Statistics::Int96(value_statistics) => value_statistics.heap_size(),
+            Statistics::Float(value_statistics) => value_statistics.heap_size(),
+            Statistics::Double(value_statistics) => value_statistics.heap_size(),
+            Statistics::ByteArray(value_statistics) => value_statistics.heap_size(),
+            Statistics::FixedLenByteArray(value_statistics) => value_statistics.heap_size(),
+        }
+    }
+}
+
+impl HeapSize for Index {
+    fn heap_size(&self) -> usize {
+        match self {
+            Index::NONE => 0,
+            Index::BOOLEAN(native_index) => native_index.heap_size(),
+            Index::INT32(native_index) => native_index.heap_size(),
+            Index::INT64(native_index) => native_index.heap_size(),
+            Index::INT96(native_index) => native_index.heap_size(),
+            Index::FLOAT(native_index) => native_index.heap_size(),
+            Index::DOUBLE(native_index) => native_index.heap_size(),
+            Index::BYTE_ARRAY(native_index) => native_index.heap_size(),
+            Index::FIXED_LEN_BYTE_ARRAY(native_index) => native_index.heap_size(),
+        }
+    }
+}
+
+impl<T: ParquetValueType> HeapSize for NativeIndex<T> {
+    fn heap_size(&self) -> usize {
+        self.indexes.heap_size() + self.boundary_order.heap_size()
+    }
+}
+
+impl<T: ParquetValueType> HeapSize for PageIndex<T> {
+    fn heap_size(&self) -> usize {
+        self.min.heap_size() + self.max.heap_size() + self.null_count.heap_size()
+    }
+}
+
+impl<T: ParquetValueType> HeapSize for ValueStatistics<T> {
+    fn heap_size(&self) -> usize {
+        self.min().heap_size() + self.max().heap_size()
+    }
+}
+impl HeapSize for bool {
+    fn heap_size(&self) -> usize {
+        0 // no heap allocations
+    }
+}
+impl HeapSize for i32 {
+    fn heap_size(&self) -> usize {
+        0 // no heap allocations
+    }
+}
+impl HeapSize for i64 {
+    fn heap_size(&self) -> usize {
+        0 // no heap allocations
+    }
+}
+
+impl HeapSize for f32 {
+    fn heap_size(&self) -> usize {
+        0 // no heap allocations
+    }
+}
+impl HeapSize for f64 {
+    fn heap_size(&self) -> usize {
+        0 // no heap allocations
+    }
+}
+
+impl HeapSize for usize {
+    fn heap_size(&self) -> usize {
+        0 // no heap allocations
+    }
+}
+
+impl HeapSize for BoundaryOrder {
+    fn heap_size(&self) -> usize {
+        0 // no heap allocations
+    }
+}
+
+impl HeapSize for PageLocation {
+    fn heap_size(&self) -> usize {
+        0 // no heap allocations
+    }
+}
+
+impl HeapSize for ColumnOrder {
+    fn heap_size(&self) -> usize {
+        0 // no heap allocations in ColumnOrder
+    }
+}
diff --git a/parquet/src/file/metadata.rs b/parquet/src/file/metadata/mod.rs
similarity index 92%
rename from parquet/src/file/metadata.rs
rename to parquet/src/file/metadata/mod.rs
index fb8f798fd..40922d52b 100644
--- a/parquet/src/file/metadata.rs
+++ b/parquet/src/file/metadata/mod.rs
@@ -29,6 +29,8 @@
 //! * [`ColumnChunkMetaData`]: Metadata for each column chunk (primitive leaf)
 //!   within a Row Group including encoding and compression information,
 //!   number of values, statistics, etc.
+mod memory;
+
 use std::ops::Range;
 use std::sync::Arc;
 
@@ -39,6 +41,7 @@ use crate::format::{
 
 use crate::basic::{ColumnOrder, Compression, Encoding, Type};
 use crate::errors::{ParquetError, Result};
+pub(crate) use crate::file::metadata::memory::HeapSize;
 use crate::file::page_encoding_stats::{self, PageEncodingStats};
 use crate::file::page_index::index::Index;
 use crate::file::statistics::{self, Statistics};
@@ -176,6 +179,28 @@ impl ParquetMetaData {
         self.offset_index.as_ref()
     }
 
+    /// Estimate of the bytes allocated to store `ParquetMetadata`
+    ///
+    /// # Notes:
+    ///
+    /// 1. Includes size of self
+    ///
+    /// 2. Includes heap memory for sub fields such as [`FileMetaData`] and
+    /// [`RowGroupMetaData`].
+    ///
+    /// 3. Includes memory from shared pointers (e.g. [`SchemaDescPtr`]). This
+    /// means `memory_size` will over estimate the memory size if such pointers
+    /// are shared.
+    ///
+    /// 4. Does not include any allocator overheads
+    pub fn memory_size(&self) -> usize {
+        std::mem::size_of::<Self>()
+            + self.file_metadata.heap_size()
+            + self.row_groups.heap_size()
+            + self.column_index.heap_size()
+            + self.offset_index.heap_size()
+    }
+
     /// Override the column index
     #[cfg(feature = "arrow")]
     pub(crate) fn set_column_index(&mut self, index: Option<ParquetColumnIndex>) {
@@ -1034,7 +1059,8 @@ impl OffsetIndexBuilder {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::basic::PageType;
+    use crate::basic::{PageType, SortOrder};
+    use crate::file::page_index::index::NativeIndex;
 
     #[test]
     fn test_row_group_metadata_thrift_conversion() {
@@ -1227,6 +1253,68 @@ mod tests {
         assert_eq!(compressed_size_res, compressed_size_exp);
     }
 
+    #[test]
+    fn test_memory_size() {
+        let schema_descr = get_test_schema_descr();
+
+        let columns = schema_descr
+            .columns()
+            .iter()
+            .map(|column_descr| ColumnChunkMetaData::builder(column_descr.clone()).build())
+            .collect::<Result<Vec<_>>>()
+            .unwrap();
+        let row_group_meta = RowGroupMetaData::builder(schema_descr.clone())
+            .set_num_rows(1000)
+            .set_column_metadata(columns)
+            .build()
+            .unwrap();
+        let row_group_meta = vec![row_group_meta];
+
+        let version = 2;
+        let num_rows = 1000;
+        let created_by = Some(String::from("test harness"));
+        let key_value_metadata = Some(vec![KeyValue::new(
+            String::from("Foo"),
+            Some(String::from("bar")),
+        )]);
+        let column_orders = Some(vec![
+            ColumnOrder::UNDEFINED,
+            ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::UNSIGNED),
+        ]);
+        let file_metadata = FileMetaData::new(
+            version,
+            num_rows,
+            created_by,
+            key_value_metadata,
+            schema_descr,
+            column_orders,
+        );
+        let parquet_meta = ParquetMetaData::new(file_metadata.clone(), row_group_meta.clone());
+        let base_expected_size = 1320;
+        assert_eq!(parquet_meta.memory_size(), base_expected_size);
+
+        let mut column_index = ColumnIndexBuilder::new();
+        column_index.append(false, vec![1u8], vec![2u8, 3u8], 4);
+        let column_index = column_index.build_to_thrift();
+        let native_index = NativeIndex::<bool>::try_new(column_index).unwrap();
+
+        // Now, add in OffsetIndex
+        let parquet_meta = ParquetMetaData::new_with_page_index(
+            file_metadata,
+            row_group_meta,
+            Some(vec![vec![Index::BOOLEAN(native_index)]]),
+            Some(vec![vec![
+                vec![PageLocation::new(1, 2, 3)],
+                vec![PageLocation::new(1, 2, 3)],
+            ]]),
+        );
+
+        let bigger_expected_size = 2304;
+        // more set fields means more memory usage
+        assert!(bigger_expected_size > base_expected_size);
+        assert_eq!(parquet_meta.memory_size(), bigger_expected_size);
+    }
+
     /// Returns sample schema descriptor so we can create column metadata.
     fn get_test_schema_descr() -> SchemaDescPtr {
         let schema = SchemaType::group_type_builder("schema")
diff --git a/parquet/src/schema/types.rs b/parquet/src/schema/types.rs
index 13cc016df..0eb08f557 100644
--- a/parquet/src/schema/types.rs
+++ b/parquet/src/schema/types.rs
@@ -19,6 +19,7 @@
 
 use std::{collections::HashMap, fmt, sync::Arc};
 
+use crate::file::metadata::HeapSize;
 use crate::format::SchemaElement;
 
 use crate::basic::{
@@ -57,6 +58,15 @@ pub enum Type {
     },
 }
 
+impl HeapSize for Type {
+    fn heap_size(&self) -> usize {
+        match self {
+            Type::PrimitiveType { basic_info, .. } => basic_info.heap_size(),
+            Type::GroupType { basic_info, fields } => basic_info.heap_size() + fields.heap_size(),
+        }
+    }
+}
+
 impl Type {
     /// Creates primitive type builder with provided field name and physical type.
     pub fn primitive_type_builder(name: &str, physical_type: PhysicalType) -> PrimitiveTypeBuilder {
@@ -625,6 +635,13 @@ pub struct BasicTypeInfo {
     id: Option<i32>,
 }
 
+impl HeapSize for BasicTypeInfo {
+    fn heap_size(&self) -> usize {
+        // no heap allocations in any other subfield
+        self.name.heap_size()
+    }
+}
+
 impl BasicTypeInfo {
     /// Returns field name.
     pub fn name(&self) -> &str {
@@ -693,6 +710,12 @@ pub struct ColumnPath {
     parts: Vec<String>,
 }
 
+impl HeapSize for ColumnPath {
+    fn heap_size(&self) -> usize {
+        self.parts.heap_size()
+    }
+}
+
 impl ColumnPath {
     /// Creates new column path from vector of field names.
     pub fn new(parts: Vec<String>) -> Self {
@@ -781,6 +804,12 @@ pub struct ColumnDescriptor {
     path: ColumnPath,
 }
 
+impl HeapSize for ColumnDescriptor {
+    fn heap_size(&self) -> usize {
+        self.primitive_type.heap_size() + self.path.heap_size()
+    }
+}
+
 impl ColumnDescriptor {
     /// Creates new descriptor for leaf-level column.
     pub fn new(
@@ -925,6 +954,13 @@ impl fmt::Debug for SchemaDescriptor {
     }
 }
 
+// Need to implement HeapSize in this module as the fields are private
+impl HeapSize for SchemaDescriptor {
+    fn heap_size(&self) -> usize {
+        self.schema.heap_size() + self.leaves.heap_size() + self.leaf_to_base.heap_size()
+    }
+}
+
 impl SchemaDescriptor {
     /// Creates new schema descriptor from Parquet schema.
     pub fn new(tp: TypePtr) -> Self {
diff --git a/parquet_derive/src/parquet_field.rs b/parquet_derive/src/parquet_field.rs
index 9fff76c42..f99ea3e03 100644
--- a/parquet_derive/src/parquet_field.rs
+++ b/parquet_derive/src/parquet_field.rs
@@ -239,7 +239,8 @@ impl Field {
     /// because this parsing logic is not sophisticated enough for definition
     /// levels beyond 2.
     ///
-    /// `Option` types and references not supported
+    /// `Option` types and references not supported, but the column itself can be nullable
+    /// (i.e., def_level==1), as long as the values are all valid.
     pub fn reader_snippet(&self) -> proc_macro2::TokenStream {
         let ident = &self.ident;
         let column_reader = self.ty.column_reader();
@@ -248,7 +249,13 @@ impl Field {
         let write_batch_expr = quote! {
             let mut vals = Vec::new();
             if let #column_reader(mut typed) = column_reader {
-                typed.read_records(num_records, None, None, &mut vals)?;
+                let mut definition_levels = Vec::new();
+                let (total_num, valid_num, decoded_num) = typed.read_records(
+                    num_records, Some(&mut definition_levels), None, &mut vals)?;
+                if valid_num != decoded_num {
+                    panic!("Support only valid records, found {} null records in column type {}",
+                        decoded_num - valid_num, stringify!{#ident});
+                }
             } else {
                 panic!("Schema and struct disagree on type for {}", stringify!{#ident});
             }
@@ -876,15 +883,21 @@ mod test {
             snippet,
             (quote! {
                  {
-                     let mut vals = Vec::new();
-                     if let ColumnReader::Int64ColumnReader(mut typed) = column_reader {
-                         typed.read_records(num_records, None, None, &mut vals)?;
-                     } else {
-                         panic!("Schema and struct disagree on type for {}", stringify!{ counter });
-                     }
-                     for (i, r) in &mut records[..num_records].iter_mut().enumerate() {
-                         r.counter = vals[i] as usize;
-                     }
+                    let mut vals = Vec::new();
+                    if let ColumnReader::Int64ColumnReader(mut typed) = column_reader {
+                        let mut definition_levels = Vec::new();
+                        let (total_num, valid_num, decoded_num) = typed.read_records(
+                            num_records, Some(&mut definition_levels), None, &mut vals)?;
+                        if valid_num != decoded_num {
+                            panic!("Support only valid records, found {} null records in column type {}",
+                                decoded_num - valid_num, stringify!{counter});
+                        }
+                    } else {
+                        panic!("Schema and struct disagree on type for {}", stringify!{counter});
+                    }
+                    for (i, r) in &mut records[..num_records].iter_mut().enumerate() {
+                        r.counter = vals[i] as usize;
+                    }
                  }
             })
             .to_string()
@@ -1291,7 +1304,13 @@ mod test {
             {
                 let mut vals = Vec::new();
                 if let ColumnReader::Int64ColumnReader(mut typed) = column_reader {
-                    typed.read_records(num_records, None, None, &mut vals)?;
+                    let mut definition_levels = Vec::new();
+                    let (total_num, valid_num, decoded_num) = typed.read_records(
+                        num_records, Some(&mut definition_levels), None, &mut vals)?;
+                    if valid_num != decoded_num {
+                        panic!("Support only valid records, found {} null records in column type {}",
+                            decoded_num - valid_num, stringify!{henceforth});
+                    }
                 } else {
                     panic!("Schema and struct disagree on type for {}", stringify!{ henceforth });
                 }
@@ -1359,7 +1378,13 @@ mod test {
             {
                 let mut vals = Vec::new();
                 if let ColumnReader::Int32ColumnReader(mut typed) = column_reader {
-                    typed.read_records(num_records, None, None, &mut vals)?;
+                    let mut definition_levels = Vec::new();
+                    let (total_num, valid_num, decoded_num) = typed.read_records(
+                        num_records, Some(&mut definition_levels), None, &mut vals)?;
+                    if valid_num != decoded_num {
+                        panic!("Support only valid records, found {} null records in column type {}",
+                            decoded_num - valid_num, stringify!{henceforth});
+                    }
                 } else {
                     panic!("Schema and struct disagree on type for {}", stringify!{ henceforth });
                 }
@@ -1427,7 +1452,13 @@ mod test {
             {
                 let mut vals = Vec::new();
                 if let ColumnReader::FixedLenByteArrayColumnReader(mut typed) = column_reader {
-                    typed.read_records(num_records, None, None, &mut vals)?;
+                    let mut definition_levels = Vec::new();
+                    let (total_num, valid_num, decoded_num) = typed.read_records(
+                        num_records, Some(&mut definition_levels), None, &mut vals)?;
+                    if valid_num != decoded_num {
+                        panic!("Support only valid records, found {} null records in column type {}",
+                            decoded_num - valid_num, stringify!{unique_id});
+                    }
                 } else {
                     panic!("Schema and struct disagree on type for {}", stringify!{ unique_id });
                 }


Test patch:
diff --git a/arrow-integration-testing/src/flight_client_scenarios/integration_test.rs b/arrow-integration-testing/src/flight_client_scenarios/integration_test.rs
index c6b5a72ca..ec88ce36a 100644
--- a/arrow-integration-testing/src/flight_client_scenarios/integration_test.rs
+++ b/arrow-integration-testing/src/flight_client_scenarios/integration_test.rs
@@ -123,7 +123,7 @@ async fn send_batch(
     options: &writer::IpcWriteOptions,
 ) -> Result {
     let data_gen = writer::IpcDataGenerator::default();
-    let mut dictionary_tracker = writer::DictionaryTracker::new(false);
+    let mut dictionary_tracker = writer::DictionaryTracker::new_with_preserve_dict_id(false, true);
 
     let (encoded_dictionaries, encoded_batch) = data_gen
         .encoded_batch(batch, &mut dictionary_tracker, options)
diff --git a/arrow-integration-testing/src/flight_server_scenarios/integration_test.rs b/arrow-integration-testing/src/flight_server_scenarios/integration_test.rs
index 25203ecb7..a03c1cd1a 100644
--- a/arrow-integration-testing/src/flight_server_scenarios/integration_test.rs
+++ b/arrow-integration-testing/src/flight_server_scenarios/integration_test.rs
@@ -119,7 +119,8 @@ impl FlightService for FlightServiceImpl {
             .enumerate()
             .flat_map(|(counter, batch)| {
                 let data_gen = writer::IpcDataGenerator::default();
-                let mut dictionary_tracker = writer::DictionaryTracker::new(false);
+                let mut dictionary_tracker =
+                    writer::DictionaryTracker::new_with_preserve_dict_id(false, true);
 
                 let (encoded_dictionaries, encoded_batch) = data_gen
                     .encoded_batch(batch, &mut dictionary_tracker, &options)
diff --git a/parquet_derive_test/src/lib.rs b/parquet_derive_test/src/lib.rs
index b8e6c27f5..e7c7896cb 100644
--- a/parquet_derive_test/src/lib.rs
+++ b/parquet_derive_test/src/lib.rs
@@ -66,6 +66,25 @@ struct APartiallyCompleteRecord {
     pub byte_vec: Vec<u8>,
 }
 
+// This struct has OPTIONAL columns
+// If these fields are guaranteed to be valid
+// we can load this struct into APartiallyCompleteRecord
+#[derive(PartialEq, ParquetRecordWriter, Debug)]
+struct APartiallyOptionalRecord {
+    pub bool: bool,
+    pub string: String,
+    pub maybe_i16: Option<i16>,
+    pub maybe_i32: Option<i32>,
+    pub maybe_u64: Option<u64>,
+    pub isize: isize,
+    pub float: f32,
+    pub double: f64,
+    pub now: chrono::NaiveDateTime,
+    pub date: chrono::NaiveDate,
+    pub uuid: uuid::Uuid,
+    pub byte_vec: Vec<u8>,
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -218,6 +237,47 @@ mod tests {
         assert_eq!(drs[0], out[0]);
     }
 
+    #[test]
+    fn test_parquet_derive_read_optional_but_valid_column() {
+        let file = get_temp_file("test_parquet_derive_read_optional", &[]);
+        let drs: Vec<APartiallyOptionalRecord> = vec![APartiallyOptionalRecord {
+            bool: true,
+            string: "a string".into(),
+            maybe_i16: Some(-45),
+            maybe_i32: Some(456),
+            maybe_u64: Some(4563424),
+            isize: -365,
+            float: 3.5,
+            double: f64::NAN,
+            now: chrono::Utc::now().naive_local(),
+            date: chrono::naive::NaiveDate::from_ymd_opt(2015, 3, 14).unwrap(),
+            uuid: uuid::Uuid::new_v4(),
+            byte_vec: vec![0x65, 0x66, 0x67],
+        }];
+
+        let generated_schema = drs.as_slice().schema().unwrap();
+
+        let props = Default::default();
+        let mut writer =
+            SerializedFileWriter::new(file.try_clone().unwrap(), generated_schema, props).unwrap();
+
+        let mut row_group = writer.next_row_group().unwrap();
+        drs.as_slice().write_to_row_group(&mut row_group).unwrap();
+        row_group.close().unwrap();
+        writer.close().unwrap();
+
+        use parquet::file::{reader::FileReader, serialized_reader::SerializedFileReader};
+        let reader = SerializedFileReader::new(file).unwrap();
+        let mut out: Vec<APartiallyCompleteRecord> = Vec::new();
+
+        let mut row_group = reader.get_row_group(0).unwrap();
+        out.read_from_row_group(&mut *row_group, 1).unwrap();
+
+        assert_eq!(drs[0].maybe_i16.unwrap(), out[0].i16);
+        assert_eq!(drs[0].maybe_i32.unwrap(), out[0].i32);
+        assert_eq!(drs[0].maybe_u64.unwrap(), out[0].u64);
+    }
+
     /// Returns file handle for a temp file in 'target' directory with a provided content
     pub fn get_temp_file(file_name: &str, content: &[u8]) -> fs::File {
         // build tmp path to a file in "target/debug/testdata"