Repository: apache/arrow-rs
Pull request #6540 (2024-10-10 10:26:53)

Improve UnionArray::is_nullable
# Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->

Relates to #6538.

# Rationale for this change
 
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->

Always returning true is overly pessimistic, and whilst this is still not accurate, it is at least slightly more accurate

# What changes are included in this PR?

<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->

# Are there any user-facing changes?


<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->

<!---
If there are any breaking changes to public APIs, please add the `breaking change` label.
-->



Patch:
diff --git a/arrow-array/src/array/dictionary_array.rs b/arrow-array/src/array/dictionary_array.rs
index d6c5dd4c3..bdb6f0d4b 100644
--- a/arrow-array/src/array/dictionary_array.rs
+++ b/arrow-array/src/array/dictionary_array.rs
@@ -1072,6 +1072,74 @@ mod tests {
         assert_eq!(dict_array.keys(), &Int16Array::from(vec![3_i16, 4]));
     }
 
+    #[test]
+    fn test_dictionary_builder_append_many() {
+        let mut builder = PrimitiveDictionaryBuilder::<UInt8Type, UInt32Type>::new();
+
+        builder.append(1).unwrap();
+        builder.append_n(2, 2).unwrap();
+        builder.append_options(None, 2);
+        builder.append_options(Some(3), 3);
+
+        let array = builder.finish();
+
+        let values = array
+            .values()
+            .as_primitive::<UInt32Type>()
+            .iter()
+            .map(Option::unwrap)
+            .collect::<Vec<_>>();
+        assert_eq!(values, &[1, 2, 3]);
+        let keys = array.keys().iter().collect::<Vec<_>>();
+        assert_eq!(
+            keys,
+            &[
+                Some(0),
+                Some(1),
+                Some(1),
+                None,
+                None,
+                Some(2),
+                Some(2),
+                Some(2)
+            ]
+        );
+    }
+
+    #[test]
+    fn test_string_dictionary_builder_append_many() {
+        let mut builder = StringDictionaryBuilder::<Int8Type>::new();
+
+        builder.append("a").unwrap();
+        builder.append_n("b", 2).unwrap();
+        builder.append_options(None::<&str>, 2);
+        builder.append_options(Some("c"), 3);
+
+        let array = builder.finish();
+
+        let values = array
+            .values()
+            .as_string::<i32>()
+            .iter()
+            .map(Option::unwrap)
+            .collect::<Vec<_>>();
+        assert_eq!(values, &["a", "b", "c"]);
+        let keys = array.keys().iter().collect::<Vec<_>>();
+        assert_eq!(
+            keys,
+            &[
+                Some(0),
+                Some(1),
+                Some(1),
+                None,
+                None,
+                Some(2),
+                Some(2),
+                Some(2)
+            ]
+        );
+    }
+
     #[test]
     fn test_dictionary_array_fmt_debug() {
         let mut builder = PrimitiveDictionaryBuilder::<UInt8Type, UInt32Type>::with_capacity(3, 2);
diff --git a/arrow-array/src/array/list_array.rs b/arrow-array/src/array/list_array.rs
index dc1b9f07d..13b446b03 100644
--- a/arrow-array/src/array/list_array.rs
+++ b/arrow-array/src/array/list_array.rs
@@ -549,7 +549,7 @@ pub type LargeListArray = GenericListArray<i64>;
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::builder::{FixedSizeListBuilder, Int32Builder, ListBuilder};
+    use crate::builder::{FixedSizeListBuilder, Int32Builder, ListBuilder, UnionBuilder};
     use crate::cast::AsArray;
     use crate::types::Int32Type;
     use crate::{Int32Array, Int64Array};
@@ -1181,4 +1181,18 @@ mod tests {
             .collect();
         assert_eq!(values, vec![Some(vec![1, 2, 3]), None, Some(vec![4, 5, 6])])
     }
+
+    #[test]
+    fn test_nullable_union() {
+        let offsets = OffsetBuffer::new(vec![0, 1, 4, 5].into());
+        let mut builder = UnionBuilder::new_dense();
+        builder.append::<Int32Type>("a", 1).unwrap();
+        builder.append::<Int32Type>("b", 2).unwrap();
+        builder.append::<Int32Type>("b", 3).unwrap();
+        builder.append::<Int32Type>("a", 4).unwrap();
+        builder.append::<Int32Type>("a", 5).unwrap();
+        let values = builder.build().unwrap();
+        let field = Arc::new(Field::new("element", values.data_type().clone(), false));
+        ListArray::new(field.clone(), offsets, Arc::new(values), None);
+    }
 }
diff --git a/arrow-array/src/array/union_array.rs b/arrow-array/src/array/union_array.rs
index 1feef8c56..3c6da5a7b 100644
--- a/arrow-array/src/array/union_array.rs
+++ b/arrow-array/src/array/union_array.rs
@@ -875,7 +875,10 @@ impl Array for UnionArray {
     }
 
     fn is_nullable(&self) -> bool {
-        true
+        self.fields
+            .iter()
+            .flatten()
+            .any(|field| field.is_nullable())
     }
 
     fn get_buffer_memory_size(&self) -> usize {
@@ -2138,4 +2141,43 @@ mod tests {
         .into_iter()
         .collect()
     }
+
+    #[test]
+    fn test_is_nullable() {
+        assert!(!create_union_array(false, false).is_nullable());
+        assert!(create_union_array(true, false).is_nullable());
+        assert!(create_union_array(false, true).is_nullable());
+        assert!(create_union_array(true, true).is_nullable());
+    }
+
+    /// Create a union array with a float and integer field
+    ///
+    /// If the `int_nullable` is true, the integer field will have nulls
+    /// If the `float_nullable` is true, the float field will have nulls
+    ///
+    /// Note the `Field` definitions are always declared to be nullable
+    fn create_union_array(int_nullable: bool, float_nullable: bool) -> UnionArray {
+        let int_array = if int_nullable {
+            Int32Array::from(vec![Some(1), None, Some(3)])
+        } else {
+            Int32Array::from(vec![1, 2, 3])
+        };
+        let float_array = if float_nullable {
+            Float64Array::from(vec![Some(3.2), None, Some(4.2)])
+        } else {
+            Float64Array::from(vec![3.2, 4.2, 5.2])
+        };
+        let type_ids = [0, 1, 0].into_iter().collect::<ScalarBuffer<i8>>();
+        let offsets = [0, 0, 0].into_iter().collect::<ScalarBuffer<i32>>();
+        let union_fields = [
+            (0, Arc::new(Field::new("A", DataType::Int32, true))),
+            (1, Arc::new(Field::new("B", DataType::Float64, true))),
+        ]
+        .into_iter()
+        .collect::<UnionFields>();
+
+        let children = vec![Arc::new(int_array) as Arc<dyn Array>, Arc::new(float_array)];
+
+        UnionArray::try_new(union_fields, type_ids, Some(offsets), children).unwrap()
+    }
 }
diff --git a/arrow-array/src/builder/generic_bytes_dictionary_builder.rs b/arrow-array/src/builder/generic_bytes_dictionary_builder.rs
index 128a4f820..bb0fb5e91 100644
--- a/arrow-array/src/builder/generic_bytes_dictionary_builder.rs
+++ b/arrow-array/src/builder/generic_bytes_dictionary_builder.rs
@@ -195,12 +195,7 @@ where
     K: ArrowDictionaryKeyType,
     T: ByteArrayType,
 {
-    /// Append a value to the array. Return an existing index
-    /// if already present in the values array or a new index if the
-    /// value is appended to the values array.
-    ///
-    /// Returns an error if the new index would overflow the key type.
-    pub fn append(&mut self, value: impl AsRef<T::Native>) -> Result<K::Native, ArrowError> {
+    fn get_or_insert_key(&mut self, value: impl AsRef<T::Native>) -> Result<K::Native, ArrowError> {
         let value_native: &T::Native = value.as_ref();
         let value_bytes: &[u8] = value_native.as_ref();
 
@@ -223,8 +218,32 @@ where
             .get();
 
         let key = K::Native::from_usize(idx).ok_or(ArrowError::DictionaryKeyOverflowError)?;
+
+        Ok(key)
+    }
+
+    /// Append a value to the array. Return an existing index
+    /// if already present in the values array or a new index if the
+    /// value is appended to the values array.
+    ///
+    /// Returns an error if the new index would overflow the key type.
+    pub fn append(&mut self, value: impl AsRef<T::Native>) -> Result<K::Native, ArrowError> {
+        let key = self.get_or_insert_key(value)?;
         self.keys_builder.append_value(key);
+        Ok(key)
+    }
 
+    /// Append a value multiple times to the array.
+    /// This is the same as `append` but allows to append the same value multiple times without doing multiple lookups.
+    ///
+    /// Returns an error if the new index would overflow the key type.
+    pub fn append_n(
+        &mut self,
+        value: impl AsRef<T::Native>,
+        count: usize,
+    ) -> Result<K::Native, ArrowError> {
+        let key = self.get_or_insert_key(value)?;
+        self.keys_builder.append_value_n(key, count);
         Ok(key)
     }
 
@@ -237,12 +256,29 @@ where
         self.append(value).expect("dictionary key overflow");
     }
 
+    /// Infallibly append a value to this builder repeatedly `count` times.
+    /// This is the same as `append_value` but allows to append the same value multiple times without doing multiple lookups.
+    ///
+    /// # Panics
+    ///
+    /// Panics if the resulting length of the dictionary values array would exceed `T::Native::MAX`
+    pub fn append_values(&mut self, value: impl AsRef<T::Native>, count: usize) {
+        self.append_n(value, count)
+            .expect("dictionary key overflow");
+    }
+
     /// Appends a null slot into the builder
     #[inline]
     pub fn append_null(&mut self) {
         self.keys_builder.append_null()
     }
 
+    /// Infallibly append `n` null slots into the builder
+    #[inline]
+    pub fn append_nulls(&mut self, n: usize) {
+        self.keys_builder.append_nulls(n)
+    }
+
     /// Append an `Option` value into the builder
     ///
     /// # Panics
@@ -256,6 +292,19 @@ where
         };
     }
 
