Repository: apache/arrow-rs
Pull request #5912 (2024-06-18 12:50:39)

Make ObjectStoreScheme public
# Which issue does this PR close?

Closes https://github.com/apache/arrow-rs/issues/5911.

# Rationale for this change
 
`ObjectStoreScheme::parse` contains non-trivial logic to resolve a given url into an object store scheme. Exposing this is useful, as it allows consumers to set up their own builders as they would like.

# What changes are included in this PR?

This just makes the types public.

# Are there any user-facing changes?

The types become public.


alamb: It appears this PR also has a clippy CI failure that we need to fix before merging
orf: Hey, I will fix these issues up next week. Thanks for the review 😌
drmorr0: Would it be possible to add `#[derive(Clone)]` on this as well now that it's public?
alamb: I took the liberty of adding additional docs / examples to this PR and merging up from main 



Patch:
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-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/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/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/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/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"),
+            ]
+        );
+    }
+}


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)