Repository: apache/arrow-rs
Pull request #5566 (2024-03-29 00:57:32)

Update py03 from 0.20 to 0.21
# 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.
-->

Closes #.

# 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.
-->

# 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.
-->

Per guidance from pyo3 migration guide from 0.20 to 0.21:

https://pyo3.rs/v0.21.0/migration.html

Change usages of `&PyAny` to `Bound<PyAny>` and adjust the public trait `FromPyArrow` accordingly with a new function that accepts a bound.

# Are there any user-facing changes?


<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->

Yes

<!---
If there are any breaking changes to public APIs, please add the `breaking change` label.
-->


tustvold: Perhaps @wjones127 or @kylebarron might be able to advise on this
kylebarron: I haven't tried to upgrade my projects to 0.21 yet, but yes

> So it seems they are moving away from `&PyAny` is my understanding.

that's my understanding as well. I'd say that the PyArrow API should _probably_ follow suit. Maybe while upgrading to 0.21 is the best opportunity? 
abhiaagarwal: Thanks for this PR :)

Fyi, I pulled in this branch to update delta-rs (see delta-io/delta-rs#2363). When `gil-refs` is enabled it compiles successfully, but when `gil-refs` is disabled I get a traits bounds error.

```
error[E0277]: the trait bound `&str: PyClass` is not satisfied
    --> /Users/abhiagarwal/.cargo/git/checkouts/arrow-rs-d4122f2d122601f4/b1f5c49/arrow/src/pyarrow.rs:114:47
     |
114  |         let expected_module = expected_module.extract::<&str>()?;
     |                                               ^^^^^^^ the trait `PyClass` is not implemented for `&str`, which is required by `&str: FromPyObjectBound<'_, '_>`
     |
     = help: the following other types implement trait `FromPyObjectBound<'a, 'py>`:
               Cow<'a, [u8]>
               Cow<'a, str>
               &'a [u8]
     = note: required for `&str` to implement `pyo3::FromPyObject<'_>`
     = note: required for `&str` to implement `FromPyObjectBound<'_, '_>`
note: required by a bound in `pyo3::prelude::PyAnyMethods::extract`
    --> /Users/abhiagarwal/.cargo/registry/src/index.crates.io-6f17d22bba15001f/pyo3-0.21.0/src/types/any.rs:1649:12
     |
1647 |     fn extract<'a, T>(&'a self) -> PyResult<T>
     |        ------- required by a bound in this associated function
1648 |     where
1649 |         T: FromPyObjectBound<'a, 'py>;
     |            ^^^^^^^^^^^^^^^^^^^^^^^^^^ required by this bound in `PyAnyMethods::extract`

error[E0277]: the trait bound `&str: PyClass` is not satisfied
    --> /Users/abhiagarwal/.cargo/git/checkouts/arrow-rs-d4122f2d122601f4/b1f5c49/arrow/src/pyarrow.rs:116:43
     |
116  |         let expected_name = expected_name.extract::<&str>()?;
     |                                           ^^^^^^^ the trait `PyClass` is not implemented for `&str`, which is required by `&str: FromPyObjectBound<'_, '_>`
     |
     = help: the following other types implement trait `FromPyObjectBound<'a, 'py>`:
               Cow<'a, [u8]>
               Cow<'a, str>
               &'a [u8]
     = note: required for `&str` to implement `pyo3::FromPyObject<'_>`
     = note: required for `&str` to implement `FromPyObjectBound<'_, '_>`
note: required by a bound in `pyo3::prelude::PyAnyMethods::extract`
    --> /Users/abhiagarwal/.cargo/registry/src/index.crates.io-6f17d22bba15001f/pyo3-0.21.0/src/types/any.rs:1649:12
     |
1647 |     fn extract<'a, T>(&'a self) -> PyResult<T>
     |        ------- required by a bound in this associated function
1648 |     where
1649 |         T: FromPyObjectBound<'a, 'py>;
     |            ^^^^^^^^^^^^^^^^^^^^^^^^^^ required by this bound in `PyAnyMethods::extract`

error[E0277]: the trait bound `&str: PyClass` is not satisfied
    --> /Users/abhiagarwal/.cargo/git/checkouts/arrow-rs-d4122f2d122601f4/b1f5c49/arrow/src/pyarrow.rs:119:41
     |
119  |         let found_module = found_module.extract::<&str>()?;
     |                                         ^^^^^^^ the trait `PyClass` is not implemented for `&str`, which is required by `&str: FromPyObjectBound<'_, '_>`
     |
     = help: the following other types implement trait `FromPyObjectBound<'a, 'py>`:
               Cow<'a, [u8]>
               Cow<'a, str>
               &'a [u8]
     = note: required for `&str` to implement `pyo3::FromPyObject<'_>`
     = note: required for `&str` to implement `FromPyObjectBound<'_, '_>`
note: required by a bound in `pyo3::prelude::PyAnyMethods::extract`
    --> /Users/abhiagarwal/.cargo/registry/src/index.crates.io-6f17d22bba15001f/pyo3-0.21.0/src/types/any.rs:1649:12
     |
1647 |     fn extract<'a, T>(&'a self) -> PyResult<T>
     |        ------- required by a bound in this associated function
1648 |     where
1649 |         T: FromPyObjectBound<'a, 'py>;
     |            ^^^^^^^^^^^^^^^^^^^^^^^^^^ required by this bound in `PyAnyMethods::extract`

error[E0277]: the trait bound `&str: PyClass` is not satisfied
    --> /Users/abhiagarwal/.cargo/git/checkouts/arrow-rs-d4122f2d122601f4/b1f5c49/arrow/src/pyarrow.rs:121:37
     |
121  |         let found_name = found_name.extract::<&str>()?;
     |                                     ^^^^^^^ the trait `PyClass` is not implemented for `&str`, which is required by `&str: FromPyObjectBound<'_, '_>`
     |
     = help: the following other types implement trait `FromPyObjectBound<'a, 'py>`:
               Cow<'a, [u8]>
               Cow<'a, str>
               &'a [u8]
     = note: required for `&str` to implement `pyo3::FromPyObject<'_>`
     = note: required for `&str` to implement `FromPyObjectBound<'_, '_>`
note: required by a bound in `pyo3::prelude::PyAnyMethods::extract`
    --> /Users/abhiagarwal/.cargo/registry/src/index.crates.io-6f17d22bba15001f/pyo3-0.21.0/src/types/any.rs:1649:12
     |
1647 |     fn extract<'a, T>(&'a self) -> PyResult<T>
     |        ------- required by a bound in this associated function
1648 |     where
1649 |         T: FromPyObjectBound<'a, 'py>;
     |            ^^^^^^^^^^^^^^^^^^^^^^^^^^ required by this bound in `PyAnyMethods::extract`