+    /// Append an `Option` value into the builder repeatedly `count` times.
+    /// This is the same as `append_option` but allows to append the same value multiple times without doing multiple lookups.
+    ///
+    /// # Panics
+    ///
+    /// Panics if the resulting length of the dictionary values array would exceed `T::Native::MAX`
+    pub fn append_options(&mut self, value: Option<impl AsRef<T::Native>>, count: usize) {
+        match value {
+            None => self.keys_builder.append_nulls(count),
+            Some(v) => self.append_values(v, count),
+        };
+    }
+
     /// Builds the `DictionaryArray` and reset this builder.
     pub fn finish(&mut self) -> DictionaryArray<K> {
         self.dedup.clear();
@@ -331,8 +380,7 @@ fn get_bytes<T: ByteArrayType>(values: &GenericByteBuilder<T>, idx: usize) -> &[
 /// // The builder builds the dictionary value by value
 /// builder.append("abc").unwrap();
 /// builder.append_null();
-/// builder.append("def").unwrap();
-/// builder.append("def").unwrap();
+/// builder.append_n("def", 2).unwrap();  // appends "def" twice with a single lookup
 /// builder.append("abc").unwrap();
 /// let array = builder.finish();
 ///
diff --git a/arrow-array/src/builder/primitive_builder.rs b/arrow-array/src/builder/primitive_builder.rs
index 39b27bfca..3191fea6e 100644
--- a/arrow-array/src/builder/primitive_builder.rs
+++ b/arrow-array/src/builder/primitive_builder.rs
@@ -202,6 +202,13 @@ impl<T: ArrowPrimitiveType> PrimitiveBuilder<T> {
         self.values_builder.append(v);
     }
 
+    /// Appends a value of type `T` into the builder `n` times
+    #[inline]
+    pub fn append_value_n(&mut self, v: T::Native, n: usize) {
+        self.null_buffer_builder.append_n_non_nulls(n);
+        self.values_builder.append_n(n, v);
+    }
+
     /// Appends a null slot into the builder
     #[inline]
     pub fn append_null(&mut self) {
diff --git a/arrow-array/src/builder/primitive_dictionary_builder.rs b/arrow-array/src/builder/primitive_dictionary_builder.rs
index a764fa4c2..ac40f8a46 100644
--- a/arrow-array/src/builder/primitive_dictionary_builder.rs
+++ b/arrow-array/src/builder/primitive_dictionary_builder.rs
@@ -21,7 +21,6 @@ use crate::{Array, ArrayRef, ArrowPrimitiveType, DictionaryArray};
 use arrow_buffer::{ArrowNativeType, ToByteSlice};
 use arrow_schema::{ArrowError, DataType};
 use std::any::Any;
-use std::collections::hash_map::Entry;
 use std::collections::HashMap;
 use std::sync::Arc;
 
@@ -210,26 +209,41 @@ where
     K: ArrowDictionaryKeyType,
     V: ArrowPrimitiveType,
 {
-    /// Append a primitive value to the array. Return an existing index
-    /// if already present in the values array or a new index if the
-    /// value is appended to the values array.
     #[inline]
-    pub fn append(&mut self, value: V::Native) -> Result<K::Native, ArrowError> {
-        let key = match self.map.entry(Value(value)) {
-            Entry::Vacant(vacant) => {
-                // Append new value.
+    fn get_or_insert_key(&mut self, value: V::Native) -> Result<K::Native, ArrowError> {
+        match self.map.get(&Value(value)) {
+            Some(&key) => {
+                Ok(K::Native::from_usize(key).ok_or(ArrowError::DictionaryKeyOverflowError)?)
+            }
+            None => {
                 let key = self.values_builder.len();
                 self.values_builder.append_value(value);
-                vacant.insert(key);
-                K::Native::from_usize(key).ok_or(ArrowError::DictionaryKeyOverflowError)?
+                self.map.insert(Value(value), key);
+                Ok(K::Native::from_usize(key).ok_or(ArrowError::DictionaryKeyOverflowError)?)
             }
-            Entry::Occupied(o) => K::Native::usize_as(*o.get()),
-        };
+        }
+    }
 
+    /// Append a primitive value to the array. Return an existing index
+    /// if already present in the values array or a new index if the
+    /// value is appended to the values array.
+    #[inline]
+    pub fn append(&mut self, value: V::Native) -> Result<K::Native, ArrowError> {
+        let key = self.get_or_insert_key(value)?;
         self.keys_builder.append_value(key);
         Ok(key)
     }
 
+    /// Append a value multiple times to the array.
+    /// This is the same as `append` but allows to append the same value multiple times without doing multiple lookups.
+    ///
+    /// Returns an error if the new index would overflow the key type.
+    pub fn append_n(&mut self, value: V::Native, count: usize) -> Result<K::Native, ArrowError> {
+        let key = self.get_or_insert_key(value)?;
+        self.keys_builder.append_value_n(key, count);
+        Ok(key)
+    }
+
     /// Infallibly append a value to this builder
     ///
     /// # Panics
@@ -240,12 +254,29 @@ where
         self.append(value).expect("dictionary key overflow");
     }
 
+    /// Infallibly append a value to this builder repeatedly `count` times.
+    /// This is the same as `append_value` but allows to append the same value multiple times without doing multiple lookups.
+    ///
+    /// # Panics
+    ///
+    /// Panics if the resulting length of the dictionary values array would exceed `T::Native::MAX`
+    pub fn append_values(&mut self, value: V::Native, count: usize) {
+        self.append_n(value, count)
+            .expect("dictionary key overflow");
+    }
+
     /// Appends a null slot into the builder
     #[inline]
     pub fn append_null(&mut self) {
         self.keys_builder.append_null()
     }
 
+    /// Append `n` null slots into the builder
+    #[inline]
+    pub fn append_nulls(&mut self, n: usize) {
+        self.keys_builder.append_nulls(n)
+    }
+
     /// Append an `Option` value into the builder
     ///
     /// # Panics
@@ -259,6 +290,19 @@ where
         };
     }
 
+    /// Append an `Option` value into the builder repeatedly `count` times.
+    /// This is the same as `append_option` but allows to append the same value multiple times without doing multiple lookups.
+    ///
+    /// # Panics
+    ///
+    /// Panics if the resulting length of the dictionary values array would exceed `T::Native::MAX`
+    pub fn append_options(&mut self, value: Option<V::Native>, count: usize) {
+        match value {
+            None => self.keys_builder.append_nulls(count),
+            Some(v) => self.append_values(v, count),
+        };
+    }
+
     /// Builds the `DictionaryArray` and reset this builder.
     pub fn finish(&mut self) -> DictionaryArray<K> {
         self.map.clear();
diff --git a/arrow-array/src/lib.rs b/arrow-array/src/lib.rs
index 90bc5e312..0fc9d30ab 100644
--- a/arrow-array/src/lib.rs
+++ b/arrow-array/src/lib.rs
@@ -161,7 +161,52 @@
 //!     array.as_primitive::<Float32Type>().values()
 //! }
 //! ```
+//! # Alternatives to ChunkedArray Support
 //!
+//! The Rust implementation does not provide the ChunkedArray abstraction implemented by the Python
+//! and C++ Arrow implementations. The recommended alternative is to use one of the following:
+//! - `Vec<ArrayRef>` a simple, eager version of a `ChunkedArray`
+//! - `impl Iterator<Item=ArrayRef>` a lazy version of a `ChunkedArray`
+//! - `impl Stream<Item=ArrayRef>` a lazy async version of a `ChunkedArray`
+//!
+//! Similar patterns can be applied at the `RecordBatch` level. For example, [DataFusion] makes
+//! extensive use of [RecordBatchStream].
+//!
+//! This approach integrates well into the Rust ecosystem, simplifies the implementation and
+//! encourages the use of performant lazy and async patterns.
+//! ```rust
+//! use std::sync::Arc;
+//! use arrow_array::{ArrayRef, Float32Array, RecordBatch, StringArray};
+//! use arrow_array::cast::AsArray;
+//! use arrow_array::types::Float32Type;
+//! use arrow_schema::DataType;
+//!
+//! let batches = [
+//!    RecordBatch::try_from_iter(vec![
+//!         ("label", Arc::new(StringArray::from(vec!["A", "B", "C"])) as ArrayRef),
+//!         ("value", Arc::new(Float32Array::from(vec![0.1, 0.2, 0.3])) as ArrayRef),
+//!     ]).unwrap(),
+//!    RecordBatch::try_from_iter(vec![
+//!         ("label", Arc::new(StringArray::from(vec!["D", "E"])) as ArrayRef),
+//!         ("value", Arc::new(Float32Array::from(vec![0.4, 0.5])) as ArrayRef),
+//!    ]).unwrap(),
+//! ];
+//!
+//! let labels: Vec<&str> = batches
+//!    .iter()
+//!    .flat_map(|batch| batch.column(0).as_string::<i32>())
+//!    .map(Option::unwrap)
+//!    .collect();
+//!
+//! let values: Vec<f32> = batches
+//!    .iter()
+//!    .flat_map(|batch| batch.column(1).as_primitive::<Float32Type>().values())
+//!    .copied()
+//!    .collect();
+//!
+//! assert_eq!(labels, ["A", "B", "C", "D", "E"]);
+//! assert_eq!(values, [0.1, 0.2, 0.3, 0.4, 0.5]);
+//!```
 //! [`ScalarBuffer<T>`]: arrow_buffer::ScalarBuffer
 //! [`ScalarBuffer<i16>`]: arrow_buffer::ScalarBuffer
 //! [`OffsetBuffer<i32>`]: arrow_buffer::OffsetBuffer
@@ -173,6 +218,8 @@
 //! [`compute`]: https://docs.rs/arrow/latest/arrow/compute/index.html
 //! [`json`]: https://docs.rs/arrow/latest/arrow/json/index.html
 //! [`csv`]: https://docs.rs/arrow/latest/arrow/csv/index.html
+//! [DataFusion]: https://github.com/apache/arrow-datafusion
+//! [RecordBatchStream]: https://docs.rs/datafusion/latest/datafusion/execution/trait.RecordBatchStream.html
 
 #![deny(rustdoc::broken_intra_doc_links)]
 #![warn(missing_docs)]
diff --git a/arrow-cast/src/cast/decimal.rs b/arrow-cast/src/cast/decimal.rs
index 637cbc417..d6b2f884f 100644
--- a/arrow-cast/src/cast/decimal.rs
+++ b/arrow-cast/src/cast/decimal.rs
@@ -250,7 +250,7 @@ where
         }
     };
 
-    let integers = first_part.trim_start_matches('0');
+    let integers = first_part;
     let decimals = if parts.len() == 2 { parts[1] } else { "" };
 
     if !integers.is_empty() && !integers.as_bytes()[0].is_ascii_digit() {
@@ -567,3 +567,48 @@ where
     let array = array.unary::<_, T>(op);
     Ok(Arc::new(array))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_parse_string_to_decimal_native() -> Result<(), ArrowError> {
+        assert_eq!(
+            parse_string_to_decimal_native::<Decimal128Type>("0", 0)?,
+            0_i128
+        );
+        assert_eq!(
+            parse_string_to_decimal_native::<Decimal128Type>("0", 5)?,
+            0_i128
+        );
+
+        assert_eq!(
+            parse_string_to_decimal_native::<Decimal128Type>("123", 0)?,
+            123_i128
+        );
+        assert_eq!(
+            parse_string_to_decimal_native::<Decimal128Type>("123", 5)?,
+            12300000_i128
+        );
+
+        assert_eq!(
+            parse_string_to_decimal_native::<Decimal128Type>("123.45", 0)?,
+            123_i128
+        );
+        assert_eq!(
+            parse_string_to_decimal_native::<Decimal128Type>("123.45", 5)?,
+            12345000_i128
+        );
+
+        assert_eq!(
+            parse_string_to_decimal_native::<Decimal128Type>("123.4567891", 0)?,
+            123_i128
+        );
+        assert_eq!(
+            parse_string_to_decimal_native::<Decimal128Type>("123.4567891", 5)?,
+            12345679_i128
+        );
+        Ok(())
+    }
+}
diff --git a/arrow-cast/src/cast/mod.rs b/arrow-cast/src/cast/mod.rs
index e3fad3da1..7abadf579 100644
--- a/arrow-cast/src/cast/mod.rs
+++ b/arrow-cast/src/cast/mod.rs
@@ -8481,6 +8481,10 @@ mod tests {
         assert!(decimal_arr.is_null(25));
         assert!(decimal_arr.is_null(26));
         assert!(decimal_arr.is_null(27));
+        assert_eq!("0.00", decimal_arr.value_as_string(28));
+        assert_eq!("0.00", decimal_arr.value_as_string(29));
+        assert_eq!("12345.00", decimal_arr.value_as_string(30));
+        assert_eq!(decimal_arr.len(), 31);
 
         // Decimal256
         let output_type = DataType::Decimal256(76, 3);
@@ -8517,6 +8521,10 @@ mod tests {
         assert!(decimal_arr.is_null(25));
         assert!(decimal_arr.is_null(26));
         assert!(decimal_arr.is_null(27));
+        assert_eq!("0.000", decimal_arr.value_as_string(28));
+        assert_eq!("0.000", decimal_arr.value_as_string(29));
+        assert_eq!("12345.000", decimal_arr.value_as_string(30));
+        assert_eq!(decimal_arr.len(), 31);
     }
 
     #[test]
@@ -8550,10 +8558,30 @@ mod tests {
             Some("1.-23499999"),
             Some("-1.-23499999"),
             Some("--1.23499999"),
+            Some("0"),
+            Some("000.000"),
+            Some("0000000000000000012345.000"),
         ]);
         let array = Arc::new(str_array) as ArrayRef;
 
         test_cast_string_to_decimal(array);
+
+        let test_cases = [
+            (None, None),
+            // (Some(""), None),
+            // (Some("   "), None),
+            (Some("0"), Some("0")),
+            (Some("000.000"), Some("0")),
+            (Some("12345"), Some("12345")),
+            (Some("000000000000000000000000000012345"), Some("12345")),
+            (Some("-123"), Some("-123")),
+            (Some("+123"), Some("123")),
+        ];
+        let inputs = test_cases.iter().map(|entry| entry.0).collect::<Vec<_>>();
+        let expected = test_cases.iter().map(|entry| entry.1).collect::<Vec<_>>();
+
+        let array = Arc::new(StringArray::from(inputs)) as ArrayRef;
+        test_cast_string_to_decimal_scale_zero(array, &expected);
     }
 
     #[test]
@@ -8587,10 +8615,67 @@ mod tests {
             Some("1.-23499999"),
             Some("-1.-23499999"),
             Some("--1.23499999"),
+            Some("0"),
+            Some("000.000"),
+            Some("0000000000000000012345.000"),
         ]);
         let array = Arc::new(str_array) as ArrayRef;
 
         test_cast_string_to_decimal(array);
+
+        let test_cases = [
+            (None, None),
+            (Some(""), None),
+            (Some("   "), None),
+            (Some("0"), Some("0")),
+            (Some("000.000"), Some("0")),
+            (Some("12345"), Some("12345")),
+            (Some("000000000000000000000000000012345"), Some("12345")),
+            (Some("-123"), Some("-123")),
+            (Some("+123"), Some("123")),
+        ];
+        let inputs = test_cases.iter().map(|entry| entry.0).collect::<Vec<_>>();
+        let expected = test_cases.iter().map(|entry| entry.1).collect::<Vec<_>>();
+
+        let array = Arc::new(LargeStringArray::from(inputs)) as ArrayRef;
+        test_cast_string_to_decimal_scale_zero(array, &expected);
+    }
+
+    fn test_cast_string_to_decimal_scale_zero(
+        array: ArrayRef,
+        expected_as_string: &[Option<&str>],
+    ) {
+        // Decimal128
+        let output_type = DataType::Decimal128(38, 0);
+        assert!(can_cast_types(array.data_type(), &output_type));
+        let casted_array = cast(&array, &output_type).unwrap();
+        let decimal_arr = casted_array.as_primitive::<Decimal128Type>();
+        assert_decimal_array_contents(decimal_arr, expected_as_string);
+
+        // Decimal256
+        let output_type = DataType::Decimal256(76, 0);
+        assert!(can_cast_types(array.data_type(), &output_type));
+        let casted_array = cast(&array, &output_type).unwrap();
+        let decimal_arr = casted_array.as_primitive::<Decimal256Type>();
+        assert_decimal_array_contents(decimal_arr, expected_as_string);
+    }
+
+    fn assert_decimal_array_contents<T>(
+        array: &PrimitiveArray<T>,
+        expected_as_string: &[Option<&str>],
+    ) where
+        T: DecimalType + ArrowPrimitiveType,
+    {
+        assert_eq!(array.len(), expected_as_string.len());
+        for (i, expected) in expected_as_string.iter().enumerate() {
+            let actual = if array.is_null(i) {
+                None
+            } else {
+                Some(array.value_as_string(i))
+            };
+            let actual = actual.as_ref().map(|s| s.as_ref());
+            assert_eq!(*expected, actual, "Expected at position {}", i);
+        }
     }
 
     #[test]