```

The `FromPyObject` trait gets disabled without `gil-refs`, leading to the error. ([source](https://github.com/PyO3/pyo3/blob/63ba371db021985e01818694cc0987ee213fd4b9/src/conversions/std/slice.rs#L21-L43)) It should be able to be fixed by changing all the `extract` to  `extract_bound` per the migration guide. 

Patch:
diff --git a/arrow-buffer/src/buffer/scalar.rs b/arrow-buffer/src/buffer/scalar.rs
index 2019cc798..343b8549e 100644
--- a/arrow-buffer/src/buffer/scalar.rs
+++ b/arrow-buffer/src/buffer/scalar.rs
@@ -63,7 +63,7 @@ impl<T: ArrowNativeType> ScalarBuffer<T> {
     /// This method will panic if
     ///
     /// * `offset` or `len` would result in overflow
-    /// * `buffer` is not aligned to a multiple of `std::mem::size_of::<T>`
+    /// * `buffer` is not aligned to a multiple of `std::mem::align_of::<T>`
     /// * `bytes` is not large enough for the requested slice
     pub fn new(buffer: Buffer, offset: usize, len: usize) -> Self {
         let size = std::mem::size_of::<T>();
diff --git a/arrow-cast/src/cast/mod.rs b/arrow-cast/src/cast/mod.rs
index 52eb0d367..3e2bf4392 100644
--- a/arrow-cast/src/cast/mod.rs
+++ b/arrow-cast/src/cast/mod.rs
@@ -40,9 +40,11 @@
 mod decimal;
 mod dictionary;
 mod list;
+mod string;
 use crate::cast::decimal::*;
 use crate::cast::dictionary::*;
 use crate::cast::list::*;
+use crate::cast::string::*;
 
 use chrono::{NaiveTime, Offset, TimeZone, Utc};
 use std::cmp::Ordering;
@@ -2001,26 +2003,6 @@ where
     from.unary_opt::<_, R>(num::cast::cast::<T::Native, R::Native>)
 }
 
-fn value_to_string<O: OffsetSizeTrait>(
-    array: &dyn Array,
-    options: &CastOptions,
-) -> Result<ArrayRef, ArrowError> {
-    let mut builder = GenericStringBuilder::<O>::new();
-    let formatter = ArrayFormatter::try_new(array, &options.format_options)?;
-    let nulls = array.nulls();
-    for i in 0..array.len() {
-        match nulls.map(|x| x.is_null(i)).unwrap_or_default() {
-            true => builder.append_null(),
-            false => {
-                formatter.value(i).write(&mut builder)?;
-                // tell the builder the row is finished
-                builder.append_value("");
-            }
-        }
-    }
-    Ok(Arc::new(builder.finish()))
-}
-
 fn cast_numeric_to_binary<FROM: ArrowPrimitiveType, O: OffsetSizeTrait>(
     array: &dyn Array,
 ) -> Result<ArrayRef, ArrowError> {
@@ -2034,172 +2016,6 @@ fn cast_numeric_to_binary<FROM: ArrowPrimitiveType, O: OffsetSizeTrait>(
     )))
 }
 
-/// Parse UTF-8
-fn parse_string<P: Parser, O: OffsetSizeTrait>(
-    array: &dyn Array,
-    cast_options: &CastOptions,
-) -> Result<ArrayRef, ArrowError> {
-    let string_array = array.as_string::<O>();
-    let array = if cast_options.safe {
-        let iter = string_array.iter().map(|x| x.and_then(P::parse));
-
-        // Benefit:
-        //     20% performance improvement
-        // Soundness:
-        //     The iterator is trustedLen because it comes from an `StringArray`.
-        unsafe { PrimitiveArray::<P>::from_trusted_len_iter(iter) }
-    } else {
-        let v = string_array
-            .iter()
-            .map(|x| match x {
-                Some(v) => P::parse(v).ok_or_else(|| {
-                    ArrowError::CastError(format!(
-                        "Cannot cast string '{}' to value of {:?} type",
-                        v,
-                        P::DATA_TYPE
-                    ))
-                }),
-                None => Ok(P::Native::default()),
-            })
-            .collect::<Result<Vec<_>, ArrowError>>()?;
-        PrimitiveArray::new(v.into(), string_array.nulls().cloned())
-    };
-
-    Ok(Arc::new(array) as ArrayRef)
-}
-
-/// Casts generic string arrays to an ArrowTimestampType (TimeStampNanosecondArray, etc.)
-fn cast_string_to_timestamp<O: OffsetSizeTrait, T: ArrowTimestampType>(
-    array: &dyn Array,
-    to_tz: &Option<Arc<str>>,
-    cast_options: &CastOptions,
-) -> Result<ArrayRef, ArrowError> {
-    let array = array.as_string::<O>();
-    let out: PrimitiveArray<T> = match to_tz {
-        Some(tz) => {
-            let tz: Tz = tz.as_ref().parse()?;
-            cast_string_to_timestamp_impl(array, &tz, cast_options)?
-        }
-        None => cast_string_to_timestamp_impl(array, &Utc, cast_options)?,
-    };
-    Ok(Arc::new(out.with_timezone_opt(to_tz.clone())))
-}
-
-fn cast_string_to_timestamp_impl<O: OffsetSizeTrait, T: ArrowTimestampType, Tz: TimeZone>(
-    array: &GenericStringArray<O>,
-    tz: &Tz,
-    cast_options: &CastOptions,
-) -> Result<PrimitiveArray<T>, ArrowError> {
-    if cast_options.safe {
-        let iter = array.iter().map(|v| {
-            v.and_then(|v| {
-                let naive = string_to_datetime(tz, v).ok()?.naive_utc();
-                T::make_value(naive)
-            })
-        });
-        // Benefit:
-        //     20% performance improvement
-        // Soundness:
-        //     The iterator is trustedLen because it comes from an `StringArray`.
-
-        Ok(unsafe { PrimitiveArray::from_trusted_len_iter(iter) })
-    } else {
-        let vec = array
-            .iter()
-            .map(|v| {
-                v.map(|v| {
-                    let naive = string_to_datetime(tz, v)?.naive_utc();
-                    T::make_value(naive).ok_or_else(|| {
-                        ArrowError::CastError(format!(
-                            "Overflow converting {naive} to {:?}",
-                            T::UNIT
-                        ))
-                    })
-                })
-                .transpose()
-            })
-            .collect::<Result<Vec<Option<i64>>, _>>()?;
-
-        // Benefit:
-        //     20% performance improvement
-        // Soundness:
-        //     The iterator is trustedLen because it comes from an `StringArray`.
-        Ok(unsafe { PrimitiveArray::from_trusted_len_iter(vec.iter()) })
-    }
-}
-
-fn cast_string_to_interval<Offset, F, ArrowType>(
-    array: &dyn Array,
-    cast_options: &CastOptions,
-    parse_function: F,
-) -> Result<ArrayRef, ArrowError>
-where
-    Offset: OffsetSizeTrait,
-    ArrowType: ArrowPrimitiveType,
-    F: Fn(&str) -> Result<ArrowType::Native, ArrowError> + Copy,
-{
-    let string_array = array
-        .as_any()
-        .downcast_ref::<GenericStringArray<Offset>>()
-        .unwrap();
-    let interval_array = if cast_options.safe {
-        let iter = string_array
-            .iter()
-            .map(|v| v.and_then(|v| parse_function(v).ok()));
-
-        // Benefit:
-        //     20% performance improvement
-        // Soundness:
-        //     The iterator is trustedLen because it comes from an `StringArray`.
-        unsafe { PrimitiveArray::<ArrowType>::from_trusted_len_iter(iter) }
-    } else {
-        let vec = string_array
-            .iter()
-            .map(|v| v.map(parse_function).transpose())
-            .collect::<Result<Vec<_>, ArrowError>>()?;
-
-        // Benefit:
-        //     20% performance improvement
-        // Soundness:
-        //     The iterator is trustedLen because it comes from an `StringArray`.
-        unsafe { PrimitiveArray::<ArrowType>::from_trusted_len_iter(vec) }
-    };
-    Ok(Arc::new(interval_array) as ArrayRef)
-}
-
-fn cast_string_to_year_month_interval<Offset: OffsetSizeTrait>(
-    array: &dyn Array,
-    cast_options: &CastOptions,
-) -> Result<ArrayRef, ArrowError> {
-    cast_string_to_interval::<Offset, _, IntervalYearMonthType>(
-        array,
-        cast_options,
-        parse_interval_year_month,
-    )
-}
-
-fn cast_string_to_day_time_interval<Offset: OffsetSizeTrait>(
-    array: &dyn Array,
-    cast_options: &CastOptions,
-) -> Result<ArrayRef, ArrowError> {
-    cast_string_to_interval::<Offset, _, IntervalDayTimeType>(
-        array,
-        cast_options,
-        parse_interval_day_time,
-    )
-}
-
-fn cast_string_to_month_day_nano_interval<Offset: OffsetSizeTrait>(
-    array: &dyn Array,
-    cast_options: &CastOptions,
-) -> Result<ArrayRef, ArrowError> {
-    cast_string_to_interval::<Offset, _, IntervalMonthDayNanoType>(
-        array,
-        cast_options,
-        parse_interval_month_day_nano,
-    )
-}
-
 fn adjust_timestamp_to_timezone<T: ArrowTimestampType>(
     array: PrimitiveArray<Int64Type>,
     to_tz: &Tz,
@@ -2222,41 +2038,6 @@ fn adjust_timestamp_to_timezone<T: ArrowTimestampType>(
     Ok(adjusted)
 }
 
-/// Casts Utf8 to Boolean
-fn cast_utf8_to_boolean<OffsetSize>(
-    from: &dyn Array,
-    cast_options: &CastOptions,
-) -> Result<ArrayRef, ArrowError>
-where
-    OffsetSize: OffsetSizeTrait,
-{
-    let array = from
-        .as_any()
-        .downcast_ref::<GenericStringArray<OffsetSize>>()
-        .unwrap();
-
-    let output_array = array
-        .iter()
-        .map(|value| match value {
-            Some(value) => match value.to_ascii_lowercase().trim() {
-                "t" | "tr" | "tru" | "true" | "y" | "ye" | "yes" | "on" | "1" => Ok(Some(true)),
-                "f" | "fa" | "fal" | "fals" | "false" | "n" | "no" | "of" | "off" | "0" => {
-                    Ok(Some(false))
-                }
-                invalid_value => match cast_options.safe {
-                    true => Ok(None),
-                    false => Err(ArrowError::CastError(format!(
-                        "Cannot cast value '{invalid_value}' to value of Boolean type",
-                    ))),
-                },
-            },
-            None => Ok(None),
-        })
-        .collect::<Result<BooleanArray, _>>()?;
-
-    Ok(Arc::new(output_array))
-}
-
 /// Cast numeric types to Boolean
 ///
 /// Any zero value returns `false` while non-zero returns `true`
@@ -2325,37 +2106,6 @@ where
     unsafe { PrimitiveArray::<T>::from_trusted_len_iter(iter) }
 }
 
-/// A specified helper to cast from `GenericBinaryArray` to `GenericStringArray` when they have same
-/// offset size so re-encoding offset is unnecessary.
-fn cast_binary_to_string<O: OffsetSizeTrait>(
-    array: &dyn Array,
-    cast_options: &CastOptions,
-) -> Result<ArrayRef, ArrowError> {
-    let array = array
-        .as_any()
-        .downcast_ref::<GenericByteArray<GenericBinaryType<O>>>()
-        .unwrap();
-
-    match GenericStringArray::<O>::try_from_binary(array.clone()) {
-        Ok(a) => Ok(Arc::new(a)),
-        Err(e) => match cast_options.safe {
-            true => {
-                // Fallback to slow method to convert invalid sequences to nulls
-                let mut builder =
-                    GenericStringBuilder::<O>::with_capacity(array.len(), array.value_data().len());
-
-                let iter = array
-                    .iter()
-                    .map(|v| v.and_then(|v| std::str::from_utf8(v).ok()));
-
-                builder.extend(iter);
-                Ok(Arc::new(builder.finish()))
-            }
-            false => Err(e),
-        },
-    }
-}
-
 /// Helper function to cast from one `BinaryArray` or 'LargeBinaryArray' to 'FixedSizeBinaryArray'.
 fn cast_binary_to_fixed_size_binary<O: OffsetSizeTrait>(
     array: &dyn Array,
diff --git a/arrow-cast/src/cast/string.rs b/arrow-cast/src/cast/string.rs
new file mode 100644
index 000000000..e9c1ff58d
--- /dev/null
+++ b/arrow-cast/src/cast/string.rs
@@ -0,0 +1,270 @@
+// 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::cast::*;
+
+pub(crate) fn value_to_string<O: OffsetSizeTrait>(
+    array: &dyn Array,
+    options: &CastOptions,
+) -> Result<ArrayRef, ArrowError> {
+    let mut builder = GenericStringBuilder::<O>::new();
+    let formatter = ArrayFormatter::try_new(array, &options.format_options)?;
+    let nulls = array.nulls();
+    for i in 0..array.len() {
+        match nulls.map(|x| x.is_null(i)).unwrap_or_default() {
+            true => builder.append_null(),
+            false => {
+                formatter.value(i).write(&mut builder)?;
+                // tell the builder the row is finished
+                builder.append_value("");
+            }
+        }
+    }
+    Ok(Arc::new(builder.finish()))
+}
+
+/// Parse UTF-8
+pub(crate) fn parse_string<P: Parser, O: OffsetSizeTrait>(
+    array: &dyn Array,
+    cast_options: &CastOptions,
+) -> Result<ArrayRef, ArrowError> {
+    let string_array = array.as_string::<O>();
+    let array = if cast_options.safe {
+        let iter = string_array.iter().map(|x| x.and_then(P::parse));
+
+        // Benefit:
+        //     20% performance improvement
+        // Soundness:
+        //     The iterator is trustedLen because it comes from an `StringArray`.
+        unsafe { PrimitiveArray::<P>::from_trusted_len_iter(iter) }
+    } else {
+        let v = string_array
+            .iter()
+            .map(|x| match x {
+                Some(v) => P::parse(v).ok_or_else(|| {
+                    ArrowError::CastError(format!(
+                        "Cannot cast string '{}' to value of {:?} type",
+                        v,
+                        P::DATA_TYPE
+                    ))
+                }),
+                None => Ok(P::Native::default()),
+            })
+            .collect::<Result<Vec<_>, ArrowError>>()?;
+        PrimitiveArray::new(v.into(), string_array.nulls().cloned())
+    };
+
+    Ok(Arc::new(array) as ArrayRef)
+}
+
+/// Casts generic string arrays to an ArrowTimestampType (TimeStampNanosecondArray, etc.)
+pub(crate) fn cast_string_to_timestamp<O: OffsetSizeTrait, T: ArrowTimestampType>(
+    array: &dyn Array,
+    to_tz: &Option<Arc<str>>,
+    cast_options: &CastOptions,
+) -> Result<ArrayRef, ArrowError> {
+    let array = array.as_string::<O>();
+    let out: PrimitiveArray<T> = match to_tz {
+        Some(tz) => {
+            let tz: Tz = tz.as_ref().parse()?;
+            cast_string_to_timestamp_impl(array, &tz, cast_options)?
+        }
+        None => cast_string_to_timestamp_impl(array, &Utc, cast_options)?,
+    };
+    Ok(Arc::new(out.with_timezone_opt(to_tz.clone())))
+}
+
+fn cast_string_to_timestamp_impl<O: OffsetSizeTrait, T: ArrowTimestampType, Tz: TimeZone>(
+    array: &GenericStringArray<O>,
+    tz: &Tz,
+    cast_options: &CastOptions,
+) -> Result<PrimitiveArray<T>, ArrowError> {
+    if cast_options.safe {
+        let iter = array.iter().map(|v| {
+            v.and_then(|v| {
+                let naive = string_to_datetime(tz, v).ok()?.naive_utc();
+                T::make_value(naive)
+            })
+        });
+        // Benefit:
+        //     20% performance improvement
+        // Soundness:
+        //     The iterator is trustedLen because it comes from an `StringArray`.
+
+        Ok(unsafe { PrimitiveArray::from_trusted_len_iter(iter) })
+    } else {
+        let vec = array
+            .iter()
+            .map(|v| {
+                v.map(|v| {
+                    let naive = string_to_datetime(tz, v)?.naive_utc();
+                    T::make_value(naive).ok_or_else(|| {
+                        ArrowError::CastError(format!(
+                            "Overflow converting {naive} to {:?}",
+                            T::UNIT
+                        ))
+                    })
+                })
+                .transpose()
+            })
+            .collect::<Result<Vec<Option<i64>>, _>>()?;
+
+        // Benefit:
+        //     20% performance improvement
+        // Soundness:
+        //     The iterator is trustedLen because it comes from an `StringArray`.
+        Ok(unsafe { PrimitiveArray::from_trusted_len_iter(vec.iter()) })
+    }
+}
+
+pub(crate) fn cast_string_to_interval<Offset, F, ArrowType>(
+    array: &dyn Array,
+    cast_options: &CastOptions,
+    parse_function: F,
+) -> Result<ArrayRef, ArrowError>
+where
+    Offset: OffsetSizeTrait,
+    ArrowType: ArrowPrimitiveType,
+    F: Fn(&str) -> Result<ArrowType::Native, ArrowError> + Copy,
+{
+    let string_array = array
+        .as_any()
+        .downcast_ref::<GenericStringArray<Offset>>()
+        .unwrap();
+    let interval_array = if cast_options.safe {
+        let iter = string_array
+            .iter()
+            .map(|v| v.and_then(|v| parse_function(v).ok()));
+
+        // Benefit:
+        //     20% performance improvement
+        // Soundness:
+        //     The iterator is trustedLen because it comes from an `StringArray`.
+        unsafe { PrimitiveArray::<ArrowType>::from_trusted_len_iter(iter) }
+    } else {
+        let vec = string_array
+            .iter()
+            .map(|v| v.map(parse_function).transpose())
+            .collect::<Result<Vec<_>, ArrowError>>()?;
+
+        // Benefit:
+        //     20% performance improvement
+        // Soundness:
+        //     The iterator is trustedLen because it comes from an `StringArray`.
+        unsafe { PrimitiveArray::<ArrowType>::from_trusted_len_iter(vec) }
+    };
+    Ok(Arc::new(interval_array) as ArrayRef)
+}
+
+pub(crate) fn cast_string_to_year_month_interval<Offset: OffsetSizeTrait>(
+    array: &dyn Array,
+    cast_options: &CastOptions,
+) -> Result<ArrayRef, ArrowError> {
+    cast_string_to_interval::<Offset, _, IntervalYearMonthType>(
+        array,
+        cast_options,
+        parse_interval_year_month,
+    )
+}
+
+pub(crate) fn cast_string_to_day_time_interval<Offset: OffsetSizeTrait>(
+    array: &dyn Array,
+    cast_options: &CastOptions,
+) -> Result<ArrayRef, ArrowError> {
+    cast_string_to_interval::<Offset, _, IntervalDayTimeType>(
+        array,
+        cast_options,
+        parse_interval_day_time,
+    )
+}
+
+pub(crate) fn cast_string_to_month_day_nano_interval<Offset: OffsetSizeTrait>(
+    array: &dyn Array,
+    cast_options: &CastOptions,
+) -> Result<ArrayRef, ArrowError> {
+    cast_string_to_interval::<Offset, _, IntervalMonthDayNanoType>(
+        array,
+        cast_options,
+        parse_interval_month_day_nano,
+    )
+}
+
+/// A specified helper to cast from `GenericBinaryArray` to `GenericStringArray` when they have same
+/// offset size so re-encoding offset is unnecessary.
+pub(crate) fn cast_binary_to_string<O: OffsetSizeTrait>(
+    array: &dyn Array,
+    cast_options: &CastOptions,
+) -> Result<ArrayRef, ArrowError> {
+    let array = array
+        .as_any()
+        .downcast_ref::<GenericByteArray<GenericBinaryType<O>>>()
+        .unwrap();
+
+    match GenericStringArray::<O>::try_from_binary(array.clone()) {
+        Ok(a) => Ok(Arc::new(a)),
+        Err(e) => match cast_options.safe {
+            true => {
+                // Fallback to slow method to convert invalid sequences to nulls
+                let mut builder =
+                    GenericStringBuilder::<O>::with_capacity(array.len(), array.value_data().len());
+
+                let iter = array
+                    .iter()
+                    .map(|v| v.and_then(|v| std::str::from_utf8(v).ok()));
+
+                builder.extend(iter);
+                Ok(Arc::new(builder.finish()))
+            }
+            false => Err(e),
+        },
+    }
+}
+
+/// Casts Utf8 to Boolean
+pub(crate) fn cast_utf8_to_boolean<OffsetSize>(
+    from: &dyn Array,
+    cast_options: &CastOptions,
+) -> Result<ArrayRef, ArrowError>
+where
+    OffsetSize: OffsetSizeTrait,
+{
+    let array = from
+        .as_any()
+        .downcast_ref::<GenericStringArray<OffsetSize>>()
+        .unwrap();
+
+    let output_array = array
+        .iter()
+        .map(|value| match value {
+            Some(value) => match value.to_ascii_lowercase().trim() {
+                "t" | "tr" | "tru" | "true" | "y" | "ye" | "yes" | "on" | "1" => Ok(Some(true)),
+                "f" | "fa" | "fal" | "fals" | "false" | "n" | "no" | "of" | "off" | "0" => {
+                    Ok(Some(false))
+                }
+                invalid_value => match cast_options.safe {
+                    true => Ok(None),
+                    false => Err(ArrowError::CastError(format!(
+                        "Cannot cast value '{invalid_value}' to value of Boolean type",
+                    ))),
+                },
+            },
+            None => Ok(None),
+        })
+        .collect::<Result<BooleanArray, _>>()?;
+
+    Ok(Arc::new(output_array))
+}
diff --git a/arrow-flight/src/encode.rs b/arrow-flight/src/encode.rs
index efd688129..7604f3cd4 100644
--- a/arrow-flight/src/encode.rs
+++ b/arrow-flight/src/encode.rs
@@ -627,6 +627,7 @@ mod tests {
     use arrow_array::{cast::downcast_array, types::*};
     use arrow_buffer::Buffer;
     use arrow_cast::pretty::pretty_format_batches;
+    use arrow_ipc::MetadataVersion;
     use arrow_schema::UnionMode;
     use std::collections::HashMap;
 
@@ -638,7 +639,8 @@ mod tests {
     /// ensure only the batch's used data (not the allocated data) is sent
     /// <https://github.com/apache/arrow-rs/issues/208>
     fn test_encode_flight_data() {
-        let options = IpcWriteOptions::default();
+        // use 8-byte alignment - default alignment is 64 which produces bigger ipc data
+        let options = IpcWriteOptions::try_new(8, false, MetadataVersion::V5).unwrap();
         let c1 = UInt32Array::from(vec![1, 2, 3, 4, 5, 6]);
 
         let batch = RecordBatch::try_from_iter(vec![("a", Arc::new(c1) as ArrayRef)])
@@ -1343,6 +1345,8 @@ mod tests {
 
             let mut stream = FlightDataEncoderBuilder::new()
                 .with_max_flight_data_size(max_flight_data_size)
+                // use 8-byte alignment - default alignment is 64 which produces bigger ipc data
+                .with_options(IpcWriteOptions::try_new(8, false, MetadataVersion::V5).unwrap())
                 .build(futures::stream::iter([Ok(batch.clone())]));
 
             let mut i = 0;
diff --git a/arrow-ipc/src/convert.rs b/arrow-ipc/src/convert.rs
index 51e54215e..49da0efae 100644
--- a/arrow-ipc/src/convert.rs
+++ b/arrow-ipc/src/convert.rs
@@ -247,8 +247,10 @@ pub(crate) fn get_data_type(field: crate::Field, may_be_dictionary: bool) -> Dat
             }
         }
         crate::Type::Binary => DataType::Binary,
+        crate::Type::BinaryView => DataType::BinaryView,
         crate::Type::LargeBinary => DataType::LargeBinary,
         crate::Type::Utf8 => DataType::Utf8,
+        crate::Type::Utf8View => DataType::Utf8View,
         crate::Type::LargeUtf8 => DataType::LargeUtf8,
         crate::Type::FixedSizeBinary => {
             let fsb = field.type_as_fixed_size_binary().unwrap();
@@ -548,7 +550,16 @@ pub(crate) fn get_fb_field_type<'a>(
                 .as_union_value(),
             children: Some(fbb.create_vector(&empty_fields[..])),
         },
-        BinaryView | Utf8View => unimplemented!("unimplemented"),
+        BinaryView => FBFieldType {
+            type_type: crate::Type::BinaryView,
+            type_: crate::BinaryViewBuilder::new(fbb).finish().as_union_value(),
+            children: Some(fbb.create_vector(&empty_fields[..])),
+        },
+        Utf8View => FBFieldType {
+            type_type: crate::Type::Utf8View,
+            type_: crate::Utf8ViewBuilder::new(fbb).finish().as_union_value(),
+            children: Some(fbb.create_vector(&empty_fields[..])),
+        },
         Utf8 => FBFieldType {
             type_type: crate::Type::Utf8,
             type_: crate::Utf8Builder::new(fbb).finish().as_union_value(),
@@ -921,7 +932,9 @@ mod tests {
                     true,
                 ),
                 Field::new("utf8", DataType::Utf8, false),
+                Field::new("utf8_view", DataType::Utf8View, false),
                 Field::new("binary", DataType::Binary, false),
+                Field::new("binary_view", DataType::BinaryView, false),
                 Field::new_list("list[u8]", Field::new("item", DataType::UInt8, false), true),
                 Field::new_fixed_size_list(
                     "fixed_size_list[u8]",
diff --git a/arrow-ipc/src/reader.rs b/arrow-ipc/src/reader.rs
index dd0365da4..8eac17e20 100644
--- a/arrow-ipc/src/reader.rs
+++ b/arrow-ipc/src/reader.rs
@@ -25,7 +25,7 @@ mod stream;
 pub use stream::*;
 
 use flatbuffers::{VectorIter, VerifierOptions};
-use std::collections::HashMap;
+use std::collections::{HashMap, VecDeque};
 use std::fmt;
 use std::io::{BufReader, Read, Seek, SeekFrom};
 use std::sync::Arc;
@@ -64,6 +64,9 @@ fn read_buffer(
 
 /// Coordinates reading arrays based on data types.
 ///
+/// `variadic_counts` encodes the number of buffers to read for variadic types (e.g., Utf8View, BinaryView)
+/// When encounter such types, we pop from the front of the queue to get the number of buffers to read.
+///
 /// Notes:
 /// * In the IPC format, null buffers are always set, but may be empty. We discard them if an array has 0 nulls
 /// * Numeric values inside list arrays are often stored as 64-bit values regardless of their data type size.
@@ -71,7 +74,12 @@ fn read_buffer(
 ///     - check if the bit width of non-64-bit numbers is 64, and
 ///     - read the buffer as 64-bit (signed integer or float), and
 ///     - cast the 64-bit array to the appropriate data type
-fn create_array(reader: &mut ArrayReader, field: &Field) -> Result<ArrayRef, ArrowError> {
+fn create_array(
+    reader: &mut ArrayReader,
+    field: &Field,
+    variadic_counts: &mut VecDeque<i64>,
+    require_alignment: bool,
+) -> Result<ArrayRef, ArrowError> {
     let data_type = field.data_type();
     match data_type {
         Utf8 | Binary | LargeBinary | LargeUtf8 => create_primitive_array(
@@ -82,23 +90,54 @@ fn create_array(reader: &mut ArrayReader, field: &Field) -> Result<ArrayRef, Arr
                 reader.next_buffer()?,
                 reader.next_buffer()?,
             ],
+            require_alignment,
         ),
+        BinaryView | Utf8View => {
+            let count = variadic_counts
+                .pop_front()
+                .ok_or(ArrowError::IpcError(format!(
+                    "Missing variadic count for {data_type} column"
+                )))?;
+            let count = count + 2; // view and null buffer.
+            let buffers = (0..count)
+                .map(|_| reader.next_buffer())
+                .collect::<Result<Vec<_>, _>>()?;
+            create_primitive_array(
+                reader.next_node(field)?,
+                data_type,
+                &buffers,
+                require_alignment,
+            )
+        }
         FixedSizeBinary(_) => create_primitive_array(
             reader.next_node(field)?,
             data_type,
             &[reader.next_buffer()?, reader.next_buffer()?],
+            require_alignment,
         ),
         List(ref list_field) | LargeList(ref list_field) | Map(ref list_field, _) => {
             let list_node = reader.next_node(field)?;
             let list_buffers = [reader.next_buffer()?, reader.next_buffer()?];
-            let values = create_array(reader, list_field)?;
-            create_list_array(list_node, data_type, &list_buffers, values)
+            let values = create_array(reader, list_field, variadic_counts, require_alignment)?;
+            create_list_array(
+                list_node,
+                data_type,
+                &list_buffers,
+                values,
+                require_alignment,
+            )
         }
         FixedSizeList(ref list_field, _) => {
             let list_node = reader.next_node(field)?;
             let list_buffers = [reader.next_buffer()?];
-            let values = create_array(reader, list_field)?;
-            create_list_array(list_node, data_type, &list_buffers, values)
+            let values = create_array(reader, list_field, variadic_counts, require_alignment)?;
+            create_list_array(
+                list_node,
+                data_type,
+                &list_buffers,
+                values,
+                require_alignment,
+            )
         }
         Struct(struct_fields) => {
             let struct_node = reader.next_node(field)?;
@@ -109,7 +148,7 @@ fn create_array(reader: &mut ArrayReader, field: &Field) -> Result<ArrayRef, Arr
             // TODO investigate whether just knowing the number of buffers could
             // still work
             for struct_field in struct_fields {
-                let child = create_array(reader, struct_field)?;
+                let child = create_array(reader, struct_field, variadic_counts, require_alignment)?;
                 struct_arrays.push((struct_field.clone(), child));
             }
             let null_count = struct_node.null_count() as usize;
@@ -123,18 +162,24 @@ fn create_array(reader: &mut ArrayReader, field: &Field) -> Result<ArrayRef, Arr
         }
         RunEndEncoded(run_ends_field, values_field) => {
             let run_node = reader.next_node(field)?;
-            let run_ends = create_array(reader, run_ends_field)?;
-            let values = create_array(reader, values_field)?;
+            let run_ends =
+                create_array(reader, run_ends_field, variadic_counts, require_alignment)?;
+            let values = create_array(reader, values_field, variadic_counts, require_alignment)?;
 
             let run_array_length = run_node.length() as usize;
-            let data = ArrayData::builder(data_type.clone())
+            let builder = ArrayData::builder(data_type.clone())
                 .len(run_array_length)
                 .offset(0)
                 .add_child_data(run_ends.into_data())
-                .add_child_data(values.into_data())
-                .build_aligned()?;
+                .add_child_data(values.into_data());
+
+            let array_data = if require_alignment {
+                builder.build()?
+            } else {
+                builder.build_aligned()?
+            };
 
-            Ok(make_array(data))
+            Ok(make_array(array_data))
         }
         // Create dictionary array from RecordBatch
         Dictionary(_, _) => {
@@ -151,7 +196,13 @@ fn create_array(reader: &mut ArrayReader, field: &Field) -> Result<ArrayRef, Arr
                 ))
             })?;
 
-            create_dictionary_array(index_node, data_type, &index_buffers, value_array.clone())
+            create_dictionary_array(
+                index_node,
+                data_type,
+                &index_buffers,
+                value_array.clone(),
+                require_alignment,
+            )
         }
         Union(fields, mode) => {
             let union_node = reader.next_node(field)?;
@@ -177,7 +228,7 @@ fn create_array(reader: &mut ArrayReader, field: &Field) -> Result<ArrayRef, Arr
             let mut ids = Vec::with_capacity(fields.len());
 
             for (id, field) in fields.iter() {
-                let child = create_array(reader, field)?;
+                let child = create_array(reader, field, variadic_counts, require_alignment)?;
                 children.push((field.as_ref().clone(), child));
                 ids.push(id);
             }
@@ -196,18 +247,24 @@ fn create_array(reader: &mut ArrayReader, field: &Field) -> Result<ArrayRef, Arr
                 )));
             }
 
-            let data = ArrayData::builder(data_type.clone())
+            let builder = ArrayData::builder(data_type.clone())
                 .len(length as usize)
-                .offset(0)
-                .build_aligned()
-                .unwrap();
+                .offset(0);
+
+            let array_data = if require_alignment {
+                builder.build()?
+            } else {
+                builder.build_aligned()?
+            };
+
             // no buffer increases
-            Ok(Arc::new(NullArray::from(data)))
+            Ok(Arc::new(NullArray::from(array_data)))
         }
         _ => create_primitive_array(
             reader.next_node(field)?,
             data_type,
             &[reader.next_buffer()?, reader.next_buffer()?],
+            require_alignment,
         ),
     }
 }
@@ -218,29 +275,38 @@ fn create_primitive_array(
     field_node: &FieldNode,
     data_type: &DataType,
     buffers: &[Buffer],
+    require_alignment: bool,
 ) -> Result<ArrayRef, ArrowError> {
     let length = field_node.length() as usize;
     let null_buffer = (field_node.null_count() > 0).then_some(buffers[0].clone());
-    let array_data = match data_type {
+    let builder = match data_type {
         Utf8 | Binary | LargeBinary | LargeUtf8 => {
             // read 3 buffers: null buffer (optional), offsets buffer and data buffer
             ArrayData::builder(data_type.clone())
                 .len(length)
                 .buffers(buffers[1..3].to_vec())
                 .null_bit_buffer(null_buffer)
-                .build_aligned()?
         }
+        BinaryView | Utf8View => ArrayData::builder(data_type.clone())
+            .len(length)
+            .buffers(buffers[1..].to_vec())
+            .null_bit_buffer(null_buffer),
         _ if data_type.is_primitive() || matches!(data_type, Boolean | FixedSizeBinary(_)) => {
             // read 2 buffers: null buffer (optional) and data buffer
             ArrayData::builder(data_type.clone())
                 .len(length)
                 .add_buffer(buffers[1].clone())
                 .null_bit_buffer(null_buffer)
-                .build_aligned()?
         }
         t => unreachable!("Data type {:?} either unsupported or not primitive", t),
     };
 
+    let array_data = if require_alignment {
+        builder.build()?
+    } else {
+        builder.build_aligned()?
+    };
+
     Ok(make_array(array_data))
 }
 
@@ -251,6 +317,7 @@ fn create_list_array(
     data_type: &DataType,
     buffers: &[Buffer],
     child_array: ArrayRef,
+    require_alignment: bool,
 ) -> Result<ArrayRef, ArrowError> {
     let null_buffer = (field_node.null_count() > 0).then_some(buffers[0].clone());
     let length = field_node.length() as usize;
@@ -269,7 +336,14 @@ fn create_list_array(
 
         _ => unreachable!("Cannot create list or map array from {:?}", data_type),
     };
-    Ok(make_array(builder.build_aligned()?))
+
+    let array_data = if require_alignment {
+        builder.build()?
+    } else {
+        builder.build_aligned()?
+    };
+
+    Ok(make_array(array_data))
 }
 
 /// Reads the correct number of buffers based on list type and null_count, and creates a
@@ -279,6 +353,7 @@ fn create_dictionary_array(
     data_type: &DataType,
     buffers: &[Buffer],
     value_array: ArrayRef,
+    require_alignment: bool,
 ) -> Result<ArrayRef, ArrowError> {
     if let Dictionary(_, _) = *data_type {
         let null_buffer = (field_node.null_count() > 0).then_some(buffers[0].clone());
@@ -288,7 +363,13 @@ fn create_dictionary_array(
             .add_child_data(value_array.into_data())
             .null_bit_buffer(null_buffer);
 
-        Ok(make_array(builder.build_aligned()?))
+        let array_data = if require_alignment {
+            builder.build()?
+        } else {
+            builder.build_aligned()?
+        };
+
+        Ok(make_array(array_data))
     } else {
         unreachable!("Cannot create dictionary array from {:?}", data_type)
     }
@@ -328,7 +409,11 @@ impl<'a> ArrayReader<'a> {
         })
     }
 
-    fn skip_field(&mut self, field: &Field) -> Result<(), ArrowError> {
+    fn skip_field(
+        &mut self,
+        field: &Field,
+        variadic_count: &mut VecDeque<i64>,
+    ) -> Result<(), ArrowError> {
         self.next_node(field)?;
 
         match field.data_type() {
@@ -337,6 +422,18 @@ impl<'a> ArrayReader<'a> {
                     self.skip_buffer()
                 }
             }
+            Utf8View | BinaryView => {
+                let count = variadic_count
+                    .pop_front()
+                    .ok_or(ArrowError::IpcError(format!(
+                        "Missing variadic count for {} column",
+                        field.data_type()
+                    )))?;
+                let count = count + 2; // view and null buffer.
+                for _i in 0..count {
+                    self.skip_buffer()
+                }
+            }
             FixedSizeBinary(_) => {
                 self.skip_buffer();
                 self.skip_buffer();
@@ -344,23 +441,23 @@ impl<'a> ArrayReader<'a> {
             List(list_field) | LargeList(list_field) | Map(list_field, _) => {
                 self.skip_buffer();
                 self.skip_buffer();
-                self.skip_field(list_field)?;
+                self.skip_field(list_field, variadic_count)?;
             }
             FixedSizeList(list_field, _) => {
                 self.skip_buffer();
-                self.skip_field(list_field)?;
+                self.skip_field(list_field, variadic_count)?;
             }
             Struct(struct_fields) => {
                 self.skip_buffer();
 
                 // skip for each field
                 for struct_field in struct_fields {
-                    self.skip_field(struct_field)?
+                    self.skip_field(struct_field, variadic_count)?
                 }
             }
             RunEndEncoded(run_ends_field, values_field) => {
-                self.skip_field(run_ends_field)?;
-                self.skip_field(values_field)?;
+                self.skip_field(run_ends_field, variadic_count)?;
+                self.skip_field(values_field, variadic_count)?;
             }
             Dictionary(_, _) => {
                 self.skip_buffer(); // Nulls
@@ -375,7 +472,7 @@ impl<'a> ArrayReader<'a> {
                 };
 
                 for (_, field) in fields.iter() {
-                    self.skip_field(field)?
+                    self.skip_field(field, variadic_count)?
                 }
             }
             Null => {} // No buffer increases
@@ -388,7 +485,16 @@ impl<'a> ArrayReader<'a> {
     }
 }
 
-/// Creates a record batch from binary data using the `crate::RecordBatch` indexes and the `Schema`
+/// Creates a record batch from binary data using the `crate::RecordBatch` indexes and the `Schema`.
+///
+/// If `require_alignment` is true, this function will return an error if any array data in the
+/// input `buf` is not properly aligned.
+/// Under the hood it will use [`arrow_data::ArrayDataBuilder::build`] to construct [`arrow_data::ArrayData`].
+///
+/// If `require_alignment` is false, this function will automatically allocate a new aligned buffer
+/// and copy over the data if any array data in the input `buf` is not properly aligned.
+/// (Properly aligned array data will remain zero-copy.)
+/// Under the hood it will use [`arrow_data::ArrayDataBuilder::build_aligned`] to construct [`arrow_data::ArrayData`].
 pub fn read_record_batch(
     buf: &Buffer,
     batch: crate::RecordBatch,
@@ -396,6 +502,38 @@ pub fn read_record_batch(
     dictionaries_by_id: &HashMap<i64, ArrayRef>,
     projection: Option<&[usize]>,
     metadata: &MetadataVersion,
+) -> Result<RecordBatch, ArrowError> {
+    read_record_batch_impl(
+        buf,
+        batch,
+        schema,
+        dictionaries_by_id,
+        projection,
+        metadata,
+        false,
+    )
+}
+
+/// Read the dictionary from the buffer and provided metadata,
+/// updating the `dictionaries_by_id` with the resulting dictionary
+pub fn read_dictionary(
+    buf: &Buffer,
+    batch: crate::DictionaryBatch,
+    schema: &Schema,
+    dictionaries_by_id: &mut HashMap<i64, ArrayRef>,
+    metadata: &MetadataVersion,
+) -> Result<(), ArrowError> {
+    read_dictionary_impl(buf, batch, schema, dictionaries_by_id, metadata, false)
+}
+
+fn read_record_batch_impl(
+    buf: &Buffer,
+    batch: crate::RecordBatch,
+    schema: SchemaRef,
+    dictionaries_by_id: &HashMap<i64, ArrayRef>,
+    projection: Option<&[usize]>,
+    metadata: &MetadataVersion,
+    require_alignment: bool,
 ) -> Result<RecordBatch, ArrowError> {
     let buffers = batch.buffers().ok_or_else(|| {
         ArrowError::IpcError("Unable to get buffers from IPC RecordBatch".to_string())
@@ -403,6 +541,10 @@ pub fn read_record_batch(
     let field_nodes = batch.nodes().ok_or_else(|| {
         ArrowError::IpcError("Unable to get field nodes from IPC RecordBatch".to_string())
     })?;
+
+    let mut variadic_counts: VecDeque<i64> =
+        batch.variadicBufferCounts().into_iter().flatten().collect();
+
     let batch_compression = batch.compression();
     let compression = batch_compression
         .map(|batch_compression| batch_compression.codec().try_into())
@@ -425,12 +567,14 @@ pub fn read_record_batch(
         for (idx, field) in schema.fields().iter().enumerate() {
             // Create array for projected field
             if let Some(proj_idx) = projection.iter().position(|p| p == &idx) {
-                let child = create_array(&mut reader, field)?;
+                let child =
+                    create_array(&mut reader, field, &mut variadic_counts, require_alignment)?;
                 arrays.push((proj_idx, child));
             } else {
-                reader.skip_field(field)?;
+                reader.skip_field(field, &mut variadic_counts)?;
             }
         }
+        assert!(variadic_counts.is_empty());
         arrays.sort_by_key(|t| t.0);
         RecordBatch::try_new_with_options(
             Arc::new(schema.project(projection)?),
@@ -441,21 +585,21 @@ pub fn read_record_batch(
         let mut children = vec![];
         // keep track of index as lists require more than one node
         for field in schema.fields() {
-            let child = create_array(&mut reader, field)?;
+            let child = create_array(&mut reader, field, &mut variadic_counts, require_alignment)?;
             children.push(child);
         }
+        assert!(variadic_counts.is_empty());
         RecordBatch::try_new_with_options(schema, children, &options)
     }
 }
 
-/// Read the dictionary from the buffer and provided metadata,
-/// updating the `dictionaries_by_id` with the resulting dictionary
-pub fn read_dictionary(
+fn read_dictionary_impl(
     buf: &Buffer,
     batch: crate::DictionaryBatch,
     schema: &Schema,
     dictionaries_by_id: &mut HashMap<i64, ArrayRef>,
     metadata: &MetadataVersion,
+    require_alignment: bool,
 ) -> Result<(), ArrowError> {
     if batch.isDelta() {
         return Err(ArrowError::InvalidArgumentError(
@@ -478,13 +622,14 @@ pub fn read_dictionary(
             let value = value_type.as_ref().clone();
             let schema = Schema::new(vec![Field::new("", value, true)]);
             // Read a single column
-            let record_batch = read_record_batch(
+            let record_batch = read_record_batch_impl(
                 buf,
                 batch.data().unwrap(),
                 Arc::new(schema),
                 dictionaries_by_id,
                 None,
                 metadata,
+                require_alignment,
             )?;
             Some(record_batch.column(0).clone())
         }
@@ -609,6 +754,7 @@ pub struct FileDecoder {
     dictionaries: HashMap<i64, ArrayRef>,
     version: MetadataVersion,
     projection: Option<Vec<usize>>,
+    require_alignment: bool,
 }
 
 impl FileDecoder {
@@ -619,6 +765,7 @@ impl FileDecoder {
             version,
             dictionaries: Default::default(),
             projection: None,
+            require_alignment: false,
         }
     }
 
@@ -628,6 +775,23 @@ impl FileDecoder {
         self
     }
 
+    /// Specifies whether or not array data in input buffers is required to be properly aligned.
+    ///
+    /// If `require_alignment` is true, this decoder will return an error if any array data in the
+    /// input `buf` is not properly aligned.
+    /// Under the hood it will use [`arrow_data::ArrayDataBuilder::build`] to construct
+    /// [`arrow_data::ArrayData`].
+    ///
+    /// If `require_alignment` is false (the default), this decoder will automatically allocate a
+    /// new aligned buffer and copy over the data if any array data in the input `buf` is not
+    /// properly aligned. (Properly aligned array data will remain zero-copy.)
+    /// Under the hood it will use [`arrow_data::ArrayDataBuilder::build_aligned`] to construct
+    /// [`arrow_data::ArrayData`].
+    pub fn with_require_alignment(mut self, require_alignment: bool) -> Self {
+        self.require_alignment = require_alignment;
+        self
+    }
+
     fn read_message<'a>(&self, buf: &'a [u8]) -> Result<Message<'a>, ArrowError> {
         let message = parse_message(buf)?;
 
@@ -646,12 +810,13 @@ impl FileDecoder {
         match message.header_type() {
             crate::MessageHeader::DictionaryBatch => {
                 let batch = message.header_as_dictionary_batch().unwrap();
-                read_dictionary(
+                read_dictionary_impl(
                     &buf.slice(block.metaDataLength() as _),
                     batch,
                     &self.schema,
                     &mut self.dictionaries,
                     &message.version(),
+                    self.require_alignment,
                 )
             }
             t => Err(ArrowError::ParseError(format!(
@@ -676,13 +841,14 @@ impl FileDecoder {
                     ArrowError::IpcError("Unable to read IPC message as record batch".to_string())
                 })?;
                 // read the block that makes up the record batch into a buffer
-                read_record_batch(
+                read_record_batch_impl(
                     &buf.slice(block.metaDataLength() as _),
                     batch,
                     self.schema.clone(),
                     &self.dictionaries,
                     self.projection.as_deref(),
                     &message.version(),
+                    self.require_alignment,
                 )
                 .map(Some)
             }
@@ -1118,13 +1284,14 @@ impl<R: Read> StreamReader<R> {
                 let mut buf = MutableBuffer::from_len_zeroed(message.bodyLength() as usize);
                 self.reader.read_exact(&mut buf)?;
 
-                read_record_batch(
+                read_record_batch_impl(
                     &buf.into(),
                     batch,
                     self.schema(),
                     &self.dictionaries_by_id,
                     self.projection.as_ref().map(|x| x.0.as_ref()),
                     &message.version(),
+                    false,
                 )
                 .map(Some)
             }
@@ -1138,12 +1305,13 @@ impl<R: Read> StreamReader<R> {
                 let mut buf = MutableBuffer::from_len_zeroed(message.bodyLength() as usize);
                 self.reader.read_exact(&mut buf)?;
 
-                read_dictionary(
+                read_dictionary_impl(
                     &buf.into(),
                     batch,
                     &self.schema,
                     &mut self.dictionaries_by_id,
                     &message.version(),
+                    false,
                 )?;
 
                 // read the next message until we encounter a RecordBatch
@@ -1759,6 +1927,121 @@ mod tests {
         assert_eq!(input_batch, output_batch);
     }
 
+    const LONG_TEST_STRING: &str =
+        "This is a long string to make sure binary view array handles it";
+
+    #[test]
+    fn test_roundtrip_view_types() {
+        let schema = Schema::new(vec![
+            Field::new("field_1", DataType::BinaryView, true),
+            Field::new("field_2", DataType::Utf8, true),
+            Field::new("field_3", DataType::Utf8View, true),
+        ]);
+        let bin_values: Vec<Option<&[u8]>> = vec![
+            Some(b"foo"),
+            None,
+            Some(b"bar"),
+            Some(LONG_TEST_STRING.as_bytes()),
+        ];
+        let utf8_values: Vec<Option<&str>> =
+            vec![Some("foo"), None, Some("bar"), Some(LONG_TEST_STRING)];
+        let bin_view_array = BinaryViewArray::from_iter(bin_values);
+        let utf8_array = StringArray::from_iter(utf8_values.iter());
+        let utf8_view_array = StringViewArray::from_iter(utf8_values);
+        let record_batch = RecordBatch::try_new(
+            Arc::new(schema.clone()),
+            vec![
+                Arc::new(bin_view_array),
+                Arc::new(utf8_array),
+                Arc::new(utf8_view_array),
+            ],
+        )
+        .unwrap();
+
+        assert_eq!(record_batch, roundtrip_ipc(&record_batch));
+        assert_eq!(record_batch, roundtrip_ipc_stream(&record_batch));
+
+        let sliced_batch = record_batch.slice(1, 2);
+        assert_eq!(sliced_batch, roundtrip_ipc(&sliced_batch));
+        assert_eq!(sliced_batch, roundtrip_ipc_stream(&sliced_batch));
+    }
+
+    #[test]
+    fn test_roundtrip_view_types_nested_dict() {
+        let bin_values: Vec<Option<&[u8]>> = vec![
+            Some(b"foo"),
+            None,
+            Some(b"bar"),
+            Some(LONG_TEST_STRING.as_bytes()),
+            Some(b"field"),
+        ];
+        let utf8_values: Vec<Option<&str>> = vec![
+            Some("foo"),
+            None,
+            Some("bar"),
+            Some(LONG_TEST_STRING),
+            Some("field"),
+        ];
+        let bin_view_array = Arc::new(BinaryViewArray::from_iter(bin_values));
+        let utf8_view_array = Arc::new(StringViewArray::from_iter(utf8_values));
+
+        let key_dict_keys = Int8Array::from_iter_values([0, 0, 1, 2, 0, 1, 3]);
+        let key_dict_array = DictionaryArray::new(key_dict_keys, utf8_view_array.clone());
+        let keys_field = Arc::new(Field::new_dict(
+            "keys",
+            DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Utf8View)),
+            true,
+            1,
+            false,
+        ));
+
+        let value_dict_keys = Int8Array::from_iter_values([0, 3, 0, 1, 2, 0, 1]);
+        let value_dict_array = DictionaryArray::new(value_dict_keys, bin_view_array);
+        let values_field = Arc::new(Field::new_dict(
+            "values",
+            DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::BinaryView)),
+            true,
+            2,
+            false,
+        ));
+        let entry_struct = StructArray::from(vec![
+            (keys_field, make_array(key_dict_array.into_data())),
+            (values_field, make_array(value_dict_array.into_data())),
+        ]);
+
+        let map_data_type = DataType::Map(
+            Arc::new(Field::new(
+                "entries",
+                entry_struct.data_type().clone(),
+                false,
+            )),
+            false,
+        );
+        let entry_offsets = Buffer::from_slice_ref([0, 2, 4, 7]);
+        let map_data = ArrayData::builder(map_data_type)
+            .len(3)
+            .add_buffer(entry_offsets)
+            .add_child_data(entry_struct.into_data())
+            .build()
+            .unwrap();
+        let map_array = MapArray::from(map_data);
+
+        let dict_keys = Int8Array::from_iter_values([0, 1, 0, 1, 1, 2, 0, 1, 2]);
+        let dict_dict_array = DictionaryArray::new(dict_keys, Arc::new(map_array));
+        let schema = Arc::new(Schema::new(vec![Field::new(
+            "f1",
+            dict_dict_array.data_type().clone(),
+            false,
+        )]));
+        let batch = RecordBatch::try_new(schema, vec![Arc::new(dict_dict_array)]).unwrap();
+        assert_eq!(batch, roundtrip_ipc(&batch));
+        assert_eq!(batch, roundtrip_ipc_stream(&batch));
+
+        let sliced_batch = batch.slice(1, 2);
+        assert_eq!(sliced_batch, roundtrip_ipc(&sliced_batch));
+        assert_eq!(sliced_batch, roundtrip_ipc_stream(&sliced_batch));
+    }
+
     #[test]
     fn test_no_columns_batch() {
         let schema = Arc::new(Schema::empty());
@@ -1794,18 +2077,61 @@ mod tests {
         assert_ne!(b.as_ptr().align_offset(8), 0);
 
         let ipc_batch = message.header_as_record_batch().unwrap();
-        let roundtrip = read_record_batch(
+        let roundtrip = read_record_batch_impl(
             &b,
             ipc_batch,
             batch.schema(),
             &Default::default(),
             None,
             &message.version(),
+            false,
         )
         .unwrap();
         assert_eq!(batch, roundtrip);
     }
 
+    #[test]
+    fn test_unaligned_throws_error_with_require_alignment() {
+        let batch = RecordBatch::try_from_iter(vec![(
+            "i32",
+            Arc::new(Int32Array::from(vec![1, 2, 3, 4])) as _,
+        )])
+        .unwrap();
+
+        let gen = IpcDataGenerator {};
+        let mut dict_tracker = DictionaryTracker::new(false);
+        let (_, encoded) = gen
+            .encoded_batch(&batch, &mut dict_tracker, &Default::default())
+            .unwrap();
+
+        let message = root_as_message(&encoded.ipc_message).unwrap();
+
+        // Construct an unaligned buffer
+        let mut buffer = MutableBuffer::with_capacity(encoded.arrow_data.len() + 1);
+        buffer.push(0_u8);
+        buffer.extend_from_slice(&encoded.arrow_data);
+        let b = Buffer::from(buffer).slice(1);
+        assert_ne!(b.as_ptr().align_offset(8), 0);
+
+        let ipc_batch = message.header_as_record_batch().unwrap();
+        let result = read_record_batch_impl(
+            &b,
+            ipc_batch,
+            batch.schema(),
+            &Default::default(),
+            None,
+            &message.version(),
+            true,
+        );
+
+        let error = result.unwrap_err();
+        assert_eq!(
+            error.to_string(),
+            "Invalid argument error: Misaligned buffers[0] in array of type Int32, \
+             offset from expected alignment of 4 by 1"
+        );
+    }
+
     #[test]
     fn test_file_with_massive_column_count() {
         // 499_999 is upper limit for default settings (1_000_000)
diff --git a/arrow-ipc/src/reader/stream.rs b/arrow-ipc/src/reader/stream.rs
index 780722817..64191a22b 100644
--- a/arrow-ipc/src/reader/stream.rs
+++ b/arrow-ipc/src/reader/stream.rs
@@ -24,7 +24,7 @@ use arrow_buffer::{Buffer, MutableBuffer};
 use arrow_schema::{ArrowError, SchemaRef};
 
 use crate::convert::MessageBuffer;
-use crate::reader::{read_dictionary, read_record_batch};
+use crate::reader::{read_dictionary_impl, read_record_batch_impl};
 use crate::{MessageHeader, CONTINUATION_MARKER};
 
 /// A low-level interface for reading [`RecordBatch`] data from a stream of bytes
@@ -40,6 +40,8 @@ pub struct StreamDecoder {
     state: DecoderState,
     /// A scratch buffer when a read is split across multiple `Buffer`
     buf: MutableBuffer,
+    /// Whether or not array data in input buffers are required to be aligned
+    require_alignment: bool,
 }
 
 #[derive(Debug)]
@@ -83,6 +85,23 @@ impl StreamDecoder {
         Self::default()
     }
 
+    /// Specifies whether or not array data in input buffers is required to be properly aligned.
+    ///
+    /// If `require_alignment` is true, this decoder will return an error if any array data in the
+    /// input `buf` is not properly aligned.
+    /// Under the hood it will use [`arrow_data::ArrayDataBuilder::build`] to construct
+    /// [`arrow_data::ArrayData`].
+    ///
+    /// If `require_alignment` is false (the default), this decoder will automatically allocate a
+    /// new aligned buffer and copy over the data if any array data in the input `buf` is not
+    /// properly aligned. (Properly aligned array data will remain zero-copy.)
+    /// Under the hood it will use [`arrow_data::ArrayDataBuilder::build_aligned`] to construct
+    /// [`arrow_data::ArrayData`].
+    pub fn with_require_alignment(mut self, require_alignment: bool) -> Self {
+        self.require_alignment = require_alignment;
+        self
+    }
+
     /// Try to read the next [`RecordBatch`] from the provided [`Buffer`]
     ///
     /// [`Buffer::advance`] will be called on `buffer` for any consumed bytes.
@@ -192,13 +211,14 @@ impl StreamDecoder {
                             let schema = self.schema.clone().ok_or_else(|| {
                                 ArrowError::IpcError("Missing schema".to_string())
                             })?;
-                            let batch = read_record_batch(
+                            let batch = read_record_batch_impl(
                                 &body,
                                 batch,
                                 schema,
                                 &self.dictionaries,
                                 None,
                                 &version,
+                                self.require_alignment,
                             )?;
                             self.state = DecoderState::default();
                             return Ok(Some(batch));
@@ -208,12 +228,13 @@ impl StreamDecoder {
                             let schema = self.schema.as_deref().ok_or_else(|| {
                                 ArrowError::IpcError("Missing schema".to_string())
                             })?;
-                            read_dictionary(
+                            read_dictionary_impl(
                                 &body,
                                 dictionary,
                                 schema,
                                 &mut self.dictionaries,
                                 &version,
+                                self.require_alignment,
                             )?;
                             self.state = DecoderState::default();
                         }
diff --git a/arrow-ipc/src/writer.rs b/arrow-ipc/src/writer.rs
index 22edfbc24..97136bd97 100644
--- a/arrow-ipc/src/writer.rs
+++ b/arrow-ipc/src/writer.rs
@@ -43,8 +43,8 @@ use crate::CONTINUATION_MARKER;
 #[derive(Debug, Clone)]
 pub struct IpcWriteOptions {
     /// Write padding after memory buffers to this multiple of bytes.
-    /// Generally 8 or 64, defaults to 64
-    alignment: usize,
+    /// Must be 8, 16, 32, or 64 - defaults to 64.
+    alignment: u8,
     /// The legacy format is for releases before 0.15.0, and uses metadata V4
     write_legacy_ipc_format: bool,
     /// The metadata version to write. The Rust IPC writer supports V4+
@@ -87,11 +87,14 @@ impl IpcWriteOptions {
         write_legacy_ipc_format: bool,
         metadata_version: crate::MetadataVersion,
     ) -> Result<Self, ArrowError> {
-        if alignment == 0 || alignment % 8 != 0 {
+        let is_alignment_valid =
+            alignment == 8 || alignment == 16 || alignment == 32 || alignment == 64;
+        if !is_alignment_valid {
             return Err(ArrowError::InvalidArgumentError(
-                "Alignment should be greater than 0 and be a multiple of 8".to_string(),
+                "Alignment should be 8, 16, 32, or 64.".to_string(),
             ));
         }
+        let alignment: u8 = u8::try_from(alignment).expect("range already checked");
         match metadata_version {
             crate::MetadataVersion::V1
             | crate::MetadataVersion::V2
@@ -412,6 +415,8 @@ impl IpcDataGenerator {
         let compression_codec: Option<CompressionCodec> =
             batch_compression_type.map(TryInto::try_into).transpose()?;
 
+        let mut variadic_buffer_counts = vec![];
+
         for array in batch.columns() {
             let array_data = array.to_data();
             offset = write_array_data(
@@ -425,15 +430,23 @@ impl IpcDataGenerator {
                 compression_codec,
                 write_options,
             )?;
+
+            append_variadic_buffer_counts(&mut variadic_buffer_counts, &array_data);
         }
         // pad the tail of body data
         let len = arrow_data.len();
-        let pad_len = pad_to_8(len as u32);
-        arrow_data.extend_from_slice(&vec![0u8; pad_len][..]);
+        let pad_len = pad_to_alignment(write_options.alignment, len);
+        arrow_data.extend_from_slice(&PADDING[..pad_len]);
 
         // write data
         let buffers = fbb.create_vector(&buffers);
         let nodes = fbb.create_vector(&nodes);
+        let variadic_buffer = if variadic_buffer_counts.is_empty() {
+            None
+        } else {
+            Some(fbb.create_vector(&variadic_buffer_counts))
+        };
+
         let root = {
             let mut batch_builder = crate::RecordBatchBuilder::new(&mut fbb);
             batch_builder.add_length(batch.num_rows() as i64);
@@ -442,6 +455,10 @@ impl IpcDataGenerator {
             if let Some(c) = compression {
                 batch_builder.add_compression(c);
             }
+
+            if let Some(v) = variadic_buffer {
+                batch_builder.add_variadicBufferCounts(v);
+            }
             let b = batch_builder.finish();
             b.as_union_value()
         };
@@ -501,14 +518,22 @@ impl IpcDataGenerator {
             write_options,
         )?;
 
+        let mut variadic_buffer_counts = vec![];
+        append_variadic_buffer_counts(&mut variadic_buffer_counts, array_data);
+
         // pad the tail of body data
         let len = arrow_data.len();
-        let pad_len = pad_to_8(len as u32);
-        arrow_data.extend_from_slice(&vec![0u8; pad_len][..]);
+        let pad_len = pad_to_alignment(write_options.alignment, len);
+        arrow_data.extend_from_slice(&PADDING[..pad_len]);
 
         // write data
         let buffers = fbb.create_vector(&buffers);
         let nodes = fbb.create_vector(&nodes);
+        let variadic_buffer = if variadic_buffer_counts.is_empty() {
+            None
+        } else {
+            Some(fbb.create_vector(&variadic_buffer_counts))
+        };
 
         let root = {
             let mut batch_builder = crate::RecordBatchBuilder::new(&mut fbb);
@@ -518,6 +543,9 @@ impl IpcDataGenerator {
             if let Some(c) = compression {
                 batch_builder.add_compression(c);
             }
+            if let Some(v) = variadic_buffer {
+                batch_builder.add_variadicBufferCounts(v);
+            }
             batch_builder.finish()
         };
 
@@ -547,6 +575,25 @@ impl IpcDataGenerator {
     }
 }
 
+fn append_variadic_buffer_counts(counts: &mut Vec<i64>, array: &ArrayData) {
+    match array.data_type() {
+        DataType::BinaryView | DataType::Utf8View => {
+            // The spec documents the counts only includes the variadic buffers, not the view/null buffers.
+            // https://arrow.apache.org/docs/format/Columnar.html#variadic-buffers
+            counts.push(array.buffers().len() as i64 - 1);
+        }
+        DataType::Dictionary(_, _) => {
+            // Do nothing
+            // Dictionary types are handled in `encode_dictionaries`.
+        }
+        _ => {
+            for child in array.child_data() {
+                append_variadic_buffer_counts(counts, child)
+            }
+        }
+    }
+}
+
 pub(crate) fn unslice_run_array(arr: ArrayData) -> Result<ArrayData, ArrowError> {
     match arr.data_type() {
         DataType::RunEndEncoded(k, _) => match k.data_type() {
@@ -716,11 +763,11 @@ impl<W: Write> FileWriter<W> {
     ) -> Result<Self, ArrowError> {
         let data_gen = IpcDataGenerator::default();
         let mut writer = BufWriter::new(writer);
-        // write magic to header aligned on 8 byte boundary
-        let header_size = super::ARROW_MAGIC.len() + 2;
-        assert_eq!(header_size, 8);
-        writer.write_all(&super::ARROW_MAGIC[..])?;
-        writer.write_all(&[0, 0])?;
+        // write magic to header aligned on alignment boundary
+        let pad_len = pad_to_alignment(write_options.alignment, super::ARROW_MAGIC.len());
+        let header_size = super::ARROW_MAGIC.len() + pad_len;
+        writer.write_all(&super::ARROW_MAGIC)?;
+        writer.write_all(&PADDING[..pad_len])?;
         // 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)?;
@@ -1017,13 +1064,13 @@ pub fn write_message<W: Write>(
     write_options: &IpcWriteOptions,
 ) -> Result<(usize, usize), ArrowError> {
     let arrow_data_len = encoded.arrow_data.len();
-    if arrow_data_len % 8 != 0 {
+    if arrow_data_len % usize::from(write_options.alignment) != 0 {
         return Err(ArrowError::MemoryError(
             "Arrow data not aligned".to_string(),
         ));
     }
 
-    let a = write_options.alignment - 1;
+    let a = usize::from(write_options.alignment - 1);
     let buffer = encoded.ipc_message;
     let flatbuf_size = buffer.len();
     let prefix_size = if write_options.write_legacy_ipc_format {
@@ -1045,11 +1092,11 @@ pub fn write_message<W: Write>(
         writer.write_all(&buffer)?;
     }
     // write padding
-    writer.write_all(&vec![0; padding_bytes])?;
+    writer.write_all(&PADDING[..padding_bytes])?;
 
     // write arrow data
     let body_len = if arrow_data_len > 0 {
-        write_body_buffers(&mut writer, &encoded.arrow_data)?
+        write_body_buffers(&mut writer, &encoded.arrow_data, write_options.alignment)?
     } else {
         0
     };
@@ -1057,19 +1104,23 @@ pub fn write_message<W: Write>(
     Ok((aligned_size, body_len))
 }
 
-fn write_body_buffers<W: Write>(mut writer: W, data: &[u8]) -> Result<usize, ArrowError> {
-    let len = data.len() as u32;
-    let pad_len = pad_to_8(len) as u32;
+fn write_body_buffers<W: Write>(
+    mut writer: W,
+    data: &[u8],
+    alignment: u8,
+) -> Result<usize, ArrowError> {
+    let len = data.len();
+    let pad_len = pad_to_alignment(alignment, len);
     let total_len = len + pad_len;
 
     // write body buffer
     writer.write_all(data)?;
     if pad_len > 0 {
-        writer.write_all(&vec![0u8; pad_len as usize][..])?;
+        writer.write_all(&PADDING[..pad_len])?;
     }
 
     writer.flush()?;
-    Ok(total_len as usize)
+    Ok(total_len)
 }
 
 /// Write a record batch to the writer, writing the message size before the message
@@ -1234,6 +1285,7 @@ fn write_array_data(
             arrow_data,
             offset,
             compression_codec,
+            write_options.alignment,
         )?;
     }
 
@@ -1247,6 +1299,24 @@ fn write_array_data(
                 arrow_data,
                 offset,
                 compression_codec,
+                write_options.alignment,
+            )?;
+        }
+    } else if matches!(data_type, DataType::BinaryView | DataType::Utf8View) {
+        // Slicing the views buffer is safe and easy,
+        // but pruning unneeded data buffers is much more nuanced since it's complicated to prove that no views reference the pruned buffers
+        //
+        // Current implementation just serialize the raw arrays as given and not try to optimize anything.
+        // If users wants to "compact" the arrays prior to sending them over IPC,
+        // they should consider the gc API suggested in #5513
+        for buffer in array_data.buffers() {
+            offset = write_buffer(
+                buffer.as_slice(),
+                buffers,
+                arrow_data,
+                offset,
+                compression_codec,
+                write_options.alignment,
             )?;
         }
     } else if matches!(data_type, DataType::LargeBinary | DataType::LargeUtf8) {
@@ -1258,6 +1328,7 @@ fn write_array_data(
                 arrow_data,
                 offset,
                 compression_codec,
+                write_options.alignment,
             )?;
         }
     } else if DataType::is_numeric(data_type)
@@ -1283,7 +1354,14 @@ fn write_array_data(
         } else {
             buffer.as_slice()
         };
-        offset = write_buffer(buffer_slice, buffers, arrow_data, offset, compression_codec)?;
+        offset = write_buffer(
+            buffer_slice,
+            buffers,
+            arrow_data,
+            offset,
+            compression_codec,
+            write_options.alignment,
+        )?;
     } else if matches!(data_type, DataType::Boolean) {
         // Bools are special because the payload (= 1 bit) is smaller than the physical container elements (= bytes).
         // The array data may not start at the physical boundary of the underlying buffer, so we need to shift bits around.
@@ -1291,7 +1369,14 @@ fn write_array_data(
 
         let buffer = &array_data.buffers()[0];
         let buffer = buffer.bit_slice(array_data.offset(), array_data.len());
-        offset = write_buffer(&buffer, buffers, arrow_data, offset, compression_codec)?;
+        offset = write_buffer(
+            &buffer,
+            buffers,
+            arrow_data,
+            offset,
+            compression_codec,
+            write_options.alignment,
+        )?;
     } else if matches!(
         data_type,
         DataType::List(_) | DataType::LargeList(_) | DataType::Map(_, _)
@@ -1312,6 +1397,7 @@ fn write_array_data(
             arrow_data,
             offset,
             compression_codec,
+            write_options.alignment,
         )?;
         offset = write_array_data(
             &sliced_child_data,
@@ -1327,7 +1413,14 @@ fn write_array_data(
         return Ok(offset);
     } else {
         for buffer in array_data.buffers() {
-            offset = write_buffer(buffer, buffers, arrow_data, offset, compression_codec)?;
+            offset = write_buffer(
+                buffer,
+                buffers,
+                arrow_data,
+                offset,
+                compression_codec,
+                write_options.alignment,
+            )?;
         }
     }
 
@@ -1391,6 +1484,7 @@ fn write_buffer(
     arrow_data: &mut Vec<u8>,         // output stream
     offset: i64,                      // current output stream offset
     compression_codec: Option<CompressionCodec>,
+    alignment: u8,
 ) -> Result<i64, ArrowError> {
     let len: i64 = match compression_codec {
         Some(compressor) => compressor.compress_to_vec(buffer, arrow_data)?,
@@ -1406,17 +1500,20 @@ fn write_buffer(
 
     // make new index entry
     buffers.push(crate::Buffer::new(offset, len));
-    // padding and make offset 8 bytes aligned
-    let pad_len = pad_to_8(len as u32) as i64;
-    arrow_data.extend_from_slice(&vec![0u8; pad_len as usize][..]);
+    // padding and make offset aligned
+    let pad_len = pad_to_alignment(alignment, len as usize);
+    arrow_data.extend_from_slice(&PADDING[..pad_len]);
 
-    Ok(offset + len + pad_len)
+    Ok(offset + len + (pad_len as i64))
 }
 
-/// Calculate an 8-byte boundary and return the number of bytes needed to pad to 8 bytes
+const PADDING: [u8; 64] = [0; 64];
+
+/// Calculate an alignment boundary and return the number of bytes needed to pad to the alignment boundary
 #[inline]
-fn pad_to_8(len: u32) -> usize {
-    (((len + 7) & !7) - len) as usize
+fn pad_to_alignment(alignment: u8, len: usize) -> usize {
+    let a = usize::from(alignment - 1);
+    ((len + a) & !a) - len
 }
 
 #[cfg(test)]
@@ -1430,7 +1527,9 @@ mod tests {
     use arrow_array::builder::{PrimitiveRunBuilder, UInt32Builder};
     use arrow_array::types::*;
 
+    use crate::convert::fb_to_schema;
     use crate::reader::*;
+    use crate::root_as_footer;
     use crate::MetadataVersion;
 
     use super::*;
@@ -1448,7 +1547,17 @@ mod tests {
     }
 
     fn serialize_stream(record: &RecordBatch) -> Vec<u8> {
-        let mut stream_writer = StreamWriter::try_new(vec![], record.schema_ref()).unwrap();
+        // Use 8-byte alignment so that the various `truncate_*` tests can be compactly written,
+        // without needing to construct a giant array to spill over the 64-byte default alignment
+        // boundary.
+        const IPC_ALIGNMENT: usize = 8;
+
+        let mut stream_writer = StreamWriter::try_new_with_options(
+            vec![],
+            record.schema_ref(),
+            IpcWriteOptions::try_new(IPC_ALIGNMENT, false, MetadataVersion::V5).unwrap(),
+        )
+        .unwrap();
         stream_writer.write(record).unwrap();
         stream_writer.finish().unwrap();
         stream_writer.into_inner().unwrap()
@@ -1804,6 +1913,57 @@ mod tests {
         write_union_file(IpcWriteOptions::try_new(8, false, MetadataVersion::V5).unwrap());
     }
 
+    #[test]
+    fn test_write_view_types() {
+        const LONG_TEST_STRING: &str =
+            "This is a long string to make sure binary view array handles it";
+        let schema = Schema::new(vec![
+            Field::new("field1", DataType::BinaryView, true),
+            Field::new("field2", DataType::Utf8View, true),
+        ]);
+        let values: Vec<Option<&[u8]>> = vec![
+            Some(b"foo"),
+            Some(b"bar"),
+            Some(LONG_TEST_STRING.as_bytes()),
+        ];
+        let binary_array = BinaryViewArray::from_iter(values);
+        let utf8_array =
+            StringViewArray::from_iter(vec![Some("foo"), Some("bar"), Some(LONG_TEST_STRING)]);
+        let record_batch = RecordBatch::try_new(
+            Arc::new(schema.clone()),
+            vec![Arc::new(binary_array), Arc::new(utf8_array)],
+        )
+        .unwrap();
+
+        let mut file = tempfile::tempfile().unwrap();
+        {
+            let mut writer = FileWriter::try_new(&mut file, &schema).unwrap();
+            writer.write(&record_batch).unwrap();
+            writer.finish().unwrap();
+        }
+        file.rewind().unwrap();
+        {
+            let mut reader = FileReader::try_new(&file, None).unwrap();
+            let read_batch = reader.next().unwrap().unwrap();
+            read_batch
+                .columns()
+                .iter()
+                .zip(record_batch.columns())
+                .for_each(|(a, b)| {
+                    assert_eq!(a, b);
+                });
+        }
+        file.rewind().unwrap();
+        {
+            let mut reader = FileReader::try_new(&file, Some(vec![0])).unwrap();
+            let read_batch = reader.next().unwrap().unwrap();
+            assert_eq!(read_batch.num_columns(), 1);
+            let read_array = read_batch.column(0);
+            let write_array = record_batch.column(0);
+            assert_eq!(read_array, write_array);
+        }
+    }
+
     #[test]
     fn truncate_ipc_record_batch() {
         fn create_batch(rows: usize) -> RecordBatch {
@@ -2234,4 +2394,120 @@ mod tests {
         let in_batch = RecordBatch::try_new(schema, vec![values]).unwrap();
         roundtrip_ensure_sliced_smaller(in_batch, 1000);
     }
+
+    #[test]
+    fn test_decimal128_alignment16_is_sufficient() {
+        const IPC_ALIGNMENT: usize = 16;
+
+        // Test a bunch of different dimensions to ensure alignment is never an issue.
+        // For example, if we only test `num_cols = 1` then even with alignment 8 this
+        // test would _happen_ to pass, even though for different dimensions like
+        // `num_cols = 2` it would fail.
+        for num_cols in [1, 2, 3, 17, 50, 73, 99] {
+            let num_rows = (num_cols * 7 + 11) % 100; // Deterministic swizzle
+
+            let mut fields = Vec::new();
+            let mut arrays = Vec::new();
+            for i in 0..num_cols {
+                let field = Field::new(&format!("col_{}", i), DataType::Decimal128(38, 10), true);
+                let array = Decimal128Array::from(vec![num_cols as i128; num_rows]);
+                fields.push(field);
+                arrays.push(Arc::new(array) as Arc<dyn Array>);
+            }
+            let schema = Schema::new(fields);
+            let batch = RecordBatch::try_new(Arc::new(schema), arrays).unwrap();
+
+            let mut writer = FileWriter::try_new_with_options(
+                Vec::new(),
+                batch.schema_ref(),
+                IpcWriteOptions::try_new(IPC_ALIGNMENT, false, MetadataVersion::V5).unwrap(),
+            )
+            .unwrap();
+            writer.write(&batch).unwrap();
+            writer.finish().unwrap();
+
+            let out: Vec<u8> = writer.into_inner().unwrap();
+
+            let buffer = Buffer::from_vec(out);
+            let trailer_start = buffer.len() - 10;
+            let footer_len =
+                read_footer_length(buffer[trailer_start..].try_into().unwrap()).unwrap();
+            let footer =
+                root_as_footer(&buffer[trailer_start - footer_len..trailer_start]).unwrap();
+
+            let schema = fb_to_schema(footer.schema().unwrap());
+
+            // Importantly we set `require_alignment`, checking that 16-byte alignment is sufficient
+            // for `read_record_batch` later on to read the data in a zero-copy manner.
+            let decoder =
+                FileDecoder::new(Arc::new(schema), footer.version()).with_require_alignment(true);
+
+            let batches = footer.recordBatches().unwrap();
+
+            let block = batches.get(0);
+            let block_len = block.bodyLength() as usize + block.metaDataLength() as usize;
+            let data = buffer.slice_with_length(block.offset() as _, block_len);
+
+            let batch2 = decoder.read_record_batch(block, &data).unwrap().unwrap();
+
+            assert_eq!(batch, batch2);
+        }
+    }
+
+    #[test]
+    fn test_decimal128_alignment8_is_unaligned() {
+        const IPC_ALIGNMENT: usize = 8;
+
+        let num_cols = 2;
+        let num_rows = 1;
+
+        let mut fields = Vec::new();
+        let mut arrays = Vec::new();
+        for i in 0..num_cols {
+            let field = Field::new(&format!("col_{}", i), DataType::Decimal128(38, 10), true);
+            let array = Decimal128Array::from(vec![num_cols as i128; num_rows]);
+            fields.push(field);
+            arrays.push(Arc::new(array) as Arc<dyn Array>);
+        }
+        let schema = Schema::new(fields);
+        let batch = RecordBatch::try_new(Arc::new(schema), arrays).unwrap();
+
+        let mut writer = FileWriter::try_new_with_options(
+            Vec::new(),
+            batch.schema_ref(),
+            IpcWriteOptions::try_new(IPC_ALIGNMENT, false, MetadataVersion::V5).unwrap(),
+        )
+        .unwrap();
+        writer.write(&batch).unwrap();
+        writer.finish().unwrap();
+
+        let out: Vec<u8> = writer.into_inner().unwrap();
+
+        let buffer = Buffer::from_vec(out);
+        let trailer_start = buffer.len() - 10;
+        let footer_len = read_footer_length(buffer[trailer_start..].try_into().unwrap()).unwrap();
+        let footer = root_as_footer(&buffer[trailer_start - footer_len..trailer_start]).unwrap();
+
+        let schema = fb_to_schema(footer.schema().unwrap());
+
+        // Importantly we set `require_alignment`, otherwise the error later is suppressed due to copying
+        // to an aligned buffer in `ArrayDataBuilder.build_aligned`.
+        let decoder =
+            FileDecoder::new(Arc::new(schema), footer.version()).with_require_alignment(true);
+
+        let batches = footer.recordBatches().unwrap();
+
+        let block = batches.get(0);
+        let block_len = block.bodyLength() as usize + block.metaDataLength() as usize;
+        let data = buffer.slice_with_length(block.offset() as _, block_len);
+
+        let result = decoder.read_record_batch(block, &data);
+
+        let error = result.unwrap_err();
+        assert_eq!(
+            error.to_string(),
+            "Invalid argument error: Misaligned buffers[0] in array of type Decimal128(38, 10), \
+             offset from expected alignment of 16 by 8"
+        );
+    }
 }
diff --git a/arrow-schema/src/fields.rs b/arrow-schema/src/fields.rs
index 59b7e76c7..5a1a6c84c 100644
--- a/arrow-schema/src/fields.rs
+++ b/arrow-schema/src/fields.rs
@@ -137,10 +137,22 @@ impl Fields {
     /// assert_eq!(filtered, expected);
     /// ```
     pub fn filter_leaves<F: FnMut(usize, &FieldRef) -> bool>(&self, mut filter: F) -> Self {
-        fn filter_field<F: FnMut(&FieldRef) -> bool>(
+        self.try_filter_leaves(|idx, field| Ok(filter(idx, field)))
+            .unwrap()
+    }
+
+    /// Returns a copy of this [`Fields`] containing only those [`FieldRef`] passing a predicate
+    /// or an error if the predicate fails.
+    ///
+    /// See [`Fields::filter_leaves`] for more information.
+    pub fn try_filter_leaves<F: FnMut(usize, &FieldRef) -> Result<bool, ArrowError>>(
+        &self,
+        mut filter: F,
+    ) -> Result<Self, ArrowError> {
+        fn filter_field<F: FnMut(&FieldRef) -> Result<bool, ArrowError>>(
             f: &FieldRef,
             filter: &mut F,
-        ) -> Option<FieldRef> {
+        ) -> Result<Option<FieldRef>, ArrowError> {
             use DataType::*;
 
             let v = match f.data_type() {
@@ -149,35 +161,72 @@ impl Fields {
                 d => d,
             };
             let d = match v {
-                List(child) => List(filter_field(child, filter)?),
-                LargeList(child) => LargeList(filter_field(child, filter)?),
-                Map(child, ordered) => Map(filter_field(child, filter)?, *ordered),
-                FixedSizeList(child, size) => FixedSizeList(filter_field(child, filter)?, *size),
+                List(child) => {
+                    let fields = filter_field(child, filter)?;
+                    if let Some(fields) = fields {
+                        List(fields)
+                    } else {
+                        return Ok(None);
+                    }
+                }
+                LargeList(child) => {
+                    let fields = filter_field(child, filter)?;
+                    if let Some(fields) = fields {
+                        LargeList(fields)
+                    } else {
+                        return Ok(None);
+                    }
+                }
+                Map(child, ordered) => {
+                    let fields = filter_field(child, filter)?;
+                    if let Some(fields) = fields {
+                        Map(fields, *ordered)
+                    } else {
+                        return Ok(None);
+                    }
+                }
+                FixedSizeList(child, size) => {
+                    let fields = filter_field(child, filter)?;
+                    if let Some(fields) = fields {
+                        FixedSizeList(fields, *size)
+                    } else {
+                        return Ok(None);
+                    }
+                }
                 Struct(fields) => {
-                    let filtered: Fields = fields
+                    let filtered: Result<Vec<_>, _> =
+                        fields.iter().map(|f| filter_field(f, filter)).collect();
+                    let filtered: Fields = filtered?
                         .iter()
-                        .filter_map(|f| filter_field(f, filter))
+                        .filter_map(|f| f.as_ref().cloned())
                         .collect();
 
                     if filtered.is_empty() {
-                        return None;
+                        return Ok(None);
                     }
 
                     Struct(filtered)
                 }
                 Union(fields, mode) => {
-                    let filtered: UnionFields = fields
+                    let filtered: Result<Vec<_>, _> = fields
+                        .iter()
+                        .map(|(id, f)| filter_field(f, filter).map(|f| f.map(|f| (id, f))))
+                        .collect();
+                    let filtered: UnionFields = filtered?
                         .iter()
-                        .filter_map(|(id, f)| Some((id, filter_field(f, filter)?)))
+                        .filter_map(|f| f.as_ref().cloned())
                         .collect();
 
                     if filtered.is_empty() {
-                        return None;
+                        return Ok(None);
                     }
 
                     Union(filtered, *mode)
                 }
-                _ => return filter(f).then(|| f.clone()),
+                _ => {
+                    let filtered = filter(f)?;
+                    return Ok(filtered.then(|| f.clone()));
+                }
             };
             let d = match f.data_type() {
                 Dictionary(k, _) => Dictionary(k.clone(), Box::new(d)),
@@ -186,20 +235,26 @@ impl Fields {
                 }
                 _ => d,
             };
-            Some(Arc::new(f.as_ref().clone().with_data_type(d)))
+            Ok(Some(Arc::new(f.as_ref().clone().with_data_type(d))))
         }
 
         let mut leaf_idx = 0;
         let mut filter = |f: &FieldRef| {
-            let t = filter(leaf_idx, f);
+            let t = filter(leaf_idx, f)?;
             leaf_idx += 1;
-            t
+            Ok(t)
         };
 
-        self.0
+        let filtered: Result<Vec<_>, _> = self
+            .0
             .iter()
-            .filter_map(|f| filter_field(f, &mut filter))
-            .collect()
+            .map(|f| filter_field(f, &mut filter))
+            .collect();
+        let filtered = filtered?
+            .iter()
+            .filter_map(|f| f.as_ref().cloned())
+            .collect();
+        Ok(filtered)
     }
 
     /// Remove a field by index and return it.
@@ -531,5 +586,9 @@ mod tests {
         let r = fields.filter_leaves(|idx, _| idx == 14 || idx == 15);
         assert_eq!(r.len(), 1);
         assert_eq!(r[0], fields[9]);
+
+        // Propagate error
+        let r = fields.try_filter_leaves(|_, _| Err(ArrowError::SchemaError("error".to_string())));
+        assert!(r.is_err());
     }
 }
diff --git a/arrow-select/src/filter.rs b/arrow-select/src/filter.rs
index c365d0b84..2af19ff85 100644
--- a/arrow-select/src/filter.rs
+++ b/arrow-select/src/filter.rs
@@ -17,13 +17,16 @@
 
 //! Defines filter kernels
 
+use std::ops::AddAssign;
 use std::sync::Arc;
 
 use arrow_array::builder::BooleanBufferBuilder;
 use arrow_array::cast::AsArray;
-use arrow_array::types::{ArrowDictionaryKeyType, ByteArrayType};
+use arrow_array::types::{
+    ArrowDictionaryKeyType, ArrowPrimitiveType, ByteArrayType, RunEndIndexType,
+};
 use arrow_array::*;
-use arrow_buffer::{bit_util, BooleanBuffer, NullBuffer};
+use arrow_buffer::{bit_util, BooleanBuffer, NullBuffer, RunEndBuffer};
 use arrow_buffer::{Buffer, MutableBuffer};
 use arrow_data::bit_iterator::{BitIndexIterator, BitSliceIterator};
 use arrow_data::transform::MutableArrayData;
@@ -336,6 +339,12 @@ fn filter_array(values: &dyn Array, predicate: &FilterPredicate) -> Result<Array
             DataType::LargeBinary => {
                 Ok(Arc::new(filter_bytes(values.as_binary::<i64>(), predicate)))
             }
+            DataType::RunEndEncoded(_, _) => {
+                downcast_run_array!{
+                    values => Ok(Arc::new(filter_run_end_array(values, predicate)?)),
+                    t => unimplemented!("Filter not supported for RunEndEncoded type {:?}", t)
+                }
+            }
             DataType::Dictionary(_, _) => downcast_dictionary_array! {
                 values => Ok(Arc::new(filter_dict(values, predicate))),
                 t => unimplemented!("Filter not supported for dictionary type {:?}", t)
@@ -368,6 +377,55 @@ fn filter_array(values: &dyn Array, predicate: &FilterPredicate) -> Result<Array
     }
 }
 