diff --git a/arrow-flight/README.md b/arrow-flight/README.md
index 2266b81e4..0d4d67b94 100644
--- a/arrow-flight/README.md
+++ b/arrow-flight/README.md
@@ -31,7 +31,7 @@ Add this to your Cargo.toml:
 
 ```toml
 [dependencies]
-arrow-flight = "51.0.0"
+arrow-flight = "53.1.0"
 ```
 
 Apache Arrow Flight is a gRPC based protocol for exchanging Arrow data between processes. See the blog post [Introducing Apache Arrow Flight: A Framework for Fast Data Transport](https://arrow.apache.org/blog/2019/10/13/introducing-arrow-flight/) for more information.
@@ -46,6 +46,8 @@ that demonstrate how to build a Flight server implemented with [tonic](https://d
 - `flight-sql-experimental`: Enables experimental support for
   [Apache Arrow FlightSQL], a protocol for interacting with SQL databases.
 
+- `tls`: Enables `tls` on `tonic`
+
 ## CLI
 
 This crates offers a basic [Apache Arrow FlightSQL] command line interface.
diff --git a/arrow/CONTRIBUTING.md b/arrow/CONTRIBUTING.md
index 0c795d6b9..a9a9426a4 100644
--- a/arrow/CONTRIBUTING.md
+++ b/arrow/CONTRIBUTING.md
@@ -109,6 +109,42 @@ specific JIRA issues and reference them in these code comments. For example:
 //      This is not sound because .... see https://issues.apache.org/jira/browse/ARROW-nnnnn
 ```
 
+### Usage of SIMD / auto vectorization
+
+This crate does not use SIMD intrinsics (e.g. [`std::simd`]) directly, but
+instead relies on the Rust compiler's auto-vectorization capabilities, which are
+built on LLVM.
+
+SIMD intrinsics are difficult to maintain and can be difficult to reason about.
+The auto-vectorizer in LLVM is quite good and often produces kernels that are
+faster than using hand-written SIMD intrinsics. This crate used to contain
+several kernels with hand-written SIMD instructions, which were removed after
+discovering the auto-vectorized code was faster.
+
+[`std::simd`]: https://doc.rust-lang.org/std/simd/index.html
+
+#### Tips for auto vectorization
+
+LLVM is relatively good at vectorizing vertical operations provided:
+
+1. No conditionals within the loop body (e.g no checking for nulls on each row)
+2. Not too much inlining (judicious use of `#[inline]` and `#[inline(never)]`) as the vectorizer gives up if the code is too complex
+3. No [horizontal reductions] or data dependencies
+4. Suitable SIMD instructions available in the target ISA (e.g. `target-cpu` `RUSTFLAGS` flag)
+
+[horizontal reductions]: https://rust-lang.github.io/packed_simd/perf-guide/vert-hor-ops.html
+
+The last point is especially important as the default `target-cpu` doesn't
+support many SIMD instructions. See the Performance Tips section at the
+end of <https://crates.io/crates/arrow>
+
+To ensure your code is fully vectorized, we recommend using tools like
+<https://rust.godbolt.org/> (again being sure `RUSTFLAGS` is set appropriately)
+to analyze the resulting code, and only once you've exhausted auto vectorization
+think of reaching for manual SIMD. Generally the hard part of vectorizing code
+is structuring the algorithm in such a way that it can be vectorized, regardless
+of what generates those instructions.
+
 # Releases and publishing to crates.io
 
 Please see the [release](../dev/release/README.md) for details on how to create arrow releases
diff --git a/arrow/examples/README.md b/arrow/examples/README.md
index 5c57ec00c..87aa6ee04 100644
--- a/arrow/examples/README.md
+++ b/arrow/examples/README.md
@@ -21,7 +21,7 @@
 
 - [`builders.rs`](builders.rs): Using the Builder API
 - [`collect.rs`](collect.rs): Using the `FromIter` API
-- [`dynamic_types.rs`](dynamic_types.rs):
+- [`dynamic_types.rs`](dynamic_types.rs): Dealing with mixed types dynamically at runtime
 - [`read_csv.rs`](read_csv.rs): Reading CSV files with explicit schema, pretty printing Arrays
 - [`read_csv_infer_schema.rs`](read_csv_infer_schema.rs): Reading CSV files, pretty printing Arrays
 - [`tensor_builder.rs`](tensor_builder.rs): Using tensor builder
diff --git a/dev/release/README.md b/dev/release/README.md
index 177f33bcb..d2d9e48bb 100644
--- a/dev/release/README.md
+++ b/dev/release/README.md
@@ -21,29 +21,24 @@
 
 ## Overview
 
-This file documents the release process for:
-
-1. The "Rust Arrow Crates": `arrow`, `arrow-flight`, `parquet`, and `parquet-derive`.
-2. The `object_store` crate.
+This file documents the release process for the "Rust Arrow Crates": `arrow`, `arrow-flight`, `parquet`, and `parquet-derive`.
 
 ### The Rust Arrow Crates
 
-The Rust Arrow Crates are interconnected (e.g. `parquet` has an optional dependency on `arrow`) so we increment and release all of them together. We try to release a new version of "Rust Arrow Crates" every two weeks. This cadence balances getting new features into the community without overwhelming downstream projects with too frequent changes or overly burdening maintainers.
-
-If any code has been merged to master that has a breaking API change, as defined in [Rust RFC 1105](https://github.com/rust-lang/rfcs/blob/master/text/1105-api-evolution.md), the major version number incremented changed (e.g. `9.0.2` to `9.0.2`). Otherwise the new minor version incremented (e.g. `9.0.2` to `7.1.0`).
-
-### `object_store` crate
+The Rust Arrow Crates are interconnected (e.g. `parquet` has an optional dependency on `arrow`) so we increment and release all of them together.
 
-At the time of writing, we release a new version of `object_store` on demand rather than on a regular schedule.
+If any code has been merged to master that has a breaking API change, as defined
+in [Rust RFC 1105] he major version number is incremented (e.g. `9.0.2` to `10.0.2`).
+Otherwise the new minor version incremented (e.g. `9.0.2` to `9.1.0`).
 
-As we are still in an early phase, we use the 0.x version scheme. If any code has been merged to master that has a breaking API change, as defined in [Rust RFC 1105](https://github.com/rust-lang/rfcs/blob/master/text/1105-api-evolution.md), the minor version number incremented changed (e.g. `0.3.0` to `0.4.0`). Otherwise the patch version is incremented (e.g. `0.3.0` to `0.3.1`).
+[rust rfc 1105]: https://github.com/rust-lang/rfcs/blob/master/text/1105-api-evolution.md
 
 # Release Mechanics
 
 ## Process Overview
 
 As part of the Apache governance model, official releases consist of
-signed source tarballs approved by the PMC.
+signed source tarballs approved by the Arrow PMC.
 
 We then use the code in the approved source tarball to release to
 crates.io, the Rust ecosystem's package manager.
@@ -60,9 +55,7 @@ labels associated with them.
 
 Now prepare a PR to update `CHANGELOG.md` and versions on `master` to reflect the planned release.
 
-For the Rust Arrow crates, do this in the root of this repository. For example [#2323](https://github.com/apache/arrow-rs/pull/2323)
-
-For `object_store` the same process is done in the `object_store` directory. Examples TBD
+Do this in the root of this repository. For example [#2323](https://github.com/apache/arrow-rs/pull/2323)
 
 ```bash
 git checkout master
@@ -90,7 +83,6 @@ python dev/release/label_issues.py
 # review change log / edit issues and labels if needed, rerun
 git commit -a -m 'Create changelog'
 
-
 # Manually edit ./dev/release/update_change_log.sh to reflect the release version
 # Create the changelog
 CHANGELOG_GITHUB_TOKEN=<TOKEN> ./dev/release/update_change_log.sh
@@ -98,9 +90,6 @@ CHANGELOG_GITHUB_TOKEN=<TOKEN> ./dev/release/update_change_log.sh
 git commit -a -m 'Create changelog'
 
 git push
-
-# File the release PR
-export BRANCH=<RELEASE_BRANCH> && export GITHUB_USERNAME=<USERNAME> && export GITHUB_TOKEN=<TOKEN> && ./file_release_pr.sh
 ```
 
 Note that when reviewing the change log, rather than editing the
@@ -120,9 +109,7 @@ distribution servers.
 
 While the official release artifact is a signed tarball, we also tag the commit it was created for convenience and code archaeology.
 
-For a Rust Arrow Crates release, use a string such as `4.0.1` as the `<version>`.
-
-For `object_store` releases, use a string such as `object_store_0.4.0` as the `<version>`.
+Use a string such as `43.0.0` as the `<version>`.
 
 Create and push the tag thusly:
 
@@ -147,12 +134,6 @@ Rust Arrow Crates:
 ./dev/release/create-tarball.sh 4.1.0 2
 ```
 
-`object_store`:
-
-```shell
-./object_store/dev/release/create-tarball.sh 4.1.0 2
-```
-
 The `create-tarball.sh` script
 
 1. creates and uploads a release candidate tarball to the [arrow
@@ -198,7 +179,7 @@ For the release to become "official" it needs at least three Apache Arrow PMC me
 
 ## Verifying release candidates
 
-The `dev/release/verify-release-candidate.sh` or `object_store/dev/release/verify-release-candidate.sh` are scripts in this repository that can assist in the verification process. Run it like:
+The `dev/release/verify-release-candidate.sh` script in this repository can assist in the verification process. Run it like:
 
 ```
 ./dev/release/verify-release-candidate.sh 4.1.0 2
@@ -218,19 +199,13 @@ Rust Arrow Crates:
 ./dev/release/release-tarball.sh 4.1.0 2
 ```
 
-`object_store`
-
-```shell
-./object_store/dev/release/release-tarball.sh 4.1.0 2
-```
-
 Congratulations! The release is now official!
 
 ### Publish on Crates.io
 
-Only approved releases of the tarball should be published to
-crates.io, in order to conform to Apache Software Foundation
-governance standards.
+It is important that only approved releases of the tarball should be published
+to crates.io, in order to conform to Apache Software Foundation governance
+standards.
 
 An Arrow committer can publish this crate after an official project release has
 been made to crates.io using the following instructions.
@@ -269,9 +244,3 @@ Rust Arrow Crates:
 (cd parquet_derive && cargo publish)
 (cd arrow-integration-test && cargo publish)
 ```
-
-`object_store`
-
-```shell
-cargo publish
-```
diff --git a/dev/release/file_release_pr.sh b/dev/release/file_release_pr.sh
deleted file mode 100644
index 081b7c436..000000000
--- a/dev/release/file_release_pr.sh
+++ /dev/null
@@ -1,40 +0,0 @@
-#!/bin/bash
-#
-# 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.
-#
-# generates a standard release PR like this:
-# https://github.com/apache/arrow-rs/pull/2734
-#
-# Usage:
-# export BRANCH=<RELEASE_BRANCH> && export GITHUB_USERNAME=<USERNAME> && export GITHUB_TOKEN=<TOKEN> && ./file_release_pr.sh
-
-set -e
-
-FUTURE_RELEASE="35.0.0"
-ISSUE_NUMBER=3830
-
-TITLE="Update version to \`$FUTURE_RELEASE\` and update \`CHANGELOG\`"
-BODY="# Which issue does this PR close?\n\nCloses #$ISSUE_NUMBER.\n\n# Rationale for this change\nPrepare for biweekly release\n\n# What changes are included in this PR?\n\n# Are there any user-facing changes?\nYes"
-DATA="{\"title\":\"$TITLE\", \"body\":\"$BODY\", \"head\":\"$GITHUB_USERNAME:$BRANCH\",\"base\":\"master\"}"
-
-# Create the pull request
-curl -X POST \
-    -H "Accept: application/vnd.github+json" \
-    -H "Authorization: Bearer $GITHUB_TOKEN" \
-    https://api.github.com/repos/apache/arrow-rs/pulls \
-    -d "$DATA"
diff --git a/object_store/CHANGELOG-old.md b/object_store/CHANGELOG-old.md
index 172b0f90a..28dbde4e7 100644
--- a/object_store/CHANGELOG-old.md
+++ b/object_store/CHANGELOG-old.md
@@ -19,6 +19,23 @@
 
 # Historical Changelog
 
+## [object_store_0.11.0](https://github.com/apache/arrow-rs/tree/object_store_0.11.0) (2024-08-12)
+
+[Full Changelog](https://github.com/apache/arrow-rs/compare/object_store_0.10.2...object_store_0.11.0)
+
+**Breaking changes:**
+
+- Make object\_store errors non-exhaustive [\#6165](https://github.com/apache/arrow-rs/pull/6165) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([tustvold](https://github.com/tustvold))
+- Update snafu to `0.8.0` in object\_store \(\#5930\) [\#6070](https://github.com/apache/arrow-rs/pull/6070) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([alamb](https://github.com/alamb))
+
+
+**Merged pull requests:**
+
+- Add LICENSE and NOTICE files to object_store  [\#6234](https://github.com/apache/arrow-rs/pull/6234) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([alamb](https://github.com/alamb))
+- feat\(object\_store\):  add `PermissionDenied` variant to top-level error [\#6194](https://github.com/apache/arrow-rs/pull/6194) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([kyle-mccarthy](https://github.com/kyle-mccarthy))
+- Update object store MSRV to `1.64` [\#6123](https://github.com/apache/arrow-rs/pull/6123) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([alamb](https://github.com/alamb))
+- Fix clippy in object\_store crate [\#6120](https://github.com/apache/arrow-rs/pull/6120) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([alamb](https://github.com/alamb))
+
 ## [object_store_0.10.2](https://github.com/apache/arrow-rs/tree/object_store_0.10.2) (2024-07-17)
 
 [Full Changelog](https://github.com/apache/arrow-rs/compare/object_store_0.10.1...object_store_0.10.2)
diff --git a/object_store/CHANGELOG.md b/object_store/CHANGELOG.md
index 18dde1119..955859835 100644
--- a/object_store/CHANGELOG.md
+++ b/object_store/CHANGELOG.md
@@ -19,22 +19,42 @@
 
 # Changelog
 
-## [object_store_0.11.0](https://github.com/apache/arrow-rs/tree/object_store_0.11.0) (2024-08-12)
+## [object_store_0.11.1](https://github.com/apache/arrow-rs/tree/object_store_0.11.1) (2024-10-15)
 
-[Full Changelog](https://github.com/apache/arrow-rs/compare/object_store_0.10.2...object_store_0.11.0)
+[Full Changelog](https://github.com/apache/arrow-rs/compare/object_store_0.11.0...object_store_0.11.1)
 
-**Breaking changes:**
+**Implemented enhancements:**
 
-- Make object\_store errors non-exhaustive [\#6165](https://github.com/apache/arrow-rs/pull/6165) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([tustvold](https://github.com/tustvold))
-- Update snafu to `0.8.0` in object\_store \(\#5930\) [\#6070](https://github.com/apache/arrow-rs/pull/6070) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([alamb](https://github.com/alamb))
+- There is no way to pass object store client options as environment variables [\#6333](https://github.com/apache/arrow-rs/issues/6333) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
+- Better Document Backoff Algorithm [\#6324](https://github.com/apache/arrow-rs/issues/6324) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
+- Add direction to `list_with_offset` [\#6274](https://github.com/apache/arrow-rs/issues/6274) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
+- Support server-side encryption with customer-provided keys \(SSE-C\) [\#6229](https://github.com/apache/arrow-rs/issues/6229) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
 
+**Fixed bugs:**
+
+- \[object-store\] Requested tokio version is too old - does not compile [\#6458](https://github.com/apache/arrow-rs/issues/6458) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
+- Azure SAS tokens are visible when retry errors are logged via object\_store [\#6322](https://github.com/apache/arrow-rs/issues/6322) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
 
 **Merged pull requests:**
 
-- Add LICENSE and NOTICE files to object_store  [\#6234](https://github.com/apache/arrow-rs/pull/6234) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([alamb](https://github.com/alamb))
-- feat\(object\_store\):  add `PermissionDenied` variant to top-level error [\#6194](https://github.com/apache/arrow-rs/pull/6194) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([kyle-mccarthy](https://github.com/kyle-mccarthy))
-- Update object store MSRV to `1.64` [\#6123](https://github.com/apache/arrow-rs/pull/6123) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([alamb](https://github.com/alamb))
-- Fix clippy in object\_store crate [\#6120](https://github.com/apache/arrow-rs/pull/6120) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([alamb](https://github.com/alamb))
+- object\_store: fix typo in with\_connect\_timeout\_disabled that actually disabled non-connect timeouts [\#6563](https://github.com/apache/arrow-rs/pull/6563) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([adriangb](https://github.com/adriangb))
+- object\_store: Clarify what is a prefix in list\(\) documentation [\#6520](https://github.com/apache/arrow-rs/pull/6520) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([progval](https://github.com/progval))
+- object\_store: enable lint `unreachable_pub` [\#6512](https://github.com/apache/arrow-rs/pull/6512) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([ByteBaker](https://github.com/ByteBaker))
+- \[object\_store\] Retry S3 requests with 200 response with "Error" in body [\#6508](https://github.com/apache/arrow-rs/pull/6508) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([PeterKeDer](https://github.com/PeterKeDer))
+- \[object-store\] Require tokio 1.29.0. [\#6459](https://github.com/apache/arrow-rs/pull/6459) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([ashtuchkin](https://github.com/ashtuchkin))
+- feat: expose HTTP/2 max frame size in `object_store` [\#6442](https://github.com/apache/arrow-rs/pull/6442) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([crepererum](https://github.com/crepererum))
+- Derive `Clone` for `object_store::aws::AmazonS3` [\#6414](https://github.com/apache/arrow-rs/pull/6414) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([ethe](https://github.com/ethe))
+- object\_score: Support Azure Fabric OAuth Provider [\#6382](https://github.com/apache/arrow-rs/pull/6382) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([RobinLin666](https://github.com/RobinLin666))
+- `object_store::GetOptions` derive `Clone` [\#6361](https://github.com/apache/arrow-rs/pull/6361) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([samuelcolvin](https://github.com/samuelcolvin))
+- \[object\_store\] Propagate env vars as object store client options [\#6334](https://github.com/apache/arrow-rs/pull/6334) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([ccciudatu](https://github.com/ccciudatu))
+- docs\[object\_store\]: clarify the backoff strategy that is actually implemented [\#6325](https://github.com/apache/arrow-rs/pull/6325) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([westonpace](https://github.com/westonpace))
+- fix: azure sas token visible in logs [\#6323](https://github.com/apache/arrow-rs/pull/6323) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([alexwilcoxson-rel](https://github.com/alexwilcoxson-rel))
+- object\_store/delimited: Fix `TrailingEscape` condition [\#6265](https://github.com/apache/arrow-rs/pull/6265) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([Turbo87](https://github.com/Turbo87))
+- fix\(object\_store\): only add encryption headers for SSE-C in get request [\#6260](https://github.com/apache/arrow-rs/pull/6260) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([jiachengdb](https://github.com/jiachengdb))
+- docs: Add parquet\_opendal in related projects [\#6236](https://github.com/apache/arrow-rs/pull/6236) ([Xuanwo](https://github.com/Xuanwo))
+- feat\(object\_store\): add support for server-side encryption with customer-provided keys \(SSE-C\) [\#6230](https://github.com/apache/arrow-rs/pull/6230) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([jiachengdb](https://github.com/jiachengdb))
+- feat: further TLS options on ClientOptions: \#5034 [\#6148](https://github.com/apache/arrow-rs/pull/6148) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([ByteBaker](https://github.com/ByteBaker))
+
 
 
 \* *This Changelog was automatically generated by [github_changelog_generator](https://github.com/github-changelog-generator/github-changelog-generator)*
diff --git a/object_store/Cargo.toml b/object_store/Cargo.toml
index 3a903220c..cab2ac31c 100644
--- a/object_store/Cargo.toml
+++ b/object_store/Cargo.toml
@@ -17,7 +17,7 @@
 
 [package]
 name = "object_store"
-version = "0.11.0"
+version = "0.11.1"
 edition = "2021"
 license = "MIT/Apache-2.0"
 readme = "README.md"
diff --git a/object_store/dev/release/README.md b/object_store/dev/release/README.md
index 89f6e579b..4077dcad9 100644
--- a/object_store/dev/release/README.md
+++ b/object_store/dev/release/README.md
@@ -17,4 +17,209 @@
   under the License.
 -->
 
-See instructions in [`/dev/release/README.md`](../../../dev/release/README.md)
+
+# Release Process
+
+## Overview
+
+This file documents the release process for the `object_store` crate.
+
+At the time of writing, we release a new version of `object_store` on demand rather than on a regular schedule.
+
+As we are still in an early phase, we use the 0.x version scheme. If any code has 
+been merged to master that has a breaking API change, as defined in [Rust RFC 1105]
+the minor version number is incremented changed (e.g. `0.3.0` to `0.4.0`). 
+Otherwise the patch version is incremented (e.g. `0.3.0` to `0.3.1`).
+
+[Rust RFC 1105]: https://github.com/rust-lang/rfcs/blob/master/text/1105-api-evolution.md
+# Release Mechanics
+
+## Process Overview
+
+As part of the Apache governance model, official releases consist of
+signed source tarballs approved by the PMC.
+
+We then use the code in the approved source tarball to release to
+crates.io, the Rust ecosystem's package manager.
+
+We create a `CHANGELOG.md` so our users know what has been changed between releases.
+
+The CHANGELOG is created automatically using
+[update_change_log.sh](https://github.com/apache/arrow-rs/blob/master/object_store/dev/release/update_change_log.sh)
+
+This script creates a changelog using github issues and the
+labels associated with them.
+
+## Prepare CHANGELOG and version:
+
+Now prepare a PR to update `CHANGELOG.md` and versions on `master` to reflect the planned release.
+
+Note  this process is done in the `object_store` directory. See [#6227] for an example
+
+[#6227]: https://github.com/apache/arrow-rs/pull/6227
+
+```bash
+# NOTE: Run commands in object_store sub directory (not main repo checkout)
+# cd object_store
+
+git checkout master
+git pull
+git checkout -b <RELEASE_BRANCH>
+
+# Update versions. Make sure to run it before the next step since we do not want CHANGELOG-old.md affected.
+sed -i '' -e 's/0.11.0/0.11.1/g' `find . -name 'Cargo.toml' -or -name '*.md' | grep -v CHANGELOG`
+git commit -a -m 'Update version'
+
+# ensure your github token is available
+export CHANGELOG_GITHUB_TOKEN=<TOKEN>
+
+# manually edit ./dev/release/update_change_log.sh to reflect the release version
+# create the changelog
+./dev/release/update_change_log.sh
+
+# review change log / and edit associated issues and labels if needed, rerun update_change_log.sh
+
+# Commit changes
+git commit -a -m 'Create changelog'
+
+# push changes to fork and create a PR to master
+git push
+```
+
+Note that when reviewing the change log, rather than editing the
+`CHANGELOG.md`, it is preferred to update the issues and their labels
+(e.g. add `invalid` label to exclude them from release notes)
+
+Merge this PR to `master` prior to the next step.
+
+## Prepare release candidate tarball
+
+After you have merged the updates to the `CHANGELOG` and version,
+create a release candidate using the following steps. Note you need to
+be a committer to run these scripts as they upload to the apache `svn`
+distribution servers.
+
+### Create git tag for the release:
+
+While the official release artifact is a signed tarball, we also tag the commit it was created for convenience and code archaeology.
+
+For `object_store` releases, use a string such as `object_store_0.4.0` as the `<version>`.
+
+Create and push the tag thusly:
+
+```shell
+git fetch apache
+git tag <version> apache/master
+# push tag to apache
+git push apache <version>
+```
+
+### Pick an Release Candidate (RC) number
+
+Pick numbers in sequential order, with `1` for `rc1`, `2` for `rc2`, etc.
+
+### Create, sign, and upload tarball
+
+Run `create-tarball.sh` with the `<version>` tag and `<rc>` and you found in previous steps.
+
+```shell
+./object_store/dev/release/create-tarball.sh 0.11.1 1
+```
+
+The `create-tarball.sh` script
+
+1. creates and uploads a release candidate tarball to the [arrow
+   dev](https://dist.apache.org/repos/dist/dev/arrow) location on the
+   apache distribution svn server
+
+2. provide you an email template to
+   send to dev@arrow.apache.org for release voting.
+
+### Vote on Release Candidate tarball
+
+Send an email, based on the output from the script to dev@arrow.apache.org. The email should look like
+
+```
+Draft email for dev@arrow.apache.org mailing list
+
+---------------------------------------------------------
+To: dev@arrow.apache.org
+Subject: [VOTE][RUST] Release Apache Arrow Rust Object Store 0.11.1 RC1
+
+Hi,
+
+I would like to propose a release of Apache Arrow Rust Object
+Store Implementation, version 0.11.1.
+
+This release candidate is based on commit: b945b15de9085f5961a478d4f35b0c5c3427e248 [1]
+
+The proposed release tarball and signatures are hosted at [2].
+
+The changelog is located at [3].
+
+Please download, verify checksums and signatures, run the unit tests,
+and vote on the release. There is a script [4] that automates some of
+the verification.
+
+The vote will be open for at least 72 hours.
+
+[ ] +1 Release this as Apache Arrow Rust Object Store
+[ ] +0
+[ ] -1 Do not release this as Apache Arrow Rust Object Store  because...
+
+[1]: https://github.com/apache/arrow-rs/tree/b945b15de9085f5961a478d4f35b0c5c3427e248
+[2]: https://dist.apache.org/repos/dist/dev/arrow/apache-arrow-object-store-rs-0.11.1-rc1
+[3]: https://github.com/apache/arrow-rs/blob/b945b15de9085f5961a478d4f35b0c5c3427e248/object_store/CHANGELOG.md
+[4]: https://github.com/apache/arrow-rs/blob/master/object_store/dev/release/verify-release-candidate.sh
+```
+
+For the release to become "official" it needs at least three Apache Arrow PMC members to vote +1 on it.
+
+## Verifying release candidates
+
+The `object_store/dev/release/verify-release-candidate.sh` script can assist in the verification process. Run it like:
+
+```
+./object_store/dev/release/verify-release-candidate.sh 0.11.0 1
+```
+
+#### If the release is not approved
+
+If the release is not approved, fix whatever the problem is and try again with the next RC number
+
+### If the release is approved,
+
+Move tarball to the release location in SVN, e.g. https://dist.apache.org/repos/dist/release/arrow/arrow-4.1.0/, using the `release-tarball.sh` script:
+
+
+```shell
+./object_store/dev/release/release-tarball.sh 4.1.0 2
+```
+
+Congratulations! The release is now official!
+
+### Publish on Crates.io
+
+Only approved releases of the tarball should be published to
+crates.io, in order to conform to Apache Software Foundation
+governance standards.
+
+An Arrow committer can publish this crate after an official project release has
+been made to crates.io using the following instructions.
+
+Follow [these
+instructions](https://doc.rust-lang.org/cargo/reference/publishing.html) to
+create an account and login to crates.io before asking to be added as an owner
+of the [arrow crate](https://crates.io/crates/arrow).
+
+Download and unpack the official release tarball
+
+Verify that the Cargo.toml in the tarball contains the correct version
+(e.g. `version = "0.11.0"`) and then publish the crate with the
+following commands
+
+
+```shell
+cargo publish
+```
+
diff --git a/object_store/dev/release/update_change_log.sh b/object_store/dev/release/update_change_log.sh
index 142bbb076..30724478a 100755
--- a/object_store/dev/release/update_change_log.sh
+++ b/object_store/dev/release/update_change_log.sh
@@ -29,8 +29,8 @@
 
 set -e
 
-SINCE_TAG="object_store_0.10.2"
-FUTURE_RELEASE="object_store_0.11.0"
+SINCE_TAG="object_store_0.11.0"
+FUTURE_RELEASE="object_store_0.11.1"
 
 SOURCE_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 SOURCE_TOP_DIR="$(cd "${SOURCE_DIR}/../../" && pwd)"
diff --git a/object_store/src/client/mod.rs b/object_store/src/client/mod.rs
index 7b1b469a8..b65fea743 100644
--- a/object_store/src/client/mod.rs
+++ b/object_store/src/client/mod.rs
@@ -501,7 +501,7 @@ impl ClientOptions {
     ///
     /// See [`Self::with_connect_timeout`]
     pub fn with_connect_timeout_disabled(mut self) -> Self {
-        self.timeout = None;
+        self.connect_timeout = None;
         self
     }
 
diff --git a/parquet/Cargo.toml b/parquet/Cargo.toml
index 6d99c912f..8ab78efe0 100644
--- a/parquet/Cargo.toml
+++ b/parquet/Cargo.toml
@@ -121,6 +121,12 @@ sysinfo = ["dep:sysinfo"]
 # Verify 32-bit CRC checksum when decoding parquet pages
 crc = ["dep:crc32fast"]
 
+
+[[example]]
+name = "external_metadata"
+required-features = ["arrow", "async"]
+path = "./examples/external_metadata.rs"
+
 [[example]]
 name = "read_parquet"
 required-features = ["arrow"]
diff --git a/parquet/examples/external_metadata.rs b/parquet/examples/external_metadata.rs
new file mode 100644
index 000000000..5ca322ee2
--- /dev/null
+++ b/parquet/examples/external_metadata.rs
@@ -0,0 +1,236 @@
+// 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 arrow_array::{ArrayRef, Int32Array, RecordBatch, StringArray};
+use arrow_cast::pretty::pretty_format_batches;
+use futures::TryStreamExt;
+use parquet::arrow::arrow_reader::{ArrowReaderMetadata, ArrowReaderOptions};
+use parquet::arrow::{ArrowWriter, ParquetRecordBatchStreamBuilder};
+use parquet::file::metadata::{ParquetMetaData, ParquetMetaDataReader, ParquetMetaDataWriter};
+use parquet::file::properties::{EnabledStatistics, WriterProperties};
+use std::fs::File;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use tempfile::TempDir;
+
+/// This example demonstrates advanced usage of the Parquet metadata APIs.
+///
+/// It is sometimes desired to copy the metadata for parquet files stored on
+/// remote object storage (e.g. S3) to a local file or an in-memory cache, use a
+/// query engine like DataFusion to analyze the metadata to determine which file
+/// to read, and then read any matching files with a single subsequent object
+/// store request.
+///
+/// # Example Overview
+///
+/// 1. Reads the metadata of a Parquet file using [`ParquetMetaDataReader`]
+///
+/// 2. Removes column statistics from the metadata (to make it smaller)
+///
+/// 3. Stores the metadata in a separate file using [`ParquetMetaDataWriter`]
+///
+/// 4. Reads the metadata from the separate file and uses that to read the
+///    Parquet file, thus avoiding a second IO to read metadata or reparsing
+///    the footer.
+///
+#[tokio::main(flavor = "current_thread")]
+async fn main() -> parquet::errors::Result<()> {
+    let tempdir = TempDir::new().unwrap();
+    let parquet_path = create_parquet_file(&tempdir);
+    let metadata_path = tempdir.path().join("thrift_metadata.dat");
+
+    // In this example, we use a tokio file to mimic an async remote data source
+    let mut remote_parquet_file = tokio::fs::File::open(&parquet_path).await?;
+
+    let metadata = get_metadata_from_remote_parquet_file(&mut remote_parquet_file).await;
+    println!(
+        "Metadata from 'remote' Parquet file into memory: {} bytes",
+        metadata.memory_size()
+    );
+
+    // now slim down the metadata and write it to a "local" file
+    let metadata = prepare_metadata(metadata);
+    write_metadata_to_local_file(metadata, &metadata_path);
+
+    // now read the metadata from the local file and use it to read the "remote" Parquet file
+    let metadata = read_metadata_from_local_file(&metadata_path);
+    println!("Read metadata from file");
+
+    let batches = read_remote_parquet_file_with_metadata(remote_parquet_file, metadata).await;
+
+    // display the results
+    let batches_string = pretty_format_batches(&batches).unwrap().to_string();
+    let batches_lines: Vec<_> = batches_string.split('\n').collect();
+
+    assert_eq!(
+        batches_lines,
+        [
+            "+-----+-------------+",
+            "| id  | description |",
+            "+-----+-------------+",
+            "| 100 | oranges     |",
+            "| 200 | apples      |",
+            "| 201 | grapefruit  |",
+            "| 300 | bannanas    |",
+            "| 102 | grapes      |",
+            "| 33  | pears       |",
+            "+-----+-------------+",
+        ],
+        "actual output:\n\n{batches_lines:#?}"
+    );
+
+    Ok(())
+}
+
+/// Reads the metadata from a "remote" parquet file
+///
+/// Note that this function models reading from a remote file source using a
+/// tokio file. In a real application, you would implement [`MetadataFetch`] for
+/// your own remote source.
+///
+/// [`MetadataFetch`]: parquet::arrow::async_reader::MetadataFetch
+async fn get_metadata_from_remote_parquet_file(
+    remote_file: &mut tokio::fs::File,
+) -> ParquetMetaData {
+    // the remote source must know the total file size (e.g. from an object store LIST operation)
+    let file_size = remote_file.metadata().await.unwrap().len();
+
+    // tell the reader to read the page index
+    ParquetMetaDataReader::new()
+        .with_page_indexes(true)
+        .load_and_finish(remote_file, file_size as usize)
+        .await
+        .unwrap()
+}
+
+/// modifies the metadata to reduce its size
+fn prepare_metadata(metadata: ParquetMetaData) -> ParquetMetaData {
+    let orig_size = metadata.memory_size();
+
+    let mut builder = metadata.into_builder();
+
+    // remove column statistics to reduce the size of the metadata by converting
+    // the various structures into their respective builders and modifying them
+    // as needed.
+    for row_group in builder.take_row_groups() {
+        let mut row_group_builder = row_group.into_builder();
+        for column in row_group_builder.take_columns() {
+            let column = column.into_builder().clear_statistics().build().unwrap();
+            row_group_builder = row_group_builder.add_column_metadata(column);
+        }
+        let row_group = row_group_builder.build().unwrap();
+        builder = builder.add_row_group(row_group);
+    }
+    let metadata = builder.build();
+
+    // verifiy that the size has indeed been reduced
+    let new_size = metadata.memory_size();
+    assert!(new_size < orig_size, "metadata size did not decrease");
+    println!("Reduced metadata size from {} to {}", orig_size, new_size);
+    metadata
+}
+
+/// writes the metadata to a file
+///
+/// The data is stored using the same thrift format as the Parquet file metadata
+fn write_metadata_to_local_file(metadata: ParquetMetaData, file: impl AsRef<Path>) {
+    let file = File::create(file).unwrap();
+    ParquetMetaDataWriter::new(file, &metadata)
+        .finish()
+        .unwrap()
+}
+
+/// Reads the metadata from a file
+///
+/// This function reads the format written by `write_metadata_to_file`
+fn read_metadata_from_local_file(file: impl AsRef<Path>) -> ParquetMetaData {
+    let file = File::open(file).unwrap();
+    ParquetMetaDataReader::new()
+        .with_page_indexes(true)
+        .parse_and_finish(&file)
+        .unwrap()
+}
+
+/// Reads the "remote" Parquet file using the metadata
+///
+/// This shows how to read the Parquet file using previously read metadata
+/// instead of the metadata in the Parquet file itself. This avoids an IO /
+/// having to fetch and decode the metadata from the Parquet file before
+/// beginning to read it.
+///
+/// Note that this function models reading from a remote file source using a
+/// tokio file. In a real application, you would implement [`AsyncFileReader`]
+/// for your own remote source.
+///
+/// In this example, we simply buffer the results in memory as Arrow record
+/// batches but a real application would likely process the batches as they are
+/// read.
+///
+/// [`AsyncFileReader`]: parquet::arrow::async_reader::AsyncFileReader
+async fn read_remote_parquet_file_with_metadata(
+    remote_file: tokio::fs::File,
+    metadata: ParquetMetaData,
+) -> Vec<RecordBatch> {
+    let options = ArrowReaderOptions::new()
+        // tell the reader to read the page index
+        .with_page_index(true);
+    // create a reader with pre-existing metadata
+    let arrow_reader_metadata = ArrowReaderMetadata::try_new(metadata.into(), options).unwrap();
+    let reader =
+        ParquetRecordBatchStreamBuilder::new_with_metadata(remote_file, arrow_reader_metadata)
+            .build()
+            .unwrap();
+
+    reader.try_collect::<Vec<_>>().await.unwrap()
+}
+
+/// Make a new parquet file in the temporary directory, and returns the path
+fn create_parquet_file(tmpdir: &TempDir) -> PathBuf {
+    let path = tmpdir.path().join("example.parquet");
+    let new_file = File::create(&path).unwrap();
+
+    let batch = RecordBatch::try_from_iter(vec![
+        (
+            "id",
+            Arc::new(Int32Array::from(vec![100, 200, 201, 300, 102, 33])) as ArrayRef,
+        ),
+        (
+            "description",
+            Arc::new(StringArray::from(vec![
+                "oranges",
+                "apples",
+                "grapefruit",
+                "bannanas",
+                "grapes",
+                "pears",
+            ])),
+        ),
+    ])
+    .unwrap();
+
+    let props = WriterProperties::builder()
+        // ensure we write the page index level statistics
+        .set_statistics_enabled(EnabledStatistics::Page)
+        .build();
+
+    let mut writer = ArrowWriter::try_new(new_file, batch.schema(), Some(props)).unwrap();
+
+    writer.write(&batch).unwrap();
+    writer.finish().unwrap();
+
+    path
+}
diff --git a/parquet/src/arrow/arrow_reader/mod.rs b/parquet/src/arrow/arrow_reader/mod.rs
index a109851f7..d3709c03e 100644
--- a/parquet/src/arrow/arrow_reader/mod.rs
+++ b/parquet/src/arrow/arrow_reader/mod.rs
@@ -3077,6 +3077,108 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_read_binary_as_utf8() {
+        let file = write_parquet_from_iter(vec![
+            (
+                "binary_to_utf8",
+                Arc::new(BinaryArray::from(vec![
+                    b"one".as_ref(),
+                    b"two".as_ref(),
+                    b"three".as_ref(),
+                ])) as ArrayRef,
+            ),
+            (
+                "large_binary_to_large_utf8",
+                Arc::new(LargeBinaryArray::from(vec![
+                    b"one".as_ref(),
+                    b"two".as_ref(),
+                    b"three".as_ref(),
+                ])) as ArrayRef,
+            ),
+            (
+                "binary_view_to_utf8_view",
+                Arc::new(BinaryViewArray::from(vec![
+                    b"one".as_ref(),
+                    b"two".as_ref(),
+                    b"three".as_ref(),
+                ])) as ArrayRef,
+            ),
+        ]);
+        let supplied_fields = Fields::from(vec![
+            Field::new("binary_to_utf8", ArrowDataType::Utf8, false),
+            Field::new(
+                "large_binary_to_large_utf8",
+                ArrowDataType::LargeUtf8,
+                false,
+            ),
+            Field::new("binary_view_to_utf8_view", ArrowDataType::Utf8View, false),
+        ]);
+
+        let options = ArrowReaderOptions::new().with_schema(Arc::new(Schema::new(supplied_fields)));
+        let mut arrow_reader = ParquetRecordBatchReaderBuilder::try_new_with_options(
+            file.try_clone().unwrap(),
+            options,
+        )
+        .expect("reader builder with schema")
+        .build()
+        .expect("reader with schema");
+
+        let batch = arrow_reader.next().unwrap().unwrap();
+        assert_eq!(batch.num_columns(), 3);
+        assert_eq!(batch.num_rows(), 3);
+        assert_eq!(
+            batch
+                .column(0)
+                .as_string::<i32>()
+                .iter()
+                .collect::<Vec<_>>(),
+            vec![Some("one"), Some("two"), Some("three")]
+        );
+
+        assert_eq!(
+            batch
+                .column(1)
+                .as_string::<i64>()
+                .iter()
+                .collect::<Vec<_>>(),
+            vec![Some("one"), Some("two"), Some("three")]
+        );
+
+        assert_eq!(
+            batch.column(2).as_string_view().iter().collect::<Vec<_>>(),
+            vec![Some("one"), Some("two"), Some("three")]
+        );
+    }
+
+    #[test]
+    #[should_panic(expected = "Invalid UTF8 sequence at")]
+    fn test_read_non_utf8_binary_as_utf8() {
+        let file = write_parquet_from_iter(vec![(
+            "non_utf8_binary",
+            Arc::new(BinaryArray::from(vec![
+                b"\xDE\x00\xFF".as_ref(),
+                b"\xDE\x01\xAA".as_ref(),
+                b"\xDE\x02\xFF".as_ref(),
+            ])) as ArrayRef,
+        )]);
+        let supplied_fields = Fields::from(vec![Field::new(
+            "non_utf8_binary",
+            ArrowDataType::Utf8,
+            false,
+        )]);
+
+        let options = ArrowReaderOptions::new().with_schema(Arc::new(Schema::new(supplied_fields)));
+        let mut arrow_reader = ParquetRecordBatchReaderBuilder::try_new_with_options(
+            file.try_clone().unwrap(),
+            options,
+        )
+        .expect("reader builder with schema")
+        .build()
+        .expect("reader with schema");
+        arrow_reader.next().unwrap().unwrap_err();
+    }
+
     #[test]
     fn test_with_schema() {
         let nested_fields = Fields::from(vec![
diff --git a/parquet/src/arrow/async_writer/mod.rs b/parquet/src/arrow/async_writer/mod.rs
index 50bb5c046..8155b57d9 100644
--- a/parquet/src/arrow/async_writer/mod.rs
+++ b/parquet/src/arrow/async_writer/mod.rs
@@ -239,7 +239,11 @@ impl<W: AsyncFileWriter> AsyncArrowWriter<W> {
     /// Close and finalize the writer.
     ///
     /// All the data in the inner buffer will be force flushed.
-    pub async fn close(mut self) -> Result<FileMetaData> {
+    ///
+    /// Unlike [`Self::close`] this does not consume self
+    ///
+    /// Attempting to write after calling finish will result in an error
+    pub async fn finish(&mut self) -> Result<FileMetaData> {
         let metadata = self.sync_writer.finish()?;
 
         // Force to flush the remaining data.
@@ -249,6 +253,13 @@ impl<W: AsyncFileWriter> AsyncArrowWriter<W> {
         Ok(metadata)
     }
 
+    /// Close and finalize the writer.
+    ///
+    /// All the data in the inner buffer will be force flushed.
+    pub async fn close(mut self) -> Result<FileMetaData> {
+        self.finish().await
+    }
+
     /// Flush the data written by `sync_writer` into the `async_writer`
     ///
     /// # Notes
@@ -385,6 +396,28 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn test_async_writer_bytes_written() {
+        let col = Arc::new(Int64Array::from_iter_values([1, 2, 3])) as ArrayRef;
+        let to_write = RecordBatch::try_from_iter([("col", col)]).unwrap();
+
+        let temp = tempfile::tempfile().unwrap();
+
+        let file = tokio::fs::File::from_std(temp.try_clone().unwrap());
+        let mut writer =
+            AsyncArrowWriter::try_new(file.try_clone().await.unwrap(), to_write.schema(), None)
+                .unwrap();
+        writer.write(&to_write).await.unwrap();
+        let _metadata = writer.finish().await.unwrap();
+        // After `finish` this should include the metadata and footer
+        let reported = writer.bytes_written();
+
+        // Get actual size from file metadata
+        let actual = file.metadata().await.unwrap().len() as usize;
+
+        assert_eq!(reported, actual);
+    }
+
     #[tokio::test]
     async fn test_async_writer_file() {
         let col = Arc::new(Int64Array::from_iter_values([1, 2, 3])) as ArrayRef;
diff --git a/parquet/src/arrow/schema/primitive.rs b/parquet/src/arrow/schema/primitive.rs
index 17dd7862f..9f215b4dc 100644
--- a/parquet/src/arrow/schema/primitive.rs
+++ b/parquet/src/arrow/schema/primitive.rs
@@ -57,6 +57,11 @@ fn apply_hint(parquet: DataType, hint: DataType) -> DataType {
         (DataType::Utf8, DataType::LargeUtf8) => hint,
         (DataType::Binary, DataType::LargeBinary) => hint,
 
+        // Read as Utf8
+        (DataType::Binary, DataType::Utf8) => hint,
+        (DataType::Binary, DataType::LargeUtf8) => hint,
+        (DataType::Binary, DataType::Utf8View) => hint,
+
         // Determine view type
         (DataType::Utf8, DataType::Utf8View) => hint,
         (DataType::Binary, DataType::BinaryView) => hint,
diff --git a/parquet/src/file/metadata/mod.rs b/parquet/src/file/metadata/mod.rs
index fa56894f2..32b985710 100644
--- a/parquet/src/file/metadata/mod.rs
+++ b/parquet/src/file/metadata/mod.rs
@@ -36,30 +36,27 @@
 //!
 //! # APIs for working with Parquet Metadata
 //!
-//! The Parquet readers and writers in this crate read and write
+//! The Parquet readers and writers in this crate handle reading and writing
 //! metadata into parquet files. To work with metadata directly,
-//! the following APIs are available.
+//! the following APIs are available:
 //!
-//! Reading:
-//! * Read from bytes to `ParquetMetaData`: [`decode_footer`]
-//!   and [`decode_metadata`]
-//! * Read from an `async` source to `ParquetMetaData`: [`MetadataLoader`]
-//! * Read from bytes or from an async source to `ParquetMetaData`: [`ParquetMetaDataReader`]
+//! * [`ParquetMetaDataReader`] for reading
+//! * [`ParquetMetaDataWriter`] for writing.
 //!
-//! [`MetadataLoader`]: https://docs.rs/parquet/latest/parquet/arrow/async_reader/struct.MetadataLoader.html
-//! [`decode_footer`]: crate::file::footer::decode_footer
-//! [`decode_metadata`]: crate::file::footer::decode_metadata
+//! [`ParquetMetaDataReader`]: https://docs.rs/parquet/latest/parquet/file/metadata/struct.ParquetMetaDataReader.html
+//! [`ParquetMetaDataWriter`]: https://docs.rs/parquet/latest/parquet/file/metadata/struct.ParquetMetaDataWriter.html
 //!
-//! Writing:
-//! * Write `ParquetMetaData` to bytes in memory: [`ParquetMetaDataWriter`]
-//! * Writes `ParquetMetaData` to an async target: Not yet supported
+//! # Examples
 //!
+//! Please see [`external_metadata.rs`]
+//!
+//! [`external_metadata.rs`]: https://github.com/apache/arrow-rs/tree/master/parquet/examples/external_metadata.rs
 //!
 //! # Metadata Encodings and Structures
 //!
 //! There are three different encodings of Parquet Metadata in this crate:
 //!
-//! 1. `bytes`:encoded with the Thrift TCompactProtocol as defined in
+//! 1. `bytes`:encoded with the Thrift `TCompactProtocol` as defined in
 //!    [parquet.thrift]
 //!
 //! 2. [`format`]: Rust structures automatically generated by the thrift compiler


Test patch:
diff --git a/arrow-integration-testing/src/lib.rs b/arrow-integration-testing/src/lib.rs
index ba8e3876c..c8ce01e9f 100644
--- a/arrow-integration-testing/src/lib.rs
+++ b/arrow-integration-testing/src/lib.rs
@@ -28,7 +28,7 @@ use arrow::record_batch::RecordBatch;
 use arrow::util::test_util::arrow_test_data;
 use arrow_integration_test::*;
 use std::collections::HashMap;
-use std::ffi::{c_int, CStr, CString};
+use std::ffi::{c_char, c_int, CStr, CString};
 use std::fs::File;
 use std::io::BufReader;
 use std::iter::zip;
@@ -165,7 +165,7 @@ pub fn read_gzip_json(version: &str, path: &str) -> ArrowJson {
 
 /// C Data Integration entrypoint to export the schema from a JSON file
 fn cdata_integration_export_schema_from_json(
-    c_json_name: *const i8,
+    c_json_name: *const c_char,
     out: *mut FFI_ArrowSchema,
 ) -> Result<()> {
     let json_name = unsafe { CStr::from_ptr(c_json_name) };
@@ -178,7 +178,7 @@ fn cdata_integration_export_schema_from_json(
 
 /// C Data Integration entrypoint to export a batch from a JSON file
 fn cdata_integration_export_batch_from_json(
-    c_json_name: *const i8,
+    c_json_name: *const c_char,
     batch_num: c_int,
     out: *mut FFI_ArrowArray,
 ) -> Result<()> {
@@ -192,7 +192,7 @@ fn cdata_integration_export_batch_from_json(
 }
 
 fn cdata_integration_import_schema_and_compare_to_json(
-    c_json_name: *const i8,
+    c_json_name: *const c_char,
     c_schema: *mut FFI_ArrowSchema,
 ) -> Result<()> {
     let json_name = unsafe { CStr::from_ptr(c_json_name) };
@@ -229,7 +229,7 @@ fn compare_batches(a: &RecordBatch, b: &RecordBatch) -> Result<()> {
 }
 
 fn cdata_integration_import_batch_and_compare_to_json(
-    c_json_name: *const i8,
+    c_json_name: *const c_char,
     batch_num: c_int,
     c_array: *mut FFI_ArrowArray,
 ) -> Result<()> {
@@ -248,7 +248,7 @@ fn cdata_integration_import_batch_and_compare_to_json(
 }
 
 // If Result is an error, then export a const char* to its string display, otherwise NULL
-fn result_to_c_error<T, E: std::fmt::Display>(result: &std::result::Result<T, E>) -> *mut i8 {
+fn result_to_c_error<T, E: std::fmt::Display>(result: &std::result::Result<T, E>) -> *mut c_char {
     match result {
         Ok(_) => ptr::null_mut(),
         Err(e) => CString::new(format!("{}", e)).unwrap().into_raw(),
@@ -261,7 +261,7 @@ fn result_to_c_error<T, E: std::fmt::Display>(result: &std::result::Result<T, E>
 ///
 /// The pointer is assumed to have been obtained using CString::into_raw.
 #[no_mangle]
-pub unsafe extern "C" fn arrow_rs_free_error(c_error: *mut i8) {
+pub unsafe extern "C" fn arrow_rs_free_error(c_error: *mut c_char) {
     if !c_error.is_null() {
         drop(unsafe { CString::from_raw(c_error) });
     }
@@ -270,9 +270,9 @@ pub unsafe extern "C" fn arrow_rs_free_error(c_error: *mut i8) {
 /// A C-ABI for exporting an Arrow schema from a JSON file
 #[no_mangle]
 pub extern "C" fn arrow_rs_cdata_integration_export_schema_from_json(
-    c_json_name: *const i8,
+    c_json_name: *const c_char,
     out: *mut FFI_ArrowSchema,
-) -> *mut i8 {
+) -> *mut c_char {
     let r = cdata_integration_export_schema_from_json(c_json_name, out);
     result_to_c_error(&r)
 }
@@ -280,9 +280,9 @@ pub extern "C" fn arrow_rs_cdata_integration_export_schema_from_json(
 /// A C-ABI to compare an Arrow schema against a JSON file
 #[no_mangle]
 pub extern "C" fn arrow_rs_cdata_integration_import_schema_and_compare_to_json(
-    c_json_name: *const i8,
+    c_json_name: *const c_char,
     c_schema: *mut FFI_ArrowSchema,
-) -> *mut i8 {
+) -> *mut c_char {
     let r = cdata_integration_import_schema_and_compare_to_json(c_json_name, c_schema);
     result_to_c_error(&r)
 }
@@ -290,10 +290,10 @@ pub extern "C" fn arrow_rs_cdata_integration_import_schema_and_compare_to_json(
 /// A C-ABI for exporting a RecordBatch from a JSON file
 #[no_mangle]
 pub extern "C" fn arrow_rs_cdata_integration_export_batch_from_json(
-    c_json_name: *const i8,
+    c_json_name: *const c_char,
     batch_num: c_int,
     out: *mut FFI_ArrowArray,
-) -> *mut i8 {
+) -> *mut c_char {
     let r = cdata_integration_export_batch_from_json(c_json_name, batch_num, out);
     result_to_c_error(&r)
 }
@@ -301,10 +301,10 @@ pub extern "C" fn arrow_rs_cdata_integration_export_batch_from_json(
 /// A C-ABI to compare a RecordBatch against a JSON file
 #[no_mangle]
 pub extern "C" fn arrow_rs_cdata_integration_import_batch_and_compare_to_json(
-    c_json_name: *const i8,
+    c_json_name: *const c_char,
     batch_num: c_int,
     c_array: *mut FFI_ArrowArray,
-) -> *mut i8 {
+) -> *mut c_char {
     let r = cdata_integration_import_batch_and_compare_to_json(c_json_name, batch_num, c_array);
     result_to_c_error(&r)
 }
diff --git a/parquet/pytest/requirements.txt b/parquet/pytest/requirements.txt
index aa91a8bb6..40797e5e7 100644
--- a/parquet/pytest/requirements.txt
+++ b/parquet/pytest/requirements.txt
@@ -145,8 +145,8 @@ py4j==0.10.9.5 \
     --hash=sha256:276a4a3c5a2154df1860ef3303a927460e02e97b047dc0a47c1c3fb8cce34db6 \
     --hash=sha256:52d171a6a2b031d8a5d1de6efe451cf4f5baff1a2819aabc3741c8406539ba04
     # via pyspark
-pyspark==3.3.1 \
-    --hash=sha256:e99fa7de92be406884bfd831c32b9306a3a99de44cfc39a2eefb6ed07445d5fa
+pyspark==3.3.2 \
+    --hash=sha256:0dfd5db4300c1f6cc9c16d8dbdfb82d881b4b172984da71344ede1a9d4893da8
     # via -r requirements.in
 pytest==7.2.0 \
     --hash=sha256:892f933d339f068883b6fd5a459f03d85bfcb355e4981e146d2c7616c21fef71 \