+/// Filter any supported [`RunArray`] based on a [`FilterPredicate`]
+fn filter_run_end_array<R: RunEndIndexType>(
+    re_arr: &RunArray<R>,
+    pred: &FilterPredicate,
+) -> Result<RunArray<R>, ArrowError>
+where
+    R::Native: Into<i64> + From<bool>,
+    R::Native: AddAssign,
+{
+    let run_ends: &RunEndBuffer<R::Native> = re_arr.run_ends();
+    let mut values_filter = BooleanBufferBuilder::new(run_ends.len());
+    let mut new_run_ends = vec![R::default_value(); run_ends.len()];
+
+    let mut start = 0i64;
+    let mut i = 0;
+    let filter_values = pred.filter.values();
+    let mut count = R::default_value();
+
+    for end in run_ends.inner().into_iter().map(|i| (*i).into()) {
+        let mut keep = false;
+        // in filter_array the predicate array is checked to have the same len as the run end array
+        // this means the largest value in the run_ends is == to pred.len()
+        // so we're always within bounds when calling value_unchecked
+        for pred in (start..end).map(|i| unsafe { filter_values.value_unchecked(i as usize) }) {
+            count += R::Native::from(pred);
+            keep |= pred
+        }
+        // this is to avoid branching
+        new_run_ends[i] = count;
+        i += keep as usize;
+
+        values_filter.append(keep);
+        start = end;
+    }
+
+    new_run_ends.truncate(i);
+
+    if values_filter.is_empty() {
+        new_run_ends.clear();
+    }
+
+    let values = re_arr.values();
+    let pred = BooleanArray::new(values_filter.finish(), None);
+    let values = filter(&values, &pred)?;
+
+    let run_ends = PrimitiveArray::<R>::new(new_run_ends.into(), None);
+    RunArray::try_new(&run_ends, &values)
+}
+
 /// Computes a new null mask for `data` based on `predicate`
 ///
 /// If the predicate selected no null-rows, returns `None`, otherwise returns
@@ -635,6 +693,7 @@ where
 #[cfg(test)]
 mod tests {
     use arrow_array::builder::*;
+    use arrow_array::cast::as_run_array;
     use arrow_array::types::*;
     use rand::distributions::{Alphanumeric, Standard};
     use rand::prelude::*;
@@ -844,6 +903,78 @@ mod tests {
         assert_eq!(9, d.value(1));
     }
 
+    #[test]
+    fn test_filter_run_end_encoding_array() {
+        let run_ends = Int64Array::from(vec![2, 3, 8]);
+        let values = Int64Array::from(vec![7, -2, 9]);
+        let a = RunArray::try_new(&run_ends, &values).expect("Failed to create RunArray");
+        let b = BooleanArray::from(vec![true, false, true, false, true, false, true, false]);
+        let c = filter(&a, &b).unwrap();
+        let actual: &RunArray<Int64Type> = as_run_array(&c);
+        assert_eq!(4, actual.len());
+
+        let expected = RunArray::try_new(
+            &Int64Array::from(vec![1, 2, 4]),
+            &Int64Array::from(vec![7, -2, 9]),
+        )
+        .expect("Failed to make expected RunArray test is broken");
+
+        assert_eq!(&actual.run_ends().values(), &expected.run_ends().values());
+        assert_eq!(actual.values(), expected.values())
+    }
+
+    #[test]
+    fn test_filter_run_end_encoding_array_remove_value() {
+        let run_ends = Int32Array::from(vec![2, 3, 8, 10]);
+        let values = Int32Array::from(vec![7, -2, 9, -8]);
+        let a = RunArray::try_new(&run_ends, &values).expect("Failed to create RunArray");
+        let b = BooleanArray::from(vec![
+            false, true, false, false, true, false, true, false, false, false,
+        ]);
+        let c = filter(&a, &b).unwrap();
+        let actual: &RunArray<Int32Type> = as_run_array(&c);
+        assert_eq!(3, actual.len());
+
+        let expected =
+            RunArray::try_new(&Int32Array::from(vec![1, 3]), &Int32Array::from(vec![7, 9]))
+                .expect("Failed to make expected RunArray test is broken");
+
+        assert_eq!(&actual.run_ends().values(), &expected.run_ends().values());
+        assert_eq!(actual.values(), expected.values())
+    }
+
+    #[test]
+    fn test_filter_run_end_encoding_array_remove_all_but_one() {
+        let run_ends = Int16Array::from(vec![2, 3, 8, 10]);
+        let values = Int16Array::from(vec![7, -2, 9, -8]);
+        let a = RunArray::try_new(&run_ends, &values).expect("Failed to create RunArray");
+        let b = BooleanArray::from(vec![
+            false, false, false, false, false, false, true, false, false, false,
+        ]);
+        let c = filter(&a, &b).unwrap();
+        let actual: &RunArray<Int16Type> = as_run_array(&c);
+        assert_eq!(1, actual.len());
+
+        let expected = RunArray::try_new(&Int16Array::from(vec![1]), &Int16Array::from(vec![9]))
+            .expect("Failed to make expected RunArray test is broken");
+
+        assert_eq!(&actual.run_ends().values(), &expected.run_ends().values());
+        assert_eq!(actual.values(), expected.values())
+    }
+
+    #[test]
+    fn test_filter_run_end_encoding_array_empty() {
+        let run_ends = Int64Array::from(vec![2, 3, 8, 10]);
+        let values = Int64Array::from(vec![7, -2, 9, -8]);
+        let a = RunArray::try_new(&run_ends, &values).expect("Failed to create RunArray");
+        let b = BooleanArray::from(vec![
+            false, false, false, false, false, false, false, false, false, false,
+        ]);
+        let c = filter(&a, &b).unwrap();
+        let actual: &RunArray<Int64Type> = as_run_array(&c);
+        assert_eq!(0, actual.len());
+    }
+
     #[test]
     fn test_filter_dictionary_array() {
         let values = [Some("hello"), None, Some("world"), Some("!")];
diff --git a/arrow/Cargo.toml b/arrow/Cargo.toml
index 4f7fda9b8..a938d75b1 100644
--- a/arrow/Cargo.toml
+++ b/arrow/Cargo.toml
@@ -54,7 +54,7 @@ arrow-select = { workspace = true }
 arrow-string = { workspace = true }
 
 rand = { version = "0.8", default-features = false, features = ["std", "std_rng"], optional = true }
-pyo3 = { version = "0.20", default-features = false, optional = true }
+pyo3 = { version = "0.21.1", default-features = false, optional = true }
 
 [package.metadata.docs.rs]
 features = ["prettyprint", "ipc_compression", "ffi", "pyarrow"]
diff --git a/arrow/src/pyarrow.rs b/arrow/src/pyarrow.rs
index 39702ce01..1733067c7 100644
--- a/arrow/src/pyarrow.rs
+++ b/arrow/src/pyarrow.rs
@@ -64,6 +64,7 @@ use pyo3::exceptions::{PyTypeError, PyValueError};
 use pyo3::ffi::Py_uintptr_t;
 use pyo3::import_exception;
 use pyo3::prelude::*;
+use pyo3::pybacked::PyBackedStr;
 use pyo3::types::{PyCapsule, PyList, PyTuple};
 
 use crate::array::{make_array, ArrayData};
@@ -82,7 +83,12 @@ fn to_py_err(err: ArrowError) -> PyErr {
 }
 
 pub trait FromPyArrow: Sized {
-    fn from_pyarrow(value: &PyAny) -> PyResult<Self>;
+    #[deprecated(since = "52.0.0", note = "Use from_pyarrow_bound")]
+    fn from_pyarrow(value: &PyAny) -> PyResult<Self> {
+        Self::from_pyarrow_bound(&value.as_borrowed())
+    }
+
+    fn from_pyarrow_bound(value: &Bound<PyAny>) -> PyResult<Self>;
 }
 
 /// Create a new PyArrow object from a arrow-rs type.
@@ -101,15 +107,17 @@ impl<T: ToPyArrow> IntoPyArrow for T {
     }
 }
 
-fn validate_class(expected: &str, value: &PyAny) -> PyResult<()> {
-    let pyarrow = PyModule::import(value.py(), "pyarrow")?;
+fn validate_class(expected: &str, value: &Bound<PyAny>) -> PyResult<()> {
+    let pyarrow = PyModule::import_bound(value.py(), "pyarrow")?;
     let class = pyarrow.getattr(expected)?;
-    if !value.is_instance(class)? {
-        let expected_module = class.getattr("__module__")?.extract::<&str>()?;
-        let expected_name = class.getattr("__name__")?.extract::<&str>()?;
+    if !value.is_instance(&class)? {
+        let expected_module = class.getattr("__module__")?.extract::<PyBackedStr>()?;
+        let expected_name = class.getattr("__name__")?.extract::<PyBackedStr>()?;
         let found_class = value.get_type();
-        let found_module = found_class.getattr("__module__")?.extract::<&str>()?;
-        let found_name = found_class.getattr("__name__")?.extract::<&str>()?;
+        let found_module = found_class
+            .getattr("__module__")?
+            .extract::<PyBackedStr>()?;
+        let found_name = found_class.getattr("__name__")?.extract::<PyBackedStr>()?;
         return Err(PyTypeError::new_err(format!(
             "Expected instance of {}.{}, got {}.{}",
             expected_module, expected_name, found_module, found_name
@@ -118,7 +126,7 @@ fn validate_class(expected: &str, value: &PyAny) -> PyResult<()> {
     Ok(())
 }
 
-fn validate_pycapsule(capsule: &PyCapsule, name: &str) -> PyResult<()> {
+fn validate_pycapsule(capsule: &Bound<PyCapsule>, name: &str) -> PyResult<()> {
     let capsule_name = capsule.name()?;
     if capsule_name.is_none() {
         return Err(PyValueError::new_err(
@@ -138,13 +146,13 @@ fn validate_pycapsule(capsule: &PyCapsule, name: &str) -> PyResult<()> {
 }
 
 impl FromPyArrow for DataType {
-    fn from_pyarrow(value: &PyAny) -> PyResult<Self> {
+    fn from_pyarrow_bound(value: &Bound<PyAny>) -> PyResult<Self> {
         // Newer versions of PyArrow as well as other libraries with Arrow data implement this
         // method, so prefer it over _export_to_c.
         // See https://arrow.apache.org/docs/format/CDataInterface/PyCapsuleInterface.html
         if value.hasattr("__arrow_c_schema__")? {
-            let capsule: &PyCapsule =
-                PyTryInto::try_into(value.getattr("__arrow_c_schema__")?.call0()?)?;
+            let capsule = value.getattr("__arrow_c_schema__")?.call0()?;
+            let capsule = capsule.downcast::<PyCapsule>()?;
             validate_pycapsule(capsule, "arrow_schema")?;
 
             let schema_ptr = unsafe { capsule.reference::<FFI_ArrowSchema>() };
@@ -166,7 +174,7 @@ impl ToPyArrow for DataType {
     fn to_pyarrow(&self, py: Python) -> PyResult<PyObject> {
         let c_schema = FFI_ArrowSchema::try_from(self).map_err(to_py_err)?;
         let c_schema_ptr = &c_schema as *const FFI_ArrowSchema;
-        let module = py.import("pyarrow")?;
+        let module = py.import_bound("pyarrow")?;
         let class = module.getattr("DataType")?;
         let dtype = class.call_method1("_import_from_c", (c_schema_ptr as Py_uintptr_t,))?;
         Ok(dtype.into())
@@ -174,13 +182,13 @@ impl ToPyArrow for DataType {
 }
 
 impl FromPyArrow for Field {
-    fn from_pyarrow(value: &PyAny) -> PyResult<Self> {
+    fn from_pyarrow_bound(value: &Bound<PyAny>) -> PyResult<Self> {
         // Newer versions of PyArrow as well as other libraries with Arrow data implement this
         // method, so prefer it over _export_to_c.
         // See https://arrow.apache.org/docs/format/CDataInterface/PyCapsuleInterface.html
         if value.hasattr("__arrow_c_schema__")? {
-            let capsule: &PyCapsule =
-                PyTryInto::try_into(value.getattr("__arrow_c_schema__")?.call0()?)?;
+            let capsule = value.getattr("__arrow_c_schema__")?.call0()?;
+            let capsule = capsule.downcast::<PyCapsule>()?;
             validate_pycapsule(capsule, "arrow_schema")?;
 
             let schema_ptr = unsafe { capsule.reference::<FFI_ArrowSchema>() };
@@ -202,7 +210,7 @@ impl ToPyArrow for Field {
     fn to_pyarrow(&self, py: Python) -> PyResult<PyObject> {
         let c_schema = FFI_ArrowSchema::try_from(self).map_err(to_py_err)?;
         let c_schema_ptr = &c_schema as *const FFI_ArrowSchema;
-        let module = py.import("pyarrow")?;
+        let module = py.import_bound("pyarrow")?;
         let class = module.getattr("Field")?;
         let dtype = class.call_method1("_import_from_c", (c_schema_ptr as Py_uintptr_t,))?;
         Ok(dtype.into())
@@ -210,13 +218,13 @@ impl ToPyArrow for Field {
 }
 
 impl FromPyArrow for Schema {
-    fn from_pyarrow(value: &PyAny) -> PyResult<Self> {
+    fn from_pyarrow_bound(value: &Bound<PyAny>) -> PyResult<Self> {
         // Newer versions of PyArrow as well as other libraries with Arrow data implement this
         // method, so prefer it over _export_to_c.
         // See https://arrow.apache.org/docs/format/CDataInterface/PyCapsuleInterface.html
         if value.hasattr("__arrow_c_schema__")? {
-            let capsule: &PyCapsule =
-                PyTryInto::try_into(value.getattr("__arrow_c_schema__")?.call0()?)?;
+            let capsule = value.getattr("__arrow_c_schema__")?.call0()?;
+            let capsule = capsule.downcast::<PyCapsule>()?;
             validate_pycapsule(capsule, "arrow_schema")?;
 
             let schema_ptr = unsafe { capsule.reference::<FFI_ArrowSchema>() };
@@ -238,7 +246,7 @@ impl ToPyArrow for Schema {
     fn to_pyarrow(&self, py: Python) -> PyResult<PyObject> {
         let c_schema = FFI_ArrowSchema::try_from(self).map_err(to_py_err)?;
         let c_schema_ptr = &c_schema as *const FFI_ArrowSchema;
-        let module = py.import("pyarrow")?;
+        let module = py.import_bound("pyarrow")?;
         let class = module.getattr("Schema")?;
         let schema = class.call_method1("_import_from_c", (c_schema_ptr as Py_uintptr_t,))?;
         Ok(schema.into())
@@ -246,7 +254,7 @@ impl ToPyArrow for Schema {
 }
 
 impl FromPyArrow for ArrayData {
-    fn from_pyarrow(value: &PyAny) -> PyResult<Self> {
+    fn from_pyarrow_bound(value: &Bound<PyAny>) -> PyResult<Self> {
         // Newer versions of PyArrow as well as other libraries with Arrow data implement this
         // method, so prefer it over _export_to_c.
         // See https://arrow.apache.org/docs/format/CDataInterface/PyCapsuleInterface.html
@@ -259,8 +267,10 @@ impl FromPyArrow for ArrayData {
                 ));
             }
 
-            let schema_capsule: &PyCapsule = PyTryInto::try_into(tuple.get_item(0)?)?;
-            let array_capsule: &PyCapsule = PyTryInto::try_into(tuple.get_item(1)?)?;
+            let schema_capsule = tuple.get_item(0)?;
+            let schema_capsule = schema_capsule.downcast::<PyCapsule>()?;
+            let array_capsule = tuple.get_item(1)?;
+            let array_capsule = array_capsule.downcast::<PyCapsule>()?;
 
             validate_pycapsule(schema_capsule, "arrow_schema")?;
             validate_pycapsule(array_capsule, "arrow_array")?;
@@ -296,7 +306,7 @@ impl ToPyArrow for ArrayData {
         let array = FFI_ArrowArray::new(self);
         let schema = FFI_ArrowSchema::try_from(self.data_type()).map_err(to_py_err)?;
 
-        let module = py.import("pyarrow")?;
+        let module = py.import_bound("pyarrow")?;
         let class = module.getattr("Array")?;
         let array = class.call_method1(
             "_import_from_c",
@@ -310,9 +320,9 @@ impl ToPyArrow for ArrayData {
 }
 
 impl<T: FromPyArrow> FromPyArrow for Vec<T> {
-    fn from_pyarrow(value: &PyAny) -> PyResult<Self> {
+    fn from_pyarrow_bound(value: &Bound<PyAny>) -> PyResult<Self> {
         let list = value.downcast::<PyList>()?;
-        list.iter().map(|x| T::from_pyarrow(x)).collect()
+        list.iter().map(|x| T::from_pyarrow_bound(&x)).collect()
     }
 }
 
@@ -327,7 +337,7 @@ impl<T: ToPyArrow> ToPyArrow for Vec<T> {
 }
 
 impl FromPyArrow for RecordBatch {
-    fn from_pyarrow(value: &PyAny) -> PyResult<Self> {
+    fn from_pyarrow_bound(value: &Bound<PyAny>) -> PyResult<Self> {
         // Newer versions of PyArrow as well as other libraries with Arrow data implement this
         // method, so prefer it over _export_to_c.
         // See https://arrow.apache.org/docs/format/CDataInterface/PyCapsuleInterface.html
@@ -340,8 +350,10 @@ impl FromPyArrow for RecordBatch {
                 ));
             }
 
-            let schema_capsule: &PyCapsule = PyTryInto::try_into(tuple.get_item(0)?)?;
-            let array_capsule: &PyCapsule = PyTryInto::try_into(tuple.get_item(1)?)?;
+            let schema_capsule = tuple.get_item(0)?;
+            let schema_capsule = schema_capsule.downcast::<PyCapsule>()?;
+            let array_capsule = tuple.get_item(1)?;
+            let array_capsule = array_capsule.downcast::<PyCapsule>()?;
 
             validate_pycapsule(schema_capsule, "arrow_schema")?;
             validate_pycapsule(array_capsule, "arrow_array")?;
@@ -370,12 +382,13 @@ impl FromPyArrow for RecordBatch {
         validate_class("RecordBatch", value)?;
         // TODO(kszucs): implement the FFI conversions in arrow-rs for RecordBatches
         let schema = value.getattr("schema")?;
-        let schema = Arc::new(Schema::from_pyarrow(schema)?);
+        let schema = Arc::new(Schema::from_pyarrow_bound(&schema)?);
 
-        let arrays = value.getattr("columns")?.downcast::<PyList>()?;
+        let arrays = value.getattr("columns")?;
         let arrays = arrays
+            .downcast::<PyList>()?
             .iter()
-            .map(|a| Ok(make_array(ArrayData::from_pyarrow(a)?)))
+            .map(|a| Ok(make_array(ArrayData::from_pyarrow_bound(&a)?)))
             .collect::<PyResult<_>>()?;
 
         let batch = RecordBatch::try_new(schema, arrays).map_err(to_py_err)?;
@@ -395,13 +408,13 @@ impl ToPyArrow for RecordBatch {
 
 /// Supports conversion from `pyarrow.RecordBatchReader` to [ArrowArrayStreamReader].
 impl FromPyArrow for ArrowArrayStreamReader {
-    fn from_pyarrow(value: &PyAny) -> PyResult<Self> {
+    fn from_pyarrow_bound(value: &Bound<PyAny>) -> PyResult<Self> {
         // Newer versions of PyArrow as well as other libraries with Arrow data implement this
         // method, so prefer it over _export_to_c.
         // See https://arrow.apache.org/docs/format/CDataInterface/PyCapsuleInterface.html
         if value.hasattr("__arrow_c_stream__")? {
-            let capsule: &PyCapsule =
-                PyTryInto::try_into(value.getattr("__arrow_c_stream__")?.call0()?)?;
+            let capsule = value.getattr("__arrow_c_stream__")?.call0()?;
+            let capsule = capsule.downcast::<PyCapsule>()?;
             validate_pycapsule(capsule, "arrow_array_stream")?;
 
             let stream = unsafe { FFI_ArrowArrayStream::from_raw(capsule.pointer() as _) };
@@ -421,7 +434,7 @@ impl FromPyArrow for ArrowArrayStreamReader {
         // make the conversion through PyArrow's private API
         // this changes the pointer's memory and is thus unsafe.
         // In particular, `_export_to_c` can go out of bounds
-        let args = PyTuple::new(value.py(), [stream_ptr as Py_uintptr_t]);
+        let args = PyTuple::new_bound(value.py(), [stream_ptr as Py_uintptr_t]);
         value.call_method1("_export_to_c", args)?;
 
         let stream_reader = ArrowArrayStreamReader::try_new(stream)
@@ -439,9 +452,9 @@ impl IntoPyArrow for Box<dyn RecordBatchReader + Send> {
         let mut stream = FFI_ArrowArrayStream::new(self);
 
         let stream_ptr = (&mut stream) as *mut FFI_ArrowArrayStream;
-        let module = py.import("pyarrow")?;
+        let module = py.import_bound("pyarrow")?;
         let class = module.getattr("RecordBatchReader")?;
-        let args = PyTuple::new(py, [stream_ptr as Py_uintptr_t]);
+        let args = PyTuple::new_bound(py, [stream_ptr as Py_uintptr_t]);
         let reader = class.call_method1("_import_from_c", args)?;
 
         Ok(PyObject::from(reader))
@@ -463,8 +476,8 @@ impl IntoPyArrow for ArrowArrayStreamReader {
 pub struct PyArrowType<T>(pub T);
 
 impl<'source, T: FromPyArrow> FromPyObject<'source> for PyArrowType<T> {
-    fn extract(value: &'source PyAny) -> PyResult<Self> {
-        Ok(Self(T::from_pyarrow(value)?))
+    fn extract_bound(value: &Bound<'source, PyAny>) -> PyResult<Self> {
+        Ok(Self(T::from_pyarrow_bound(value)?))
     }
 }
 
diff --git a/object_store/Cargo.toml b/object_store/Cargo.toml
index 79813a0ea..d0c3af240 100644
--- a/object_store/Cargo.toml
+++ b/object_store/Cargo.toml
@@ -32,7 +32,7 @@ all-features = true
 [dependencies] # In alphabetical order
 async-trait = "0.1.53"
 bytes = "1.0"
-chrono = { version = "0.4.31", default-features = false, features = ["clock"] }
+chrono = { version = "0.4.34", default-features = false, features = ["clock"] }
 futures = "0.3"
 humantime = "2.1"
 itertools = "0.12.0"
diff --git a/object_store/src/aws/credential.rs b/object_store/src/aws/credential.rs
index dd7fa5b41..478e56dd0 100644
--- a/object_store/src/aws/credential.rs
+++ b/object_store/src/aws/credential.rs
@@ -19,7 +19,7 @@ use crate::aws::{AwsCredentialProvider, STORE, STRICT_ENCODE_SET, STRICT_PATH_EN
 use crate::client::retry::RetryExt;
 use crate::client::token::{TemporaryToken, TokenCache};
 use crate::client::TokenProvider;
-use crate::util::hmac_sha256;
+use crate::util::{hex_digest, hex_encode, hmac_sha256};
 use crate::{CredentialProvider, Result, RetryConfig};
 use async_trait::async_trait;
 use bytes::Buf;
@@ -342,23 +342,6 @@ impl CredentialExt for RequestBuilder {
     }
 }
 
-/// Computes the SHA256 digest of `body` returned as a hex encoded string
-fn hex_digest(bytes: &[u8]) -> String {
-    let digest = ring::digest::digest(&ring::digest::SHA256, bytes);
-    hex_encode(digest.as_ref())
-}
-
-/// Returns `bytes` as a lower-case hex encoded string
-fn hex_encode(bytes: &[u8]) -> String {
-    use std::fmt::Write;
-    let mut out = String::with_capacity(bytes.len() * 2);
-    for byte in bytes {
-        // String writing is infallible
-        let _ = write!(out, "{byte:02x}");
-    }
-    out
-}
-
 /// Canonicalizes query parameters into the AWS canonical form
 ///
 /// <https://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html>
diff --git a/object_store/src/aws/mod.rs b/object_store/src/aws/mod.rs
index b33771de9..76d01d597 100644
--- a/object_store/src/aws/mod.rs
+++ b/object_store/src/aws/mod.rs
@@ -43,6 +43,7 @@ use crate::client::list::ListClientExt;
 use crate::client::CredentialProvider;
 use crate::multipart::{MultipartStore, PartId};
 use crate::signer::Signer;
+use crate::util::STRICT_ENCODE_SET;
 use crate::{
     Error, GetOptions, GetResult, ListResult, MultipartId, MultipartUpload, ObjectMeta,
     ObjectStore, Path, PutMode, PutOptions, PutResult, Result, UploadPart,
@@ -64,16 +65,6 @@ pub use dynamo::DynamoCommit;
 pub use precondition::{S3ConditionalPut, S3CopyIfNotExists};
 pub use resolve::resolve_bucket_region;
 
-// http://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html
-//
-// Do not URI-encode any of the unreserved characters that RFC 3986 defines:
-// A-Z, a-z, 0-9, hyphen ( - ), underscore ( _ ), period ( . ), and tilde ( ~ ).
-pub(crate) const STRICT_ENCODE_SET: percent_encoding::AsciiSet = percent_encoding::NON_ALPHANUMERIC
-    .remove(b'-')
-    .remove(b'.')
-    .remove(b'_')
-    .remove(b'~');
-
 /// This struct is used to maintain the URI path encoding
 const STRICT_PATH_ENCODE_SET: percent_encoding::AsciiSet = STRICT_ENCODE_SET.remove(b'/');
 
diff --git a/object_store/src/gcp/builder.rs b/object_store/src/gcp/builder.rs
index 2cf75040b..4fa91677a 100644
--- a/object_store/src/gcp/builder.rs
+++ b/object_store/src/gcp/builder.rs
@@ -21,7 +21,10 @@ use crate::gcp::credential::{
     ApplicationDefaultCredentials, InstanceCredentialProvider, ServiceAccountCredentials,
     DEFAULT_GCS_BASE_URL,
 };
-use crate::gcp::{credential, GcpCredential, GcpCredentialProvider, GoogleCloudStorage, STORE};
+use crate::gcp::{
+    credential, GcpCredential, GcpCredentialProvider, GcpSigningCredential,
+    GcpSigningCredentialProvider, GoogleCloudStorage, STORE,
+};
 use crate::{ClientConfigKey, ClientOptions, Result, RetryConfig, StaticCredentialProvider};
 use serde::{Deserialize, Serialize};
 use snafu::{OptionExt, ResultExt, Snafu};
@@ -29,6 +32,8 @@ use std::str::FromStr;
 use std::sync::Arc;
 use url::Url;
 
+use super::credential::{AuthorizedUserSigningCredentials, InstanceSigningCredentialProvider};
+
 #[derive(Debug, Snafu)]
 enum Error {
     #[snafu(display("Missing bucket name"))]
@@ -107,6 +112,8 @@ pub struct GoogleCloudStorageBuilder {
     client_options: ClientOptions,
     /// Credentials
     credentials: Option<GcpCredentialProvider>,
+    /// Credentials for sign url
+    signing_cedentials: Option<GcpSigningCredentialProvider>,
 }
 
 /// Configuration keys for [`GoogleCloudStorageBuilder`]
@@ -202,6 +209,7 @@ impl Default for GoogleCloudStorageBuilder {
             client_options: ClientOptions::new().with_allow_http(true),
             url: None,
             credentials: None,
+            signing_cedentials: None,
         }
     }
 }
@@ -452,13 +460,13 @@ impl GoogleCloudStorageBuilder {
             Arc::new(StaticCredentialProvider::new(GcpCredential {
                 bearer: "".to_string(),
             })) as _
-        } else if let Some(credentials) = service_account_credentials {
+        } else if let Some(credentials) = service_account_credentials.clone() {
             Arc::new(TokenCredentialProvider::new(
                 credentials.token_provider()?,
                 self.client_options.client()?,
                 self.retry_config.clone(),
             )) as _
-        } else if let Some(credentials) = application_default_credentials {
+        } else if let Some(credentials) = application_default_credentials.clone() {
             match credentials {
                 ApplicationDefaultCredentials::AuthorizedUser(token) => {
                     Arc::new(TokenCredentialProvider::new(
@@ -483,13 +491,44 @@ impl GoogleCloudStorageBuilder {
             )) as _
         };
 
-        let config = GoogleCloudStorageConfig {
-            base_url: gcs_base_url,
+        let signing_credentials = if let Some(signing_credentials) = self.signing_cedentials {
+            signing_credentials
+        } else if disable_oauth {
+            Arc::new(StaticCredentialProvider::new(GcpSigningCredential {
+                email: "".to_string(),
+                private_key: None,
+            })) as _
+        } else if let Some(credentials) = service_account_credentials.clone() {
+            credentials.signing_credentials()?
+        } else if let Some(credentials) = application_default_credentials.clone() {
+            match credentials {
+                ApplicationDefaultCredentials::AuthorizedUser(token) => {
+                    Arc::new(TokenCredentialProvider::new(
+                        AuthorizedUserSigningCredentials::from(token)?,
+                        self.client_options.client()?,
+                        self.retry_config.clone(),
+                    )) as _
+                }
+                ApplicationDefaultCredentials::ServiceAccount(token) => {
+                    token.signing_credentials()?
+                }
+            }
+        } else {
+            Arc::new(TokenCredentialProvider::new(
+                InstanceSigningCredentialProvider::default(),
+                self.client_options.metadata_client()?,
+                self.retry_config.clone(),
+            )) as _
+        };
+
+        let config = GoogleCloudStorageConfig::new(
+            gcs_base_url,
             credentials,
+            signing_credentials,
             bucket_name,
-            retry_config: self.retry_config,
-            client_options: self.client_options,
-        };
+            self.retry_config,
+            self.client_options,
+        );
 
         Ok(GoogleCloudStorage {
             client: Arc::new(GoogleCloudStorageClient::new(config)?),
diff --git a/object_store/src/gcp/client.rs b/object_store/src/gcp/client.rs
index def53beef..901257f91 100644
--- a/object_store/src/gcp/client.rs
+++ b/object_store/src/gcp/client.rs
@@ -24,19 +24,22 @@ use crate::client::s3::{
     ListResponse,
 };
 use crate::client::GetOptionsExt;
-use crate::gcp::{GcpCredential, GcpCredentialProvider, STORE};
+use crate::gcp::{GcpCredential, GcpCredentialProvider, GcpSigningCredentialProvider, STORE};
 use crate::multipart::PartId;
 use crate::path::{Path, DELIMITER};
+use crate::util::hex_encode;
 use crate::{
     ClientOptions, GetOptions, ListResult, MultipartId, PutMode, PutOptions, PutResult, Result,
     RetryConfig,
 };
 use async_trait::async_trait;
+use base64::prelude::BASE64_STANDARD;
+use base64::Engine;
 use bytes::{Buf, Bytes};
 use percent_encoding::{percent_encode, utf8_percent_encode, NON_ALPHANUMERIC};
 use reqwest::header::HeaderName;
 use reqwest::{header, Client, Method, RequestBuilder, Response, StatusCode};
-use serde::Serialize;
+use serde::{Deserialize, Serialize};
 use snafu::{OptionExt, ResultExt, Snafu};
 use std::sync::Arc;
 
@@ -101,6 +104,15 @@ enum Error {
 
     #[snafu(display("Got invalid multipart response: {}", source))]
     InvalidMultipartResponse { source: quick_xml::de::DeError },
+
+    #[snafu(display("Error signing blob: {}", source))]
+    SignBlobRequest { source: crate::client::retry::Error },
+
+    #[snafu(display("Got invalid signing blob repsonse: {}", source))]
+    InvalidSignBlobResponse { source: reqwest::Error },
+
+    #[snafu(display("Got invalid signing blob signature: {}", source))]
+    InvalidSignBlobSignature { source: base64::DecodeError },
 }
 
 impl From<Error> for crate::Error {
@@ -123,6 +135,8 @@ pub struct GoogleCloudStorageConfig {
 
     pub credentials: GcpCredentialProvider,
 
+    pub signing_credentials: GcpSigningCredentialProvider,
+
     pub bucket_name: String,
 
     pub retry_config: RetryConfig,
@@ -130,6 +144,30 @@ pub struct GoogleCloudStorageConfig {
     pub client_options: ClientOptions,
 }
 
+impl GoogleCloudStorageConfig {
+    pub fn new(
+        base_url: String,
+        credentials: GcpCredentialProvider,
+        signing_credentials: GcpSigningCredentialProvider,
+        bucket_name: String,
+        retry_config: RetryConfig,
+        client_options: ClientOptions,
+    ) -> Self {
+        Self {
+            base_url,
+            credentials,
+            signing_credentials,
+            bucket_name,
+            retry_config,
+            client_options,
+        }
+    }
+
+    pub fn path_url(&self, path: &Path) -> String {
+        format!("{}/{}/{}", self.base_url, self.bucket_name, path)
+    }
+}
+
 /// A builder for a put request allowing customisation of the headers and query string
 pub struct PutRequest<'a> {
     path: &'a Path,
@@ -163,6 +201,21 @@ impl<'a> PutRequest<'a> {
     }
 }
 
+/// Sign Blob Request Body
+#[derive(Debug, Serialize)]
+struct SignBlobBody {
+    /// The payload to sign
+    payload: String,
+}
+
+/// Sign Blob Response
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct SignBlobResponse {
+    /// The signature for the payload
+    signed_blob: String,
+}
+
 #[derive(Debug)]
 pub struct GoogleCloudStorageClient {
     config: GoogleCloudStorageConfig,
@@ -197,6 +250,54 @@ impl GoogleCloudStorageClient {
         self.config.credentials.get_credential().await
     }
 
+    /// Create a signature from a string-to-sign using Google Cloud signBlob method.
+    /// form like:
+    /// ```plaintext
+    /// curl -X POST --data-binary @JSON_FILE_NAME \
+    /// -H "Authorization: Bearer OAUTH2_TOKEN" \
+    /// -H "Content-Type: application/json" \
+    /// "https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/SERVICE_ACCOUNT_EMAIL:signBlob"
+    /// ```
+    ///
+    /// 'JSON_FILE_NAME' is a file containing the following JSON object:
+    /// ```plaintext
+    /// {
+    ///  "payload": "REQUEST_INFORMATION"
+    /// }
+    /// ```
+    pub async fn sign_blob(&self, string_to_sign: &str, client_email: &str) -> Result<String> {
+        let credential = self.get_credential().await?;
+        let body = SignBlobBody {
+            payload: BASE64_STANDARD.encode(string_to_sign),
+        };
+
+        let url = format!(
+            "https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/{}:signBlob",
+            client_email
+        );
+
+        let response = self
+            .client
+            .post(&url)
+            .bearer_auth(&credential.bearer)
+            .json(&body)
+            .send_retry(&self.config.retry_config)
+            .await
+            .context(SignBlobRequestSnafu)?;
+
+        //If successful, the signature is returned in the signedBlob field in the response.
+        let response = response
+            .json::<SignBlobResponse>()
+            .await
+            .context(InvalidSignBlobResponseSnafu)?;
+
+        let signed_blob = BASE64_STANDARD
+            .decode(response.signed_blob)
+            .context(InvalidSignBlobSignatureSnafu)?;
+
+        Ok(hex_encode(&signed_blob))
+    }
+
     pub fn object_url(&self, path: &Path) -> String {
         let encoded = utf8_percent_encode(path.as_ref(), NON_ALPHANUMERIC);
         format!(
diff --git a/object_store/src/gcp/credential.rs b/object_store/src/gcp/credential.rs
index 34cd6eeb6..fcd516a1b 100644
--- a/object_store/src/gcp/credential.rs
+++ b/object_store/src/gcp/credential.rs
@@ -15,19 +15,26 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use super::client::GoogleCloudStorageClient;
 use crate::client::retry::RetryExt;
 use crate::client::token::TemporaryToken;
 use crate::client::TokenProvider;
-use crate::gcp::STORE;
-use crate::RetryConfig;
+use crate::gcp::{GcpSigningCredentialProvider, STORE};
+use crate::util::{hex_digest, hex_encode, STRICT_ENCODE_SET};
+use crate::{RetryConfig, StaticCredentialProvider};
 use async_trait::async_trait;
 use base64::prelude::BASE64_URL_SAFE_NO_PAD;
 use base64::Engine;
+use chrono::{DateTime, Utc};
 use futures::TryFutureExt;
+use hyper::HeaderMap;
+use itertools::Itertools;
+use percent_encoding::utf8_percent_encode;
 use reqwest::{Client, Method};
 use ring::signature::RsaKeyPair;
 use serde::Deserialize;
 use snafu::{ResultExt, Snafu};
+use std::collections::BTreeMap;
 use std::env;
 use std::fs::File;
 use std::io::BufReader;
@@ -35,11 +42,15 @@ use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 use tracing::info;
+use url::Url;
 
-pub const DEFAULT_SCOPE: &str = "https://www.googleapis.com/auth/devstorage.full_control";
+pub const DEFAULT_SCOPE: &str = "https://www.googleapis.com/auth/cloud-platform";
 
 pub const DEFAULT_GCS_BASE_URL: &str = "https://storage.googleapis.com";
 
+const DEFAULT_GCS_PLAYLOAD_STRING: &str = "UNSIGNED-PAYLOAD";
+const DEFAULT_GCS_SIGN_BLOB_HOST: &str = "storage.googleapis.com";
+
 #[derive(Debug, Snafu)]
 pub enum Error {
     #[snafu(display("Unable to open service account file from {}: {}", path.display(), source))]
@@ -57,7 +68,7 @@ pub enum Error {
     #[snafu(display("Invalid RSA key: {}", source), context(false))]
     InvalidKey { source: ring::error::KeyRejected },
 
-    #[snafu(display("Error signing jwt: {}", source))]
+    #[snafu(display("Error signing: {}", source))]
     Sign { source: ring::error::Unspecified },
 
     #[snafu(display("Error encoding jwt payload: {}", source))]
@@ -82,6 +93,69 @@ impl From<Error> for crate::Error {
     }
 }
 
+/// A Google Cloud Storage Credential for signing
+#[derive(Debug)]
+pub struct GcpSigningCredential {
+    /// The email of the service account
+    pub email: String,
+
+    /// An optional RSA private key
+    ///
+    /// If provided this will be used to sign the URL, otherwise a call will be made to
+    /// [`iam.serviceAccounts.signBlob`]. This allows supporting credential sources
+    /// that don't expose the service account private key, e.g. [IMDS].
+    ///
+    /// [IMDS]: https://cloud.google.com/docs/authentication/get-id-token#metadata-server
+    /// [`iam.serviceAccounts.signBlob`]: https://cloud.google.com/storage/docs/authentication/creating-signatures
+    pub private_key: Option<ServiceAccountKey>,
+}
+
+/// A private RSA key for a service account
+#[derive(Debug)]
+pub struct ServiceAccountKey(RsaKeyPair);
+
+impl ServiceAccountKey {
+    /// Parses a pem-encoded RSA key
+    pub fn from_pem(encoded: &[u8]) -> Result<Self> {
+        use rustls_pemfile::Item;
+        use std::io::Cursor;
+
+        let mut cursor = Cursor::new(encoded);
+        let mut reader = BufReader::new(&mut cursor);
+
+        // Reading from string is infallible
+        match rustls_pemfile::read_one(&mut reader).unwrap() {
+            Some(Item::Pkcs8Key(key)) => Self::from_pkcs8(key.secret_pkcs8_der()),
+            Some(Item::Pkcs1Key(key)) => Self::from_der(key.secret_pkcs1_der()),
+            _ => Err(Error::MissingKey),
+        }
+    }
+
+    /// Parses an unencrypted PKCS#8-encoded RSA private key.
+    pub fn from_pkcs8(key: &[u8]) -> Result<Self> {
+        Ok(Self(RsaKeyPair::from_pkcs8(key)?))
+    }
+
+    /// Parses an unencrypted PKCS#8-encoded RSA private key.
+    pub fn from_der(key: &[u8]) -> Result<Self> {
+        Ok(Self(RsaKeyPair::from_der(key)?))
+    }
+
+    fn sign(&self, string_to_sign: &str) -> Result<String> {
+        let mut signature = vec![0; self.0.public().modulus_len()];
+        self.0
+            .sign(
+                &ring::signature::RSA_PKCS1_SHA256,
+                &ring::rand::SystemRandom::new(),
+                string_to_sign.as_bytes(),
+                &mut signature,
+            )
+            .context(SignSnafu)?;
+
+        Ok(hex_encode(&signature))
+    }
+}
+
 /// A Google Cloud Storage Credential
 #[derive(Debug, Eq, PartialEq)]
 pub struct GcpCredential {
@@ -152,9 +226,8 @@ struct TokenResponse {
 pub struct SelfSignedJwt {
     issuer: String,
     scope: String,
-    key_pair: RsaKeyPair,
-    jwt_header: String,
-    random: ring::rand::SystemRandom,
+    private_key: ServiceAccountKey,
+    key_id: String,
 }
 
 impl SelfSignedJwt {
@@ -162,23 +235,14 @@ impl SelfSignedJwt {
     pub fn new(
         key_id: String,
         issuer: String,
-        private_key_pem: String,
+        private_key: ServiceAccountKey,
         scope: String,
     ) -> Result<Self> {
-        let key_pair = decode_first_rsa_key(private_key_pem)?;
-        let jwt_header = b64_encode_obj(&JwtHeader {
-            alg: "RS256",
-            typ: Some("JWT"),
-            kid: Some(&key_id),
-            ..Default::default()
-        })?;
-
         Ok(Self {
             issuer,
-            key_pair,
             scope,
-            jwt_header,
-            random: ring::rand::SystemRandom::new(),
+            private_key,
+            key_id,
         })
     }
 }
@@ -204,13 +268,21 @@ impl TokenProvider for SelfSignedJwt {
             exp,
         };
 
+        let jwt_header = b64_encode_obj(&JwtHeader {
+            alg: "RS256",
+            typ: Some("JWT"),
+            kid: Some(&self.key_id),
+            ..Default::default()
+        })?;
+
         let claim_str = b64_encode_obj(&claims)?;
-        let message = [self.jwt_header.as_ref(), claim_str.as_ref()].join(".");
-        let mut sig_bytes = vec![0; self.key_pair.public().modulus_len()];
-        self.key_pair
+        let message = [jwt_header.as_ref(), claim_str.as_ref()].join(".");
+        let mut sig_bytes = vec![0; self.private_key.0.public().modulus_len()];
+        self.private_key
+            .0
             .sign(
                 &ring::signature::RSA_PKCS1_SHA256,
-                &self.random,
+                &ring::rand::SystemRandom::new(),
                 message.as_bytes(),
                 &mut sig_bytes,
             )
@@ -238,7 +310,7 @@ where
 }
 
 /// A deserialized `service-account-********.json`-file.
-#[derive(serde::Deserialize, Debug)]
+#[derive(serde::Deserialize, Debug, Clone)]
 pub struct ServiceAccountCredentials {
     /// The private key in RSA format.
     pub private_key: String,
@@ -281,10 +353,19 @@ impl ServiceAccountCredentials {
         Ok(SelfSignedJwt::new(
             self.private_key_id,
             self.client_email,
-            self.private_key,
+            ServiceAccountKey::from_pem(self.private_key.as_bytes())?,
             DEFAULT_SCOPE.to_string(),
         )?)
     }
+
+    pub fn signing_credentials(self) -> crate::Result<GcpSigningCredentialProvider> {
+        Ok(Arc::new(StaticCredentialProvider::new(
+            GcpSigningCredential {
+                email: self.client_email,
+                private_key: Some(ServiceAccountKey::from_pem(self.private_key.as_bytes())?),
+            },
+        )))
+    }
 }
 
 /// Returns the number of seconds since unix epoch
@@ -295,21 +376,6 @@ fn seconds_since_epoch() -> u64 {
         .as_secs()
 }
 
-fn decode_first_rsa_key(private_key_pem: String) -> Result<RsaKeyPair> {
-    use rustls_pemfile::Item;
-    use std::io::Cursor;
-
-    let mut cursor = Cursor::new(private_key_pem);
-    let mut reader = BufReader::new(&mut cursor);
-
-    // Reading from string is infallible
-    match rustls_pemfile::read_one(&mut reader).unwrap() {
-        Some(Item::Pkcs8Key(key)) => Ok(RsaKeyPair::from_pkcs8(key.secret_pkcs8_der())?),
-        Some(Item::Pkcs1Key(key)) => Ok(RsaKeyPair::from_der(key.secret_pkcs1_der())?),
-        _ => Err(Error::MissingKey),
-    }
-}
-
 fn b64_encode_obj<T: serde::Serialize>(obj: &T) -> Result<String> {
     let string = serde_json::to_string(obj).context(EncodeSnafu)?;
     Ok(BASE64_URL_SAFE_NO_PAD.encode(string))
@@ -360,6 +426,7 @@ impl TokenProvider for InstanceCredentialProvider {
         let response = make_metadata_request(client, METADATA_HOST, retry)
             .or_else(|_| make_metadata_request(client, METADATA_IP, retry))
             .await?;
+
         let token = TemporaryToken {
             token: Arc::new(GcpCredential {
                 bearer: response.access_token,
@@ -370,12 +437,69 @@ impl TokenProvider for InstanceCredentialProvider {
     }
 }
 
+/// Make a request to the metadata server to fetch the client email, using a given hostname.
+async fn make_metadata_request_for_email(
+    client: &Client,
+    hostname: &str,
+    retry: &RetryConfig,
+) -> crate::Result<String> {
+    let url =
+        format!("http://{hostname}/computeMetadata/v1/instance/service-accounts/default/email",);
+    let response = client
+        .request(Method::GET, url)
+        .header("Metadata-Flavor", "Google")
+        .send_retry(retry)
+        .await
+        .context(TokenRequestSnafu)?
+        .text()
+        .await
+        .context(TokenResponseBodySnafu)?;
+    Ok(response)
+}
+
+/// A provider that uses the Google Cloud Platform metadata server to fetch a email for signing.
+///
+/// <https://cloud.google.com/appengine/docs/legacy/standard/java/accessing-instance-metadata>
+#[derive(Debug, Default)]
+pub struct InstanceSigningCredentialProvider {}
+
+#[async_trait]
+impl TokenProvider for InstanceSigningCredentialProvider {
+    type Credential = GcpSigningCredential;
+
+    /// Fetch a token from the metadata server.
+    /// Since the connection is local we need to enable http access and don't actually use the client object passed in.
+    async fn fetch_token(
+        &self,
+        client: &Client,
+        retry: &RetryConfig,
+    ) -> crate::Result<TemporaryToken<Arc<GcpSigningCredential>>> {
+        const METADATA_IP: &str = "169.254.169.254";
+        const METADATA_HOST: &str = "metadata";
+
+        info!("fetching token from metadata server");
+
+        let email = make_metadata_request_for_email(client, METADATA_HOST, retry)
+            .or_else(|_| make_metadata_request_for_email(client, METADATA_IP, retry))
+            .await?;
+
+        let token = TemporaryToken {
+            token: Arc::new(GcpSigningCredential {
+                email,
+                private_key: None,
+            }),
+            expiry: None,
+        };
+        Ok(token)
+    }
+}
+
 /// A deserialized `application_default_credentials.json`-file.
 ///
 /// # References
 /// - <https://cloud.google.com/docs/authentication/application-default-credentials#personal>
 /// - <https://google.aip.dev/auth/4110>
-#[derive(serde::Deserialize)]
+#[derive(serde::Deserialize, Clone)]
 #[serde(tag = "type")]
 pub enum ApplicationDefaultCredentials {
     /// Service Account.
@@ -423,13 +547,65 @@ impl ApplicationDefaultCredentials {
 const DEFAULT_TOKEN_GCP_URI: &str = "https://accounts.google.com/o/oauth2/token";
 
 /// <https://google.aip.dev/auth/4113>
-#[derive(Debug, Deserialize)]
+#[derive(Debug, Deserialize, Clone)]
 pub struct AuthorizedUserCredentials {
     client_id: String,
     client_secret: String,
     refresh_token: String,
 }
 
+#[derive(Debug, Deserialize)]
+pub struct AuthorizedUserSigningCredentials {
+    credential: AuthorizedUserCredentials,
+}
+
+///<https://oauth2.googleapis.com/tokeninfo?access_token=ACCESS_TOKEN>
+#[derive(Debug, Deserialize)]
+struct EmailResponse {
+    email: String,
+}
+
+impl AuthorizedUserSigningCredentials {
+    pub fn from(credential: AuthorizedUserCredentials) -> crate::Result<Self> {
+        Ok(Self { credential })
+    }
+
+    async fn client_email(&self, client: &Client, retry: &RetryConfig) -> crate::Result<String> {
+        let response = client
+            .request(Method::GET, "https://oauth2.googleapis.com/tokeninfo")
+            .query(&[("access_token", &self.credential.refresh_token)])
+            .send_retry(retry)
+            .await
+            .context(TokenRequestSnafu)?
+            .json::<EmailResponse>()
+            .await
+            .context(TokenResponseBodySnafu)?;
+
+        Ok(response.email)
+    }
+}
+
+#[async_trait]
+impl TokenProvider for AuthorizedUserSigningCredentials {
+    type Credential = GcpSigningCredential;
+
+    async fn fetch_token(
+        &self,
+        client: &Client,
+        retry: &RetryConfig,
+    ) -> crate::Result<TemporaryToken<Arc<GcpSigningCredential>>> {
+        let email = self.client_email(client, retry).await?;
+
+        Ok(TemporaryToken {
+            token: Arc::new(GcpSigningCredential {
+                email,
+                private_key: None,
+            }),
+            expiry: None,
+        })
+    }
+}
+
 #[async_trait]
 impl TokenProvider for AuthorizedUserCredentials {
     type Credential = GcpCredential;
@@ -462,3 +638,208 @@ impl TokenProvider for AuthorizedUserCredentials {
         })
     }
 }
+
+/// Trim whitespace from header values
+fn trim_header_value(value: &str) -> String {
+    let mut ret = value.to_string();
+    ret.retain(|c| !c.is_whitespace());
+    ret
+}
+
+/// A Google Cloud Storage Authorizer for generating signed URL using [Google SigV4]
+///
+/// [Google SigV4]: https://cloud.google.com/storage/docs/access-control/signed-urls
+#[derive(Debug)]
+pub struct GCSAuthorizer {
+    date: Option<DateTime<Utc>>,
+    credential: Arc<GcpSigningCredential>,
+}
+
+impl GCSAuthorizer {
+    /// Create a new [`GCSAuthorizer`]
+    pub fn new(credential: Arc<GcpSigningCredential>) -> Self {
+        Self {
+            date: None,
+            credential,
+        }
+    }
+
+    pub(crate) async fn sign(
+        &self,
+        method: Method,
+        url: &mut Url,
+        expires_in: Duration,
+        client: &GoogleCloudStorageClient,
+    ) -> crate::Result<()> {
+        let email = &self.credential.email;
+        let date = self.date.unwrap_or_else(Utc::now);
+        let scope = self.scope(date);
+        let credential_with_scope = format!("{}/{}", email, scope);
+
+        let mut headers = HeaderMap::new();
+        headers.insert("host", DEFAULT_GCS_SIGN_BLOB_HOST.parse().unwrap());
+
+        let (_, signed_headers) = Self::canonicalize_headers(&headers);
+
+        url.query_pairs_mut()
+            .append_pair("X-Goog-Algorithm", "GOOG4-RSA-SHA256")
+            .append_pair("X-Goog-Credential", &credential_with_scope)
+            .append_pair("X-Goog-Date", &date.format("%Y%m%dT%H%M%SZ").to_string())
+            .append_pair("X-Goog-Expires", &expires_in.as_secs().to_string())
+            .append_pair("X-Goog-SignedHeaders", &signed_headers);
+
+        let string_to_sign = self.string_to_sign(date, &method, url, &headers);
+        let signature = match &self.credential.private_key {
+            Some(key) => key.sign(&string_to_sign)?,
+            None => client.sign_blob(&string_to_sign, email).await?,
+        };
+
+        url.query_pairs_mut()
+            .append_pair("X-Goog-Signature", &signature);
+        Ok(())
+    }
+
+    /// Get scope for the request
+    ///
+    /// <https://cloud.google.com/storage/docs/authentication/signatures#credential-scope>
+    fn scope(&self, date: DateTime<Utc>) -> String {
+        format!("{}/auto/storage/goog4_request", date.format("%Y%m%d"),)
+    }
+
+    /// Canonicalizes query parameters into the GCP canonical form
+    /// form like:
+    ///```plaintext
+    ///HTTP_VERB  
+    ///PATH_TO_RESOURCE  
+    ///CANONICAL_QUERY_STRING  
+    ///CANONICAL_HEADERS  
+    ///
+    ///SIGNED_HEADERS  
+    ///PAYLOAD
+    ///```
+    ///
+    /// <https://cloud.google.com/storage/docs/authentication/canonical-requests>
+    fn canonicalize_request(url: &Url, methond: &Method, headers: &HeaderMap) -> String {
+        let verb = methond.as_str();
+        let path = url.path();
+        let query = Self::canonicalize_query(url);
+        let (canaonical_headers, signed_headers) = Self::canonicalize_headers(headers);
+
+        format!(
+            "{}\n{}\n{}\n{}\n\n{}\n{}",
+            verb, path, query, canaonical_headers, signed_headers, DEFAULT_GCS_PLAYLOAD_STRING
+        )
+    }
+
+    /// Canonicalizes query parameters into the GCP canonical form
+    /// form like `max-keys=2&prefix=object`
+    ///
+    /// <https://cloud.google.com/storage/docs/authentication/canonical-requests#about-query-strings>
+    fn canonicalize_query(url: &Url) -> String {
+        url.query_pairs()
+            .sorted_unstable_by(|a, b| a.0.cmp(&b.0))
+            .map(|(k, v)| {
+                format!(
+                    "{}={}",
+                    utf8_percent_encode(k.as_ref(), &STRICT_ENCODE_SET),
+                    utf8_percent_encode(v.as_ref(), &STRICT_ENCODE_SET)
+                )
+            })
+            .join("&")
+    }
+
+    /// Canonicalizes header into the GCP canonical form
+    ///
+    /// <https://cloud.google.com/storage/docs/authentication/canonical-requests#about-headers>
+    fn canonicalize_headers(header_map: &HeaderMap) -> (String, String) {
+        //FIXME add error handling for invalid header values
+        let mut headers = BTreeMap::<String, Vec<&str>>::new();
+        for (k, v) in header_map {
+            headers
+                .entry(k.as_str().to_lowercase())
+                .or_default()
+                .push(std::str::from_utf8(v.as_bytes()).unwrap());
+        }
+
+        let canonicalize_headers = headers
+            .iter()
+            .map(|(k, v)| {
+                format!(
+                    "{}:{}",
+                    k.trim(),
+                    v.iter().map(|v| trim_header_value(v)).join(",")
+                )
+            })
+            .join("\n");
+
+        let signed_headers = headers.keys().join(";");
+
+        (canonicalize_headers, signed_headers)
+    }
+
+    ///construct the string to sign
+    ///form like:
+    ///```plaintext
+    ///SIGNING_ALGORITHM  
+    ///ACTIVE_DATETIME  
+    ///CREDENTIAL_SCOPE  
+    ///HASHED_CANONICAL_REQUEST
+    ///```
+    ///`ACTIVE_DATETIME` format:`YYYYMMDD'T'HHMMSS'Z'`
+    /// <https://cloud.google.com/storage/docs/authentication/signatures#string-to-sign>
+    pub fn string_to_sign(
+        &self,
+        date: DateTime<Utc>,
+        request_method: &Method,
+        url: &Url,
+        headers: &HeaderMap,
+    ) -> String {
+        let caninical_request = Self::canonicalize_request(url, request_method, headers);
+        let hashed_canonical_req = hex_digest(caninical_request.as_bytes());
+        let scope = self.scope(date);
+
+        format!(
+            "{}\n{}\n{}\n{}",
+            "GOOG4-RSA-SHA256",
+            date.format("%Y%m%dT%H%M%SZ"),
+            scope,
+            hashed_canonical_req
+        )
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_canonicalize_headers() {
+        let mut input_header = HeaderMap::new();
+        input_header.insert("content-type", "text/plain".parse().unwrap());
+        input_header.insert("host", "storage.googleapis.com".parse().unwrap());
+        input_header.insert("x-goog-meta-reviewer", "jane".parse().unwrap());
+        input_header.append("x-goog-meta-reviewer", "john".parse().unwrap());
+        assert_eq!(
+            GCSAuthorizer::canonicalize_headers(&input_header),
+            (
+                "content-type:text/plain
+host:storage.googleapis.com
+x-goog-meta-reviewer:jane,john"
+                    .into(),
+                "content-type;host;x-goog-meta-reviewer".to_string()
+            )
+        );
+    }
+
+    #[test]
+    fn test_canonicalize_query() {
+        let mut url = Url::parse("https://storage.googleapis.com/bucket/object").unwrap();
+        url.query_pairs_mut()
+            .append_pair("max-keys", "2")
+            .append_pair("prefix", "object");
+        assert_eq!(
+            GCSAuthorizer::canonicalize_query(&url),
+            "max-keys=2&prefix=object".to_string()
+        );
+    }
+}
diff --git a/object_store/src/gcp/mod.rs b/object_store/src/gcp/mod.rs
index 2058d1f80..96afa45f2 100644
--- a/object_store/src/gcp/mod.rs
+++ b/object_store/src/gcp/mod.rs
@@ -35,8 +35,11 @@
 //!
 //! [lifecycle rule]: https://cloud.google.com/storage/docs/lifecycle#abort-mpu
 use std::sync::Arc;
+use std::time::Duration;
 
 use crate::client::CredentialProvider;
+use crate::gcp::credential::GCSAuthorizer;
+use crate::signer::Signer;
 use crate::{
     multipart::PartId, path::Path, GetOptions, GetResult, ListResult, MultipartId, MultipartUpload,
     ObjectMeta, ObjectStore, PutOptions, PutResult, Result, UploadPart,
@@ -45,13 +48,15 @@ use async_trait::async_trait;
 use bytes::Bytes;
 use client::GoogleCloudStorageClient;
 use futures::stream::BoxStream;
+use hyper::Method;
+use url::Url;
 
 use crate::client::get::GetClientExt;
 use crate::client::list::ListClientExt;
 use crate::client::parts::Parts;
 use crate::multipart::MultipartStore;
 pub use builder::{GoogleCloudStorageBuilder, GoogleConfigKey};
-pub use credential::GcpCredential;
+pub use credential::{GcpCredential, GcpSigningCredential, ServiceAccountKey};
 
 mod builder;
 mod client;
@@ -62,6 +67,10 @@ const STORE: &str = "GCS";
 /// [`CredentialProvider`] for [`GoogleCloudStorage`]
 pub type GcpCredentialProvider = Arc<dyn CredentialProvider<Credential = GcpCredential>>;
 
+/// [`GcpSigningCredential`] for [`GoogleCloudStorage`]
+pub type GcpSigningCredentialProvider =
+    Arc<dyn CredentialProvider<Credential = GcpSigningCredential>>;
+
 /// Interface for [Google Cloud Storage](https://cloud.google.com/storage/).
 #[derive(Debug)]
 pub struct GoogleCloudStorage {
@@ -83,6 +92,11 @@ impl GoogleCloudStorage {
     pub fn credentials(&self) -> &GcpCredentialProvider {
         &self.client.config().credentials
     }
+
+    /// Returns the [`GcpSigningCredentialProvider`] used by [`GoogleCloudStorage`]
+    pub fn signing_credentials(&self) -> &GcpSigningCredentialProvider {
+        &self.client.config().signing_credentials
+    }
 }
 
 #[derive(Debug)]
@@ -215,6 +229,34 @@ impl MultipartStore for GoogleCloudStorage {
     }
 }
 
+#[async_trait]
+impl Signer for GoogleCloudStorage {
+    async fn signed_url(&self, method: Method, path: &Path, expires_in: Duration) -> Result<Url> {
+        if expires_in.as_secs() > 604800 {
+            return Err(crate::Error::Generic {
+                store: STORE,
+                source: "Expiration Time can't be longer than 604800 seconds (7 days).".into(),
+            });
+        }
+
+        let config = self.client.config();
+        let path_url = config.path_url(path);
+        let mut url = Url::parse(&path_url).map_err(|e| crate::Error::Generic {
+            store: STORE,
+            source: format!("Unable to parse url {path_url}: {e}").into(),
+        })?;
+
+        let signing_credentials = self.signing_credentials().get_credential().await?;
+        let authorizer = GCSAuthorizer::new(signing_credentials);
+
+        authorizer
+            .sign(method, &mut url, expires_in, &self.client)
+            .await?;
+
+        Ok(url)
+    }
+}
+
 #[cfg(test)]
 mod test {
 
@@ -250,6 +292,36 @@ mod test {
         }
     }
 
+    #[tokio::test]
+    #[ignore]
+    async fn gcs_test_sign() {
+        crate::test_util::maybe_skip_integration!();
+        let integration = GoogleCloudStorageBuilder::from_env().build().unwrap();
+
+        let client = reqwest::Client::new();
+
+        let path = Path::from("test_sign");
+        let url = integration
+            .signed_url(Method::PUT, &path, Duration::from_secs(3600))
+            .await
+            .unwrap();
+        println!("PUT {url}");
+
+        let resp = client.put(url).body("data").send().await.unwrap();
+        resp.error_for_status().unwrap();
+
+        let url = integration
+            .signed_url(Method::GET, &path, Duration::from_secs(3600))
+            .await
+            .unwrap();
+        println!("GET {url}");
+
+        let resp = client.get(url).send().await.unwrap();
+        let resp = resp.error_for_status().unwrap();
+        let data = resp.bytes().await.unwrap();
+        assert_eq!(data.as_ref(), b"data");
+    }
+
     #[tokio::test]
     async fn gcs_test_get_nonexistent_location() {
         crate::test_util::maybe_skip_integration!();
diff --git a/object_store/src/util.rs b/object_store/src/util.rs
index a19d5aab4..161d2d138 100644
--- a/object_store/src/util.rs
+++ b/object_store/src/util.rs
@@ -285,6 +285,35 @@ impl<T: RangeBounds<usize>> From<T> for GetRange {
         }
     }
 }
+// http://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html
+//
+// Do not URI-encode any of the unreserved characters that RFC 3986 defines:
+// A-Z, a-z, 0-9, hyphen ( - ), underscore ( _ ), period ( . ), and tilde ( ~ ).
+#[cfg(any(feature = "aws", feature = "gcp"))]
+pub(crate) const STRICT_ENCODE_SET: percent_encoding::AsciiSet = percent_encoding::NON_ALPHANUMERIC
+    .remove(b'-')
+    .remove(b'.')
+    .remove(b'_')
+    .remove(b'~');
+
+/// Computes the SHA256 digest of `body` returned as a hex encoded string
+#[cfg(any(feature = "aws", feature = "gcp"))]
+pub(crate) fn hex_digest(bytes: &[u8]) -> String {
+    let digest = ring::digest::digest(&ring::digest::SHA256, bytes);
+    hex_encode(digest.as_ref())
+}
+
+/// Returns `bytes` as a lower-case hex encoded string
+#[cfg(any(feature = "aws", feature = "gcp"))]
+pub(crate) fn hex_encode(bytes: &[u8]) -> String {
+    use std::fmt::Write;
+    let mut out = String::with_capacity(bytes.len() * 2);
+    for byte in bytes {
+        // String writing is infallible
+        let _ = write!(out, "{byte:02x}");
+    }
+    out
+}
 
 #[cfg(test)]
 mod tests {
diff --git a/parquet/Cargo.toml b/parquet/Cargo.toml
index 04dfed408..0304c7212 100644
--- a/parquet/Cargo.toml
+++ b/parquet/Cargo.toml
@@ -49,7 +49,7 @@ object_store = { version = "0.9.0", default-features = false, optional = true }
 bytes = { version = "1.1", default-features = false, features = ["std"] }
 thrift = { version = "0.17", default-features = false }
 snap = { version = "1.0", default-features = false, optional = true }
-brotli = { version = "3.3", default-features = false, features = ["std"], optional = true }
+brotli = { version = "4.0", default-features = false, features = ["std"], optional = true }
 flate2 = { version = "1.0", default-features = false, features = ["rust_backend"], optional = true }
 lz4_flex = { version = "0.11", default-features = false, features = ["std", "frame"], optional = true }
 zstd = { version = "0.13.0", optional = true, default-features = false }
@@ -76,7 +76,7 @@ base64 = { version = "0.22", default-features = false, features = ["std"] }
 criterion = { version = "0.5", default-features = false }
 snap = { version = "1.0", default-features = false }
 tempfile = { version = "3.0", default-features = false }
-brotli = { version = "3.3", default-features = false, features = ["std"] }
+brotli = { version = "4.0", default-features = false, features = ["std"] }
 flate2 = { version = "1.0", default-features = false, features = ["rust_backend"] }
 lz4_flex = { version = "0.11", default-features = false, features = ["std", "frame"] }
 zstd = { version = "0.13", default-features = false }
diff --git a/parquet/src/arrow/arrow_reader/mod.rs b/parquet/src/arrow/arrow_reader/mod.rs
index 83d6f6f55..3b4c09314 100644
--- a/parquet/src/arrow/arrow_reader/mod.rs
+++ b/parquet/src/arrow/arrow_reader/mod.rs
@@ -270,10 +270,7 @@ impl ArrowReaderMetadata {
         Self::try_new(Arc::new(metadata), options)
     }
 
-    pub(crate) fn try_new(
-        metadata: Arc<ParquetMetaData>,
-        options: ArrowReaderOptions,
-    ) -> Result<Self> {
+    pub fn try_new(metadata: Arc<ParquetMetaData>, options: ArrowReaderOptions) -> Result<Self> {
         let kv_metadata = match options.skip_arrow_metadata {
             true => None,
             false => metadata.file_metadata().key_value_metadata(),


Test patch:
diff --git a/arrow-pyarrow-integration-testing/Cargo.toml b/arrow-pyarrow-integration-testing/Cargo.toml
index 8c60c086c..6f07d42d8 100644
--- a/arrow-pyarrow-integration-testing/Cargo.toml
+++ b/arrow-pyarrow-integration-testing/Cargo.toml
@@ -34,4 +34,4 @@ crate-type = ["cdylib"]
 
 [dependencies]
 arrow = { path = "../arrow", features = ["pyarrow"] }
-pyo3 = { version = "0.20", features = ["extension-module"] }
+pyo3 = { version = "0.21.1", features = ["extension-module"] }
diff --git a/arrow-pyarrow-integration-testing/src/lib.rs b/arrow-pyarrow-integration-testing/src/lib.rs
index a53447b53..918fa74e3 100644
--- a/arrow-pyarrow-integration-testing/src/lib.rs
+++ b/arrow-pyarrow-integration-testing/src/lib.rs
@@ -40,9 +40,9 @@ fn to_py_err(err: ArrowError) -> PyErr {
 
 /// Returns `array + array` of an int64 array.
 #[pyfunction]
-fn double(array: &PyAny, py: Python) -> PyResult<PyObject> {
+fn double(array: &Bound<PyAny>, py: Python) -> PyResult<PyObject> {
     // import
-    let array = make_array(ArrayData::from_pyarrow(array)?);
+    let array = make_array(ArrayData::from_pyarrow_bound(&array)?);
 
     // perform some operation
     let array = array
@@ -60,7 +60,7 @@ fn double(array: &PyAny, py: Python) -> PyResult<PyObject> {
 /// calls a lambda function that receives and returns an array
 /// whose result must be the array multiplied by two
 #[pyfunction]
-fn double_py(lambda: &PyAny, py: Python) -> PyResult<bool> {
+fn double_py(lambda: &Bound<PyAny>, py: Python) -> PyResult<bool> {
     // create
     let array = Arc::new(Int64Array::from(vec![Some(1), None, Some(3)]));
     let expected = Arc::new(Int64Array::from(vec![Some(2), None, Some(6)])) as ArrayRef;
@@ -68,7 +68,7 @@ fn double_py(lambda: &PyAny, py: Python) -> PyResult<bool> {
     // to py
     let pyarray = array.to_data().to_pyarrow(py)?;
     let pyarray = lambda.call1((pyarray,))?;
-    let array = make_array(ArrayData::from_pyarrow(pyarray)?);
+    let array = make_array(ArrayData::from_pyarrow_bound(&pyarray)?);
 
     Ok(array == expected)
 }
@@ -82,16 +82,12 @@ fn make_empty_array(datatype: PyArrowType<DataType>, py: Python) -> PyResult<PyO
 
 /// Returns the substring
 #[pyfunction]
-fn substring(
-    array: PyArrowType<ArrayData>,
-    start: i64,
-) -> PyResult<PyArrowType<ArrayData>> {
+fn substring(array: PyArrowType<ArrayData>, start: i64) -> PyResult<PyArrowType<ArrayData>> {
     // import
     let array = make_array(array.0);
 
     // substring
-    let array =
-        kernels::substring::substring(array.as_ref(), start, None).map_err(to_py_err)?;
+    let array = kernels::substring::substring(array.as_ref(), start, None).map_err(to_py_err)?;
 
     Ok(array.to_data().into())
 }
@@ -102,8 +98,7 @@ fn concatenate(array: PyArrowType<ArrayData>, py: Python) -> PyResult<PyObject>
     let array = make_array(array.0);
 
     // concat
-    let array =
-        kernels::concat::concat(&[array.as_ref(), array.as_ref()]).map_err(to_py_err)?;
+    let array = kernels::concat::concat(&[array.as_ref(), array.as_ref()]).map_err(to_py_err)?;
 
     array.to_data().to_pyarrow(py)
 }
@@ -129,9 +124,7 @@ fn round_trip_array(obj: PyArrowType<ArrayData>) -> PyResult<PyArrowType<ArrayDa
 }
 
 #[pyfunction]
-fn round_trip_record_batch(
-    obj: PyArrowType<RecordBatch>,
-) -> PyResult<PyArrowType<RecordBatch>> {
+fn round_trip_record_batch(obj: PyArrowType<RecordBatch>) -> PyResult<PyArrowType<RecordBatch>> {
     Ok(obj)
 }
 
@@ -168,7 +161,7 @@ fn boxed_reader_roundtrip(
 }
 
 #[pymodule]
-fn arrow_pyarrow_integration_testing(_py: Python, m: &PyModule) -> PyResult<()> {
+fn arrow_pyarrow_integration_testing(_py: Python, m: &Bound<PyModule>) -> PyResult<()> {
     m.add_wrapped(wrap_pyfunction!(double))?;
     m.add_wrapped(wrap_pyfunction!(double_py))?;
     m.add_wrapped(wrap_pyfunction!(make_empty_array))?;
diff --git a/arrow/tests/pyarrow.rs b/arrow/tests/pyarrow.rs
index 4b6991da0..a1c365c31 100644
--- a/arrow/tests/pyarrow.rs
+++ b/arrow/tests/pyarrow.rs
@@ -32,9 +32,9 @@ fn test_to_pyarrow() {
 
     let res = Python::with_gil(|py| {
         let py_input = input.to_pyarrow(py)?;
-        let records = RecordBatch::from_pyarrow(py_input.as_ref(py))?;
+        let records = RecordBatch::from_pyarrow_bound(py_input.bind(py))?;
         let py_records = records.to_pyarrow(py)?;
-        RecordBatch::from_pyarrow(py_records.as_ref(py))
+        RecordBatch::from_pyarrow_bound(py_records.bind(py))
     })
     .unwrap